import re
import subprocess
import sys
import unittest
from pathlib import Path

ROOT = Path(__file__).resolve().parents[3]
SCRIPT = ROOT / "harness/agents/impl/skills/gitea-issue-rework/scripts/evidence_manifest.py"
PREPARE_SCRIPT = ROOT / "harness/lib/verification_root.py"
CACHE = ROOT / "harness/operations/repository-cache.md"
SKILLS = {
    "implementation": ROOT / "harness/agents/impl/skills/gitea-issue-implementation/SKILL.md",
    "rework": ROOT / "harness/agents/impl/skills/gitea-issue-rework/SKILL.md",
    "review": ROOT / "harness/agents/spec/skills/gitea-implementation-spec-review/SKILL.md",
}
CANONICAL = "<harness-root>/repositories/<owner>/<repo>/worktrees/issue-<number>/verification/<run-id>"


class VerificationContractDocumentationTests(unittest.TestCase):
    def test_runtime_policy_is_broker_owned_and_separate_from_verification(self):
        text = CACHE.read_text(encoding="utf-8")
        self.assertIn("runtime/<workflow-run-id>", text)
        self.assertIn("broker.handle", text)
        self.assertIn("broker.sock", text)
        self.assertIn("ready JSON", text)
        self.assertIn("runtime 생성은 승인이나 admission을 의미하지 않는다", text)
        self.assertIn("active resource", text)
        self.assertIn("unknown", text)
        self.assertIn("성공 기록을 실패로 덮어쓰지 않는다", text)
        for path in SKILLS.values():
            with self.subTest(path=path.name):
                role = path.read_text(encoding="utf-8")
                gate = role.split("## Managed-lock broker gate", 1)[1].split("\n## ", 1)[0]
                self.assertIn("broker-generated", gate)
                self.assertIn("ready JSON", gate)
                self.assertNotIn("pre-created", gate)
                self.assertNotIn("pre-create", gate)

    def test_role_runtime_startup_precedes_and_does_not_replace_admission(self):
        roles = [path for path in (ROOT / 'harness/agents').rglob('SKILL.md')
                 if 'broker-generated runtime' in path.read_text()]
        self.assertEqual(len(roles), 6)
        for path in roles:
            with self.subTest(role=path.parent.name):
                gate = path.read_text().split('## Managed-lock broker gate', 1)[1].split('\n## ', 1)[0]
                self.assertIn('ready 이후 각 workflow의 진입 검증·승인 조건을 확인한다', gate)
                self.assertIn('Runtime 생성은 approval이나 worktree admission을 대신하지 않는다', gate)
                self.assertIn('ready JSON의 `handle`/`socket`', gate)

    def test_evidence_plan_disables_python_bytecode_in_clean_checkout(self):
        import json
        template = ROOT / "harness/agents/impl/skills/gitea-issue-rework/templates/evidence-plan.json"
        plan = json.loads(template.read_text())
        self.assertIn("-B", plan["commands"][0]["argv"])
        self.assertIn("PYTHONDONTWRITEBYTECODE", plan["command_env_allowlist"])

    def test_full_suite_tracked_background_reconciles_before_retry(self):
        text = SKILLS['rework'].read_text()
        heading = '### Tracked bounded background evidence execution'
        self.assertIn(heading, text)
        section = text.split(heading, 1)[1].split('\n### ', 1)[0]
        for rule in ['foreground tool 시간창', 'tracked background',
                     'process/session handle', 'run-id', 'source SHA', 'plan digest',
                     'output path', 'timeout_seconds', '유한한 outer deadline',
                     'manifest', 'validate', 'cleanup', 'pending/partial',
                     '중복 실행하지 않는다', '새 run-id', '별도 승인',
                     'TDD RED', '세션 종료 이후의 지속성을 보장하지 않는다']:
            with self.subTest(rule=rule):
                self.assertIn(rule, section)
        self.assertIn('자동 retry하지 않는다', section)
        self.assertIn('기존 artifact를 덮어쓰지 않는다', section)

    def test_command_deadline_and_archive_policy_is_explicit(self):
        import json
        template = ROOT / "harness/agents/impl/skills/gitea-issue-rework/templates/evidence-plan.json"
        for command in json.loads(template.read_text())["commands"]:
            self.assertIsInstance(command["timeout_seconds"], (int, float))
            self.assertGreater(command["timeout_seconds"], 0)
            self.assertLessEqual(command["timeout_seconds"], 3600)
        for path in (CACHE, *SKILLS.values()):
            text = path.read_text()
            for token in ("evidence-manifest/v3", "timeout_seconds", "process group",
                          "validate-archived-v2", "TDD RED", "자동 재실행"):
                with self.subTest(path=str(path), token=token):
                    self.assertIn(token, text)
        help_result = subprocess.run([sys.executable, str(SCRIPT), "validate-archived-v2", "--help"],
                                     capture_output=True, text=True, timeout=10)
        self.assertEqual(help_result.returncode, 0, help_result.stderr)
        for option in ("--contract-path", "--repo-root", "--artifact-root", "--manifest"):
            self.assertIn(option, help_result.stdout)

    def test_all_roles_reference_canonical_external_artifacts(self):
        for path in (CACHE, *SKILLS.values()):
            with self.subTest(path=path.name):
                text = path.read_text(encoding="utf-8")
                self.assertIn(CANONICAL, text)
                self.assertIn("evidence_gap", text)
                self.assertIn("[REDACTED]", text)
                self.assertIn("evidence-manifest/v3", text)

    def test_examples_match_real_helper_cli_not_unsupported_options(self):
        for path in (SKILLS["implementation"], SKILLS["rework"]):
            section = path.read_text().split("## Canonical external verification evidence", 1)[1]
            match = re.search(r"```sh\n(.*?)```", section, re.S)
            assert match is not None
            block = match.group(1).replace("\\\n", " ")
            commands = [line for line in block.splitlines() if line.startswith("python3 ")]
            self.assertEqual(len(commands), 3)
            for op, command in zip(("prepare", "run", "validate"), commands):
                with self.subTest(skill=path.parent.name, op=op):
                    declared = set(re.findall(r"--[a-z-]+", command))
                    if op == "prepare":
                        self.assertIn("harness/lib/verification_root.py", command)
                        help_result = subprocess.run(
                            [sys.executable, str(PREPARE_SCRIPT), op, "--help"],
                            text=True,
                            capture_output=True,
                        )
                        self.assertEqual(help_result.returncode, 0, help_result.stderr)
                        supported = set(re.findall(r"--[a-z-]+", help_result.stdout))
                        required = {"--root", "--owner", "--repository", "--issue", "--run-id", "--handle"}
                        self.assertEqual(declared, required)
                        self.assertTrue(declared <= supported)
                    else:
                        help_result = subprocess.run([sys.executable, str(SCRIPT), op, "--help"], text=True, capture_output=True)
                        self.assertEqual(help_result.returncode, 0)
                        supported = set(re.findall(r"--[a-z-]+", help_result.stdout))
                        required = {"--contract-path", "--repo-root", "--artifact-root"} | ({"--plan", "--output"} if op == "run" else {"--manifest"})
                        self.assertEqual(declared, required)
                        self.assertTrue(declared <= supported)
                    self.assertIn("exact current HEAD", section)
                    self.assertIn("validate-legacy", section)
                    self.assertIn("exclusion 없음", section)

    def test_prepare_contract_binds_root_and_forbids_recursive_parent_creation(self):
        for path in (CACHE, SKILLS["implementation"], SKILLS["rework"]):
            with self.subTest(path=path.name):
                text = path.read_text(encoding="utf-8")
                self.assertIn("exact harness root", text)
                self.assertIn("Path.mkdir(parents=True, mode=0o700)", text)
                self.assertIn("자동 chmod", text)

    def test_checkout_cleanup_preserves_verification_without_recursive_delete(self):
        text = CACHE.read_text()
        cleanup = text.split("## Worktree 실패 처리와 제거", 1)[1].split("## Credential 안전", 1)[0]
        self.assertIn("`verification/`이 남으면 Issue directory를 보존한다", cleanup)
        self.assertIn("Issue directory가 남는 것은 checkout 제거 실패가 아니다", cleanup)
        self.assertIn("digest가 제거 전과 같음", cleanup)
        self.assertIn("Artifact 폐기는 별도의 exact 경로 승인", text)
        self.assertIn("mode `0700`", text)
        self.assertIn("mode `0600`", text)

    def test_spec_review_uses_immutable_source_and_not_moving_checkout(self):
        section = SKILLS["review"].read_text().split("## External Issue verification evidence boundary", 1)[1]
        self.assertIn("제품 테스트를 실행하지 않", section)
        self.assertIn("moving implementation checkout", section)
        self.assertIn("authoritative blob evidence", section)
        self.assertIn("trusted `harness/.runtime`", section)
        self.assertIn("Producer evidence와 reviewer 실행을 구분", section)

    def test_workflow_guard_records_reuse_existing_verification_location(self):
        text = CACHE.read_text()
        guard = text.split('### Shared workflow guard gates', 1)[1].split('Issue별 검증 실행 기록의 정본', 1)[0]
        self.assertNotIn('or verification root', guard)
        for path in (SKILLS['implementation'], SKILLS['rework']):
            self.assertNotIn('outside product checkout and verification root', path.read_text())
        self.assertIn('verification/<run-id>', guard)

    def test_workflow_guard_readonly_scope_does_not_ban_approved_worker_edits(self):
        text = CACHE.read_text()
        guard = text.split('### Shared workflow guard gates', 1)[1].split('Issue별 검증 실행 기록의 정본', 1)[0]
        self.assertIn('읽기 전용', guard)
        self.assertIn('쓰기 worker는 승인된 변경 allowlist', guard)

    def test_workflow_guard_snapshot_protects_scope_before_delegation(self):
        implementation = SKILLS["implementation"].read_text()
        self.assertIn("workflow_guard.py snapshot", implementation)
        self.assertIn("compare-source --repo-root", implementation)
        self.assertIn("immutable protected scope", implementation)

    def test_workflow_guard_inventory_is_checked_before_adopting_results(self):
        rework = SKILLS["rework"].read_text()
        self.assertIn("workflow_guard.py check-tests", rework)
        self.assertIn("test inventory", rework)
        self.assertIn("RED/GREEN records must be named", rework)

    def test_workflow_guard_contract_preservation_precedes_spec_verdict(self):
        spec = SKILLS["review"].read_text()
        self.assertIn("workflow_guard.py check-contracts", spec)
        self.assertIn("baseline inventory", spec)
        self.assertIn("Gate failure invalidates workflow evidence", spec)

    def test_workflow_guard_nested_schema_is_actionable_and_claims_are_not_authenticated(self):
        text = CACHE.read_text()
        section = text.split('### Shared workflow guard gates', 1)[1].split('Issue별 검증 실행 기록의 정본', 1)[0]
        for token in ('#### Nested input contracts', 'baseline_collection', 'inventory_sha256',
                      'baseline_status', 'expected_failure', 'replacement_ids', 'kind',
                      'delete', 'rename', 'status', 'review_evidence', 'scope_sha256',
                      'expected_exit_code', 'failure_kind', 'semantic_key', 'current_keys',
                      'observation', 'intentionally_changed', 'sort_keys=True',
                      'separators', 'UTF-8', 'shape-only', 'one row per baseline semantic key'):
            with self.subTest(token=token):
                self.assertIn(token, section)
        import json
        blocks = re.findall(r'```json\n(.*?)```', section, re.S)
        self.assertEqual(len(blocks), 2, 'provide runnable test and contract input examples')
        self.assertEqual(set(json.loads(blocks[0])), {'baseline_ids', 'current_ids', 'baseline_status',
            'current_status', 'baseline_collection', 'current_collection', 'authorized_changes', 'changed_tests', 'findings'})
        self.assertEqual(set(json.loads(blocks[1])), {'baseline', 'current', 'mappings', 'stages', 'precedence'})

    def test_mapping_schema_types_use_raw_source_not_redacted_tool_rendering(self):
        # Authorization-looking prose can be masked by read-tool transport.
        # Inspect the approved public schema in raw source; never disable redaction.
        declaration = 'authorization: string, verification: string'
        masked = 'authorization: ' + '*' * 3 + ' verification'
        helper = ROOT / 'harness/lib/workflow_guard.py'
        for path in (helper, CACHE):
            text = path.read_text()
            with self.subTest(path=path.name):
                self.assertEqual(text.count(declaration), 1)
                self.assertNotIn(masked, text)

    def test_rework_no_longer_excludes_repo_evidence_from_clean_tree(self):
        text = SKILLS["rework"].read_text()
        self.assertNotIn("plan과 manifest를 제외한 clean tree", text)
        self.assertIn("새 evidence는 checkout 밖", text)
        self.assertIn("archived v1 read-only", text)


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