from __future__ import annotations

import json
import re
from pathlib import Path

from investibook.core.system_prompt import build_static_protocol


PROJECT_ROOT = Path(__file__).resolve().parents[2]
AGENT_CONFIG_DIR = PROJECT_ROOT / "agentkit" / "config"
SKILLS_DIR = PROJECT_ROOT / "agentkit" / "skills"


def _catalog_rows() -> list[tuple[str, str, str]]:
    text = (AGENT_CONFIG_DIR / "SKILL_CATALOG.md").read_text(encoding="utf-8")
    rows: list[tuple[str, str, str]] = []
    for line in text.splitlines():
        match = re.match(r"^\| ([a-z-]+) \| `([^`]+)` \| `([^`]+)` \|$", line)
        if match:
            rows.append((match.group(1), match.group(2), match.group(3)))
    return rows


def test_opencode_template_registers_three_diagnosis_agents() -> None:
    config = json.loads(
        (AGENT_CONFIG_DIR / "opencode.json.template").read_text(encoding="utf-8")
    )

    assert config["default_agent"] == "investigator"
    agents = config["agent"]
    assert set(agents) == {
        "investigator",
        "evidence-agent",
        "diagnosis-analyst",
        "code-investigator",
        "build",
        "plan",
    }
    # OpenCode's built-in primaries are demoted to hidden subagents so the
    # investigator is the only visible primary — the ACP default-agent
    # fallback can then never land on "build" (same trick omo uses).
    assert agents["build"] == {"mode": "subagent", "hidden": True}
    assert agents["plan"] == {"mode": "subagent", "hidden": True}
    assert agents["investigator"]["mode"] == "primary"
    assert agents["evidence-agent"]["mode"] == "subagent"
    assert agents["diagnosis-analyst"]["mode"] == "subagent"
    # Agent descriptions are the Task-tool routing signal: they must state the
    # invocation condition so the investigator does not dispatch collection every time.
    assert "invoke only when" in agents["evidence-agent"]["description"]
    assert "does not collect data" in agents["diagnosis-analyst"]["description"]
    assert agents["investigator"]["permission"]["task"] == {
        "*": "deny",
        "evidence-agent": "allow",
        "diagnosis-analyst": "allow",
        "code-investigator": "ask",
    }
    assert agents["evidence-agent"]["permission"]["task"] == "deny"
    assert agents["diagnosis-analyst"]["permission"]["task"] == "deny"
    code_bash = agents["code-investigator"]["permission"]["bash"]
    assert code_bash["*"] == "deny"
    assert any("code_context.py prepare" in key for key in code_bash)
    assert any("code_context.py inspect " in key for key in code_bash)
    assert any("code_context.py inspect-refine" in key for key in code_bash)
    assert any("code_context.py history-log" in key for key in code_bash)
    assert any("code_context.py history-blame" in key for key in code_bash)
    assert any("code_context.py history-show" in key for key in code_bash)
    assert any("code_context.py history-diff" in key for key in code_bash)
    assert not any("code_context.py search" in key for key in code_bash)
    assert not any("code_context.py read-many" in key for key in code_bash)
    assert agents["code-investigator"]["permission"]["submit_code_artifact"] == "allow"
    # Orchestrator runs with build-parity permissions (omo-style: open tools,
    # prompt-shaped behavior); only Task dispatch stays scoped to the two
    # diagnosis subagents, which encodes the architecture.
    investigator_permission = agents["investigator"]["permission"]
    assert set(investigator_permission) == {"task"}
    # Analyst bash is scoped to the bounded raw-evidence reader and JSON
    # validation only; it must not be able to run the collector or clinic-cli.
    analyst_bash = agents["diagnosis-analyst"]["permission"]["bash"]
    assert analyst_bash["*"] == "deny"
    assert any(
        "read_raw_evidence.py" in k and v == "allow" for k, v in analyst_bash.items()
    )
    assert any(
        "write_analyst_report.py" in k and v == "allow" for k, v in analyst_bash.items()
    )
    assert any("json.tool" in k and v == "allow" for k, v in analyst_bash.items())
    assert not any("collect_evidence" in k for k in analyst_bash)
    assert not any("clinic-cli" in k for k in analyst_bash)
    assert agents["evidence-agent"]["permission"]["skill"] == {
        "*": "deny",
        "time-conversion": "allow",
        "tidb-knowledge-base": "allow",
        "diagnosis-overview": "allow",
        "clinic-cli": "allow",
        "alertstore-cli": "allow",
        "jira": "allow",
        "devops": "allow",
        "o11y-data": "allow",
        "o11y-metrics-api": "allow",
        "github-issue-search": "allow",
    }
    assert agents["diagnosis-analyst"]["permission"]["skill"] == {
        "*": "deny",
        "time-conversion": "allow",
        "tidb-knowledge-base": "allow",
        "diagnosis-overview": "allow",
        "tidb-clinic-diagnosis": "allow",
        "tidb-query-tuning-autox": "allow",
        "transaction-oncall-handbook": "allow",
        "ticdc-health-inspection": "allow",
        "tikv-cdc": "allow",
    }


def test_code_investigator_refinement_is_bounded_and_must_make_progress() -> None:
    prompt = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "code-investigator.md"
    ).read_text(encoding="utf-8")
    skill = (SKILLS_DIR / "builtin" / "tidb-code-context" / "SKILL.md").read_text(
        encoding="utf-8"
    )

    for text in (prompt, skill):
        normalized = " ".join(text.lower().split())
        assert "initial `inspect` returns `true`" in normalized
        assert "second and final" in normalized
        assert "at least one semantic anchor not used" in normalized
        assert "path hint not used" in normalized
        assert "never repeat the same anchors and path hints" in normalized
        assert "after the second refinement, submit the artifact" in normalized
        assert "when `refinement_available` is `false`" in normalized


def test_catalog_references_existing_skills() -> None:
    rows = _catalog_rows()
    assert rows

    missing = [
        (group, skill, path)
        for group, skill, path in rows
        if not (SKILLS_DIR / path / "SKILL.md").exists()
    ]
    assert missing == []


def test_catalog_default_groups_do_not_cross_load_data_and_diagnosis_skills() -> None:
    rows = _catalog_rows()
    groups_by_skill = {skill: group for group, skill, _path in rows}

    evidence_default = {
        "time-conversion",
        "tidb-knowledge-base",
        "diagnosis-overview",
        "clinic-cli",
        "alertstore-cli",
        "jira",
        "devops",
        "o11y-data",
        "o11y-metrics-api",
        "github-issue-search",
    }
    analyst_default = {
        "time-conversion",
        "tidb-knowledge-base",
        "diagnosis-overview",
    }

    assert {groups_by_skill[skill] for skill in evidence_default} <= {
        "core",
        "data-access",
    }
    assert "diagnosis" not in {groups_by_skill[skill] for skill in evidence_default}
    assert {groups_by_skill[skill] for skill in analyst_default} <= {
        "core",
        "diagnosis",
    }
    assert "data-access" not in {groups_by_skill[skill] for skill in analyst_default}
    assert "tidb-query-tuning" not in groups_by_skill
    assert "tikv-fast-tune" not in groups_by_skill
    assert "autoscaler-dashboard" not in groups_by_skill


def test_transaction_family_routes_to_cataloged_nutshell_specialist() -> None:
    rows = _catalog_rows()
    catalog = {skill: (group, path) for group, skill, path in rows}
    expected_path = "nutshell-skills/diagnosis/transaction-oncall-handbook"

    assert catalog["transaction-oncall-handbook"] == ("diagnosis", expected_path)

    family_text = (
        SKILLS_DIR
        / "builtin"
        / "tidb-clinic-diagnosis"
        / "transaction-or-lock-contention"
        / "SUBSKILL.md"
    ).read_text(encoding="utf-8")
    analyst_text = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "diagnosis-analyst.md"
    ).read_text(encoding="utf-8")

    assert (
        "When `transaction_or_lock_contention` is the selected primary class"
        in family_text
    )
    assert f".opencode/.all-skills/{expected_path}/SKILL.md" in family_text
    assert "at most one reference" in family_text
    assert "selected_diagnosis_skills" in family_text
    assert "classification family subskills own specialist routing" in analyst_text


def test_schema_files_are_valid_json() -> None:
    for path in (AGENT_CONFIG_DIR / "investibook" / "schemas").glob("*.json"):
        schema = json.loads(path.read_text(encoding="utf-8"))
        assert schema["type"] == "object"
        assert schema["required"]


def test_copilot_routing_checkpoint_is_bounded_and_backward_compatible() -> None:
    investigator = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "investigator.md"
    ).read_text(encoding="utf-8")
    routing_schema = json.loads(
        (
            AGENT_CONFIG_DIR
            / "investibook"
            / "schemas"
            / "diagnosis-routing.schema.json"
        ).read_text(encoding="utf-8")
    )
    analyst_schema = json.loads(
        (
            AGENT_CONFIG_DIR
            / "investibook"
            / "schemas"
            / "diagnosis-analyst-output.schema.json"
        ).read_text(encoding="utf-8")
    )

    assert "When the block is absent, use `diagnosis_mode=autopilot`" in investigator
    assert "API transport does not imply automation" in investigator
    assert "Route Selection Checkpoint (`copilot` only)" in investigator
    assert "Time Range Confirmation Checkpoint (`copilot` only)" in investigator
    assert "do not collect evidence before confirmation" in investigator
    assert "load the builtin `time-conversion` skill" in investigator
    assert "Pending route selection" in investigator
    assert "before the other intent rows" in investigator
    assert "routing_confidence" in investigator
    assert "not root-cause probability" in investigator
    assert "If `supports_followup=false`, fall back to `autopilot`" in investigator
    assert "Do not give a root-cause conclusion in this response" in investigator
    assert "never bend the conclusion to the user's route choice" in investigator
    assert "write_diagnosis_routing.py --from-stdin" in investigator
    assert "newly invented IDs" in investigator
    assert "are not\nroutes" in investigator
    assert routing_schema["properties"]["candidates"]["maxItems"] == 3
    assert routing_schema["properties"]["selection"]["oneOf"][1]["properties"]["routes"]["maxItems"] == 3
    assert routing_schema["properties"]["status"]["enum"] == [
        "awaiting_user",
        "selected",
    ]
    standard_routes = routing_schema["$defs"]["standardRoute"]["enum"]
    assert len(standard_routes) == 11
    assert "sql_plan_or_query" in standard_routes
    assert "transaction_or_lock_contention" in standard_routes
    assert analyst_schema["$defs"]["standardRoute"]["enum"] == standard_routes
    assert routing_schema["properties"]["candidates"]["items"]["properties"]["route"] == {
        "$ref": "#/$defs/standardRoute"
    }


def test_analyst_contract_enforces_reroute_and_canonical_window() -> None:
    analyst = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "diagnosis-analyst.md"
    ).read_text(encoding="utf-8")

    assert "A Copilot route is a hypothesis, not authority" in analyst
    assert "classification_override" in analyst
    assert "read the final family" in analyst
    assert "outside_window_context" in analyst
    assert "cannot establish the root cause" in analyst


def test_time_confirmation_schema_precedes_collection() -> None:
    investigator = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "investigator.md"
    ).read_text(encoding="utf-8")
    schema = json.loads(
        (
            AGENT_CONFIG_DIR
            / "investibook"
            / "schemas"
            / "diagnosis-time-confirmation.schema.json"
        ).read_text(encoding="utf-8")
    )

    assert schema["properties"]["status"]["enum"] == ["awaiting_user", "confirmed"]
    assert "Do not read diagnosis family skills" in investigator
    assert "A new or changed window requires a new checkpoint" in investigator
    assert '`channel!="lark"`' in investigator
    assert '`channel="lark"` enters directly' in investigator


def test_selected_route_handoff_enforces_family_loading_contract() -> None:
    catalog = (AGENT_CONFIG_DIR / "SKILL_CATALOG.md").read_text(encoding="utf-8")
    investigator = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "investigator.md"
    ).read_text(encoding="utf-8")
    analyst = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "diagnosis-analyst.md"
    ).read_text(encoding="utf-8")

    # Route handoff is mechanism-based: the family file owns specialist
    # selection, while the orchestrator only propagates its explicit contract.
    assert "reads only its matching classification family subskill" in catalog
    assert "explicit `Specialist Loading Contract`" in catalog
    assert "do not infer specialist loads directly from symptom keywords" in catalog
    assert "This rule is identical for every selection source" in investigator
    assert "exact selected family subskill path" in investigator
    assert "mandatory execution contract, not an optional allowance" in investigator

    # The analyst must execute that handoff before synthesis but still treats
    # the user-selected route as a hypothesis that can be rejected.
    assert "mandatory first-round read" in analyst
    assert "load every specialist named in the required list before synthesis" in analyst
    assert "Every task-selected route is a candidate to test, not a conclusion" in analyst


def test_multi_route_selection_loads_every_selected_family_contract() -> None:
    investigator = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "investigator.md"
    ).read_text(encoding="utf-8")
    analyst = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "diagnosis-analyst.md"
    ).read_text(encoding="utf-8")

    assert "write all selected classes to `selection.routes`" in investigator
    assert "one analyst's `Required skills to load`" in investigator
    assert "every matching classification family subskill" in analyst
    assert "Compare every selected branch explicitly" in analyst


def test_fast_diagnosis_is_isolated_and_explicitly_routed() -> None:
    skill_path = SKILLS_DIR / "builtin" / "tidb-fast-diagnosis" / "SKILL.md"
    skill_text = skill_path.read_text(encoding="utf-8")
    protocol = build_static_protocol()

    assert "name: tidb-fast-diagnosis" in skill_text
    assert "Use only when the user explicitly asks" in skill_text
    assert "--profile fast" in skill_text
    assert "at most three" in skill_text
    assert "never display" in skill_text
    assert "evidences/fast-diagnosis-state.json" in skill_text
    assert "Explicit fast-diagnosis requests are a separate diagnosis mode" in protocol
    assert "explicit override takes precedence over Ad-hoc data lookup" in protocol
    assert "Preserve any user-requested output schema" in protocol
    assert "Do not enter fast mode for an ordinary diagnosis request" in protocol
    assert "strict machine-readable schema or exact output format" in skill_text

    schema = json.loads(
        (
            AGENT_CONFIG_DIR
            / "investibook"
            / "schemas"
            / "fast-diagnosis-state.schema.json"
        ).read_text(encoding="utf-8")
    )
    assert schema["properties"]["directions"]["maxItems"] == 3
    assert (
        schema["properties"]["directions"]["items"]["properties"]["confidence"][
            "maximum"
        ]
        == 1
    )


def test_code_investigator_schema_uses_version_scoped_source_references() -> None:
    schema = json.loads(
        (
            AGENT_CONFIG_DIR
            / "investibook"
            / "schemas"
            / "code-investigator-output.schema.json"
        ).read_text(encoding="utf-8")
    )
    pattern = schema["properties"]["confirmed_code_behavior"]["items"]["properties"][
        "evidence"
    ]["items"]["pattern"]
    repo_values = schema["properties"]["repo"]["enum"]
    context_pattern = schema["properties"]["code_context"]["properties"]["context_id"]["pattern"]

    assert repo_values == ["pingcap/tidb", "pingcap/tikv"]
    assert re.fullmatch(context_pattern, "tidb:" + "a" * 40 + ":source")
    assert re.fullmatch(context_pattern, "tikv:" + "a" * 40 + ":source")
    assert re.fullmatch(pattern, "pkg/util/topsql/reporter.go:214")
    assert re.fullmatch(pattern, "pkg/util/topsql/reporter.go:214-227")
    assert not re.fullmatch(pattern, "pkg/util/topsql/reporter.go:214@" + "a" * 40)


def test_layered_diagnosis_contract_exposes_signal_and_critical_evidence_fields() -> (
    None
):
    schemas_dir = AGENT_CONFIG_DIR / "investibook" / "schemas"
    evidence_schema = json.loads(
        (schemas_dir / "evidence-agent-output.schema.json").read_text(encoding="utf-8")
    )
    analyst_schema = json.loads(
        (schemas_dir / "diagnosis-analyst-output.schema.json").read_text(
            encoding="utf-8"
        )
    )

    assert "critical_evidence_hints" in evidence_schema["properties"]
    assert "signal_tags" in analyst_schema["properties"]
    assert "fault_classification" in analyst_schema["properties"]
    assert "selected_diagnosis_skills" in analyst_schema["properties"]
    assert "critical_evidence_hints" in analyst_schema["properties"]
    assert "signal_tags" in analyst_schema["required"]
    assert "fault_classification" in analyst_schema["required"]
    assert "selected_diagnosis_skills" in analyst_schema["required"]
    assert "critical_evidence_hints" in analyst_schema["required"]

    signal_tag_schema = analyst_schema["properties"]["signal_tags"]["items"]
    assert signal_tag_schema["required"] == [
        "id",
        "source",
        "component",
        "metric_or_signal",
        "direction",
        "baseline",
        "severity",
        "confidence",
        "evidence_ref",
    ]
    assert "comparison" not in signal_tag_schema["properties"]
    assert "evidence_path" not in signal_tag_schema["properties"]

    classification_schema = analyst_schema["properties"]["fault_classification"]
    assert classification_schema["required"] == [
        "primary_class",
        "candidates",
        "rejected_classes",
        "classification_confidence",
        "missing_evidence_for_higher_confidence",
    ]
    assert (
        "resource_control_or_capacity"
        in classification_schema["properties"]["primary_class"]["enum"]
    )
    assert (
        "sql_plan_or_query"
        in classification_schema["properties"]["primary_class"]["enum"]
    )
    assert (
        "service_unavailability_or_worker_failure"
        in classification_schema["properties"]["primary_class"]["enum"]
    )

    premise_schema = analyst_schema["properties"]["premise_verification"]["items"]
    assert premise_schema["required"] == ["claim", "dimension", "verdict"]
    assert "refuted" in premise_schema["properties"]["verdict"]["enum"]

    candidate_item_schema = classification_schema["properties"]["candidates"]["items"]
    assert "oneOf" in candidate_item_schema
    assert {"type": "string"} in candidate_item_schema["oneOf"]
    structured_candidate_schema = next(
        item for item in candidate_item_schema["oneOf"] if item.get("type") == "object"
    )
    assert structured_candidate_schema["required"] == ["class", "confidence", "role"]
    assert structured_candidate_schema["properties"]["confidence"]["minimum"] == 0
    assert structured_candidate_schema["properties"]["confidence"]["maximum"] == 1
    assert "contributing" in structured_candidate_schema["properties"]["role"]["enum"]


def test_agent_prompts_require_critical_evidence_handoff() -> None:
    agents_dir = AGENT_CONFIG_DIR / "investibook" / "agents"
    evidence_prompt = (agents_dir / "evidence-agent.md").read_text(encoding="utf-8")
    analyst_prompt = (agents_dir / "diagnosis-analyst.md").read_text(encoding="utf-8")

    # Evidence agent: two collection modes, anomaly-centric handoff. Sweep
    # collects without interpreting, so it loads no skills or knowledge.
    assert "Sweep Mode" in evidence_prompt
    assert "Targeted Mode" in evidence_prompt
    assert (
        "Sweep mode collects without interpreting and needs no skill loads"
        in evidence_prompt
    )
    assert "no skill loads, no knowledge-file reads" in evidence_prompt
    # Targeted mode: slow remote queries run concurrently, not one turn each.
    assert "Run independent data commands concurrently" in evidence_prompt
    assert "never queue independent slow queries one turn at a time" in evidence_prompt
    assert "Anomaly Hints" in evidence_prompt
    assert "whether the signal is present or absent" in evidence_prompt
    assert "copy or reuse evidence from another workspace" in evidence_prompt
    assert "use the canonical user/request window exactly" in evidence_prompt
    # Time conversion is verified by round trip, never by hardcoded epoch tables.
    assert "convert each epoch value back to UTC ISO" in evidence_prompt
    assert not re.search(r"\b17[0-9]{8}\b", evidence_prompt), (
        "hardcoded epoch values must not return"
    )
    assert (
        "Prefer the runtime/backend-written `evidences/agent-runs/evidence-agent.json`"
        in evidence_prompt
    )
    assert (
        "its mtime is not older than the current summary/report artifacts"
        in evidence_prompt
    )
    assert "capacity_comparison_context_present" in evidence_prompt
    assert "coprocessor_cascade_signal_present" in evidence_prompt
    # Hard fast handoff: the fast path overrides verbose Task prompts and the
    # handoff is a single status line with no extra schema fields.
    assert (
        "This section overrides any broader or more verbose request in the Task prompt"
        in evidence_prompt
    )
    assert "must not read it into the LLM chat context" in evidence_prompt
    assert (
        "evidence-agent.json written: status=<status> files=<n> hints=<n> missing=<n>"
        in evidence_prompt
    )
    assert (
        "do not add `symptom_summary`, `artifacts_collected`, `evidence_sufficient`, or"
        in evidence_prompt
    )
    # Analyst: anomalies -> causal chain process over the contract vocabulary.
    assert "From Anomalies to Cause" in analyst_prompt
    assert "anomaly timeline" in analyst_prompt
    assert "Layered Diagnosis Checklist" in analyst_prompt
    assert "signal_tags" in analyst_prompt
    assert "fault_classification" in analyst_prompt
    assert "metric_or_signal" in analyst_prompt
    assert "evidence_ref" in analyst_prompt
    assert "classification_confidence" in analyst_prompt
    assert "Do not use aliases such as `tag`, `metric`, `signal`" in analyst_prompt
    assert "expanded windows" in analyst_prompt
    # Domain reasoning lives in families/specialists, not the role prompt;
    # the prompt only points at it. Canned verdicts must not return anywhere.
    assert "Evidence-Weighing Principles" in analyst_prompt
    assert "plan cache is not the root cause" not in analyst_prompt, (
        "canned verdict wording must not return"
    )
    assert "Bounded Fast Diagnosis" in analyst_prompt
    assert "handoff summary from evidence collection" in analyst_prompt
    assert "Do not read `evidences/workspace_evidence_report.json`" in analyst_prompt
    assert "stop tool use and return a short handoff message" in analyst_prompt
    assert "replacing the whole file content" in analyst_prompt
    assert "python3 -m json.tool" in analyst_prompt
    # The evidence-weighing principles live in diagnosis-overview so the
    # investigator's direct-conclusion path carries the same domain knowledge
    # as the analyst; the clinic playbook keeps a pointer.
    skill_text = (
        SKILLS_DIR / "builtin" / "tidb-clinic-diagnosis" / "SKILL.md"
    ).read_text(encoding="utf-8")
    assert "Evidence-Weighing Principles" in skill_text
    principles_text = (
        SKILLS_DIR / "builtin" / "diagnosis-overview" / "SKILL.md"
    ).read_text(encoding="utf-8")
    assert "Evidence-Weighing Principles" in principles_text
    assert "Capacity differential" in principles_text
    assert "fewer TiDB nodes" in principles_text
    assert "baseline=peer_nodes" in principles_text
    assert "rank alone does not make it the primary cause" in principles_text
    assert "large scan, full scan, ineffective index" in principles_text
    assert "decoded plan is missing" in principles_text
    # Background debt (compaction/GC) reaches foreground latency only through
    # a throttle path (write stall / flow control / admission control).
    assert "Background debt hurts through a throttle path" in principles_text
    assert "flow control / admission control" in principles_text
    assert "Overlap is not cause" in principles_text
    assert "plan cache is not the root cause" not in skill_text, (
        "canned verdict wording must not return"
    )
    assert "plan cache is not the root cause" not in principles_text, (
        "canned verdict wording must not return"
    )
    investigator_prompt = (agents_dir / "investigator.md").read_text(encoding="utf-8")
    assert "Finalization rule" in investigator_prompt
    assert "answer from it immediately" in investigator_prompt
    assert (
        ".opencode/investibook/schemas/diagnosis-analyst-output.schema.json"
        in investigator_prompt
    )
    assert "run_in_background=false" in investigator_prompt
    assert "Background task launched" in investigator_prompt
    # Investigation loop: plan -> sweep -> anomalies -> deep plan -> iterate.
    assert "Intent Gate" in investigator_prompt
    assert "Investigation Loop" in investigator_prompt
    assert (
        "collect evidence → surface anomalies → reason from anomalies to cause"
        in investigator_prompt
    )
    assert "sweep mode" in investigator_prompt
    assert "targeted mode" in investigator_prompt
    # User-directed collection: an explicit source/component directive routes
    # to targeted mode first; sweep stays the no-direction default.
    assert "when the user explicitly directs collection" in investigator_prompt
    assert "dispatch targeted mode first" in investigator_prompt
    assert "the user gave no direction" in investigator_prompt
    # Agent prompts stay in English; CJK text in the process layer is almost
    # always an eval-case phrase leaking in (overfitting) — none allowed.
    for prompt_name, prompt_text in (
        ("investigator", investigator_prompt),
        ("evidence-agent", evidence_prompt),
        ("diagnosis-analyst", analyst_prompt),
    ):
        assert not re.search(r"[一-鿿]", prompt_text), (
            f"{prompt_name} prompt contains CJK text - likely eval-case phrasing"
        )
    assert "Deep plan" in investigator_prompt
    assert (
        "treated as initial hypotheses and anomalies to verify, never as conclusions"
        in investigator_prompt
    )
    # Verification discipline — one rule instead of per-failure patches: no
    # claim (user premise, collector hint, own hypothesis) is evidence until
    # confirmed in its own dimension in the underlying data. Domain mapping
    # lives in the skill's Claim Verification Guide; observed failure modes
    # live in incorrect-conclusions.json; the outcome is a schema field.
    assert "Verification discipline" in investigator_prompt
    assert (
        "a claim is not evidence until confirmed in its own dimension"
        in investigator_prompt
    )
    assert "Claim Verification Guide" in investigator_prompt
    assert "not the handoff's emphasis" in investigator_prompt
    assert "premise_verification" in investigator_prompt
    assert "A refuted claim is a finding" in investigator_prompt
    assert (
        "The conclusion still must pass the verification discipline"
        in investigator_prompt
    )
    assert "Conclude yourself" not in investigator_prompt
    assert "open the answer with the verification verdict" in investigator_prompt
    assert (
        "only after confirming the signal in the evidence file it cites"
        in analyst_prompt
    )
    assert "premise_verification" in analyst_prompt
    assert "Claim Verification Guide" in analyst_prompt
    # The claim-verification guide is a standalone core skill; the clinic
    # playbook keeps only a pointer to it.
    overview_text = (
        SKILLS_DIR / "builtin" / "diagnosis-overview" / "SKILL.md"
    ).read_text(encoding="utf-8")
    assert "Claim Verification Guide" in overview_text
    assert "tikv_grpc_msg_duration" in overview_text
    assert "absence from a top-N view is not normality" in overview_text
    assert "premise_verification" in overview_text
    # Overview comparisons: the two per-entity latency ladders come first and
    # place the tier and entity; skew hints surface them from the sweep.
    assert "Overview Comparisons" in overview_text
    assert "store_latency_skew_present" in overview_text
    assert "store_latency_skew_present" in evidence_prompt
    assert "tidb_latency_skew_present" in evidence_prompt
    # Resource cross-check: a latency-elevated entity is compared to peers on
    # resource metrics; unknown metrics resolve via the PSC panel templates,
    # single-sourced in references/metric-discovery.md.
    assert "Resource Cross-Check" in overview_text
    assert "references/metric-discovery.md" in overview_text
    # Latency breakdown tree (docs.pingcap.com latency-breakdown) abstracted in
    # the skill; raw stage->metric templates in references/ (progressive load).
    assert "Latency Breakdown" in overview_text
    assert "references/latency-breakdown-metrics.md" in overview_text
    breakdown_ref = (
        SKILLS_DIR
        / "builtin"
        / "diagnosis-overview"
        / "references"
        / "latency-breakdown-metrics.md"
    ).read_text(encoding="utf-8")
    assert "tikv_raftstore_store_wf" in breakdown_ref
    assert "tidb_session_execute_duration_seconds" in breakdown_ref
    assert "tikv_grpc_msg_duration_seconds" in breakdown_ref
    assert "diagnosis-overview" in skill_text
    pitfalls = json.loads(
        (
            SKILLS_DIR
            / "builtin"
            / "tidb-knowledge-base"
            / "knowledge"
            / "incorrect-conclusions.json"
        ).read_text(encoding="utf-8")
    )
    pitfall_ids = {entry["id"] for entry in pitfalls["entries"]}
    assert {"ic-003", "ic-004", "ic-005", "ic-006"} <= pitfall_ids
    # Guardrail evidence is deterministic on every conclusion path: both agents
    # record the KB Step A/B markers so the judge can verify from artifacts
    # even when telemetry misses the file reads.
    assert "KB Step B complete: Pitfall catalog loaded" in investigator_prompt
    assert "KB Step A complete: tier architecture loaded" in investigator_prompt
    assert (
        "record `KB Step B complete: Pitfall catalog loaded` in both `quality_checks` and `final_response`"
        in analyst_prompt
    )
    assert "KB Step A complete: tier architecture loaded" in analyst_prompt
    assert "Reading a known file is never a reason to dispatch" in investigator_prompt
    assert "Do not re-collect a fully covered historical window" in investigator_prompt
    assert "Delegation trust" in investigator_prompt
    # Full diagnosis synthesis always crosses the analyst boundary when Task is
    # available, so standard-family specialist loading cannot be bypassed by a
    # clear-looking investigator conclusion.
    assert (
        "every full diagnosis synthesis belongs to `diagnosis-analyst`"
        in investigator_prompt
    )
    assert "Single-agent fallback conclusion" in investigator_prompt
    assert (
        "Never use this fallback merely because the evidence looks clear"
        in investigator_prompt
    )
    assert "Dispatch the analysis" in investigator_prompt
    assert "loads the matching family subskill" in investigator_prompt
    assert "Evidence-Weighing Principles" in investigator_prompt
    assert "hard prerequisite" in investigator_prompt
    assert "tier-architecture.json" in investigator_prompt
    # Component knowledge on demand supports route planning but cannot replace
    # the analyst or run the collector.
    assert (
        "never to replace the analyst"
        in investigator_prompt
    )
    assert (
        "write_analyst_report.py` is reserved for the single-agent fallback"
        in investigator_prompt
    )
    assert "never run `collect_evidence.py` yourself" in investigator_prompt
    # Novel or disputed cases dispatch with no pre-work; plain uncertainty gets
    # at most one cheap move before the inevitable analyst handoff.
    assert "no pre-work" in investigator_prompt
    assert "at most one cheap raw check or targeted collection" in investigator_prompt
    # Escalation checkpoint: a diagnosis request is itself the mandate; never
    # end an automated investigation turn with a question.
    assert "Escalation checkpoint" in investigator_prompt
    assert "is itself the mandate" in investigator_prompt
    assert "never end an investigation turn with a question" in investigator_prompt
    assert "invitation for hints" in investigator_prompt
    assert '"agent": "investigator"' in investigator_prompt
    assert "write_analyst_report.py" in investigator_prompt
    # The single-agent fallback still goes through the same validated writer.
    assert "--from-stdin" in investigator_prompt
    assert "Always go through the script" in investigator_prompt
    assert "read_raw_evidence.py" in investigator_prompt
    assert "never state the expected conclusion" in investigator_prompt
    # Task prompt hygiene: no restated contracts, no predefined conclusions.
    assert "Never restate collector mechanics" in investigator_prompt
    assert "never state or constrain the expected root cause" in investigator_prompt
    assert "no extra-schema fields (`symptom_summary`" in investigator_prompt
    assert "does not judge whether the evidence is enough" in investigator_prompt
    assert "minimal handoff request" in investigator_prompt
    assert "in one parallel round" in investigator_prompt
    # Root-cause classification subskill routing (upstream subskill split).
    assert "root-cause-classification/SUBSKILL.md" in analyst_prompt
    assert "standard_root_cause_class" in analyst_prompt
    # Raw evidence escalation goes through the bounded reader, never bulk reads.
    assert "read_raw_evidence.py" in analyst_prompt
    assert "never bulk-read `evidences/raw/`" in analyst_prompt
    # Round-count discipline: batched parallel reads (skills load as files in
    # the same round, not sequential skill-tool loads) and the report writer.
    assert "Batch your reads" in analyst_prompt
    assert "Load lean" in analyst_prompt
    assert "instead of sequential `skill` tool loads" in analyst_prompt
    assert "write_analyst_report.py" in analyst_prompt
    # Artifact economy: the agent hand-writes only what carries information;
    # the report writer completes defaults, hints carry-forward, derivables.
    # final_response is a ~10-line compact record, never the essay twice.
    assert "Artifact economy" in analyst_prompt
    assert "the report writer fills them" in analyst_prompt
    assert "carries the evidence-agent hints forward unchanged" in analyst_prompt
    assert "artifact-economy rules apply" in investigator_prompt
    assert "never write the full answer twice" in analyst_prompt
    assert "roughly 10 lines" in analyst_prompt
    assert "compact record, not prose to copy" in investigator_prompt
    # Scope is the canonical user/request window; query slices must never
    # masquerade as the investigation window in the report contract fields.
    assert "never a narrower query or sampling slice" in analyst_prompt
    # Rejection carries the same verification burden as promotion.
    overview = (SKILLS_DIR / "builtin" / "diagnosis-overview" / "SKILL.md").read_text(
        encoding="utf-8"
    )
    assert "Rejection needs its own dimension too" in overview
    assert "mark the candidate unverified, not rejected" in overview
    # A dimension whose collection failed wholesale was never observed.
    assert "collection failed wholesale" in overview
    # Conclusion Contract: the four cross-case quality gates (terminal
    # mechanism, weighing-decided primary, known-issue correlation,
    # four-horizon remediation) live in the overview skill; both concluding
    # paths point at it.
    assert "Conclusion Contract" in overview
    # Ordering: weighing selects the primary class before the mechanism is
    # named — deep-diving the first collectable mechanism promotes a
    # real-but-secondary cause over the true primary.
    assert "Weighing selects the primary class first" in overview
    assert "Name the terminal mechanism of the selected primary" in overview
    assert "weighing selects the primary before you name its mechanism" in overview
    assert "Correlate with known issues" in overview
    assert "cite the issue id and fixed version" in overview
    assert "Remediation covers four horizons" in overview
    assert "Conclusion Contract" in analyst_prompt
    assert "Conclusion Contract" in investigator_prompt
    # Unobservable-dimension discipline: a failed/empty collection is not a
    # negative finding; a competing class cannot be ruled out by absence.
    assert "A failed or empty collection is not a negative finding" in overview
    assert "cap your `classification_confidence`" in overview
    assert "failed to collect" in overview  # ic-006 covers wholesale collection failure
    # Metric-name discovery is a shared reference in the core-group skill so
    # every agent can reach it; clinic-cli keeps a pointer, not a copy.
    discovery = (
        SKILLS_DIR
        / "builtin"
        / "diagnosis-overview"
        / "references"
        / "metric-discovery.md"
    ).read_text(encoding="utf-8")
    assert "psc_knowledge_base" in discovery
    assert "Never guess" in discovery or "not a guess-per-call loop" in discovery
    assert "references/metric-discovery.md" in overview
    clinic_metrics = (
        SKILLS_DIR / "builtin" / "clinic-cli" / "metrics" / "SUBSKILL.md"
    ).read_text(encoding="utf-8")
    assert "references/metric-discovery.md" in clinic_metrics
    assert "psc_knowledge_base" not in clinic_metrics.replace(
        "psc_knowledge_base_<tier>.json", ""
    ), "clinic-cli keeps only the pointer, not a second copy of the lookup"


def test_sql_specialist_routing_is_mechanism_based_and_bounded() -> None:
    analyst_prompt = (
        AGENT_CONFIG_DIR / "investibook" / "agents" / "diagnosis-analyst.md"
    ).read_text(encoding="utf-8")
    clinic_skill = (
        SKILLS_DIR / "builtin" / "tidb-clinic-diagnosis" / "SKILL.md"
    ).read_text(encoding="utf-8")
    sql_family = (
        SKILLS_DIR
        / "builtin"
        / "tidb-clinic-diagnosis"
        / "sql-plan-or-query"
        / "SUBSKILL.md"
    ).read_text(encoding="utf-8")
    autox_skill = (SKILLS_DIR / "tidb-query-tuning-autox" / "SKILL.md").read_text(
        encoding="utf-8"
    )
    autox_sync = (PROJECT_ROOT / "scripts" / "sync-autox-skill.sh").read_text(
        encoding="utf-8"
    )

    # SQL specialist loading follows causal classification, not one fixed TiKV
    # hotspot shape or the mere presence of a ranked query.
    assert "SQL specialist candidate conditions" in sql_family
    assert (
        "Do not require TiKV CPU or a fixed TopSQL percentage for non-CPU claims"
        in sql_family
    )
    assert "query rank alone never does" in analyst_prompt
    assert "material contributing candidate with confidence >= 0.60" in analyst_prompt

    # AutoX must stay a bounded diagnosis-only specialist over existing evidence.
    assert ".opencode/.all-skills/tidb-query-tuning-autox/SKILL.md" in sql_family
    assert (
        "use only the classification workflow over existing evidence" in analyst_prompt
    )
    assert (
        "never AutoX collection, batch, or local-validation workflows" in analyst_prompt
    )
    assert (
        "always hand-write the actual loaded shared skills"
        in analyst_prompt
    )
    assert "validation level" in sql_family
    assert "primary or material contributing candidate" in autox_skill
    assert "primary or material contributing candidate" in autox_sync
    assert "affected-only query into a root cause" in autox_skill

    # Independent non-SQL failures still prevent accidental AutoX activation.
    assert "fully explains the symptom" in sql_family
    assert "SQL specialist candidate conditions" not in clinic_skill
    assert "affected-only" in analyst_prompt


def test_static_protocol_includes_diagnosis_orchestration_rules() -> None:
    protocol = build_static_protocol()

    assert "Request triage" in protocol
    assert "Authorized code-diagnosis override" in protocol
    assert "This override precedes every request-triage route" in protocol
    assert "Diagnosis orchestration" in protocol
    assert "investigator" in protocol
    assert "evidence-agent" in protocol
    assert "diagnosis-analyst" in protocol
    assert "File conventions" in protocol
    assert "needs_more_evidence=true" in protocol
    assert "one supplemental evidence round" in protocol
    assert "go_gc_duration_seconds" in protocol
    assert "go_gc_pauses_seconds" in protocol
    assert "bounded fast diagnosis" in protocol
    # Tiered dispatch: coverage check comes before any collection.
    assert "already covers the requested scope" in protocol
    assert "invokes `diagnosis-analyst` for synthesis" in protocol
    assert "delegates every full diagnosis synthesis to `diagnosis-analyst`" in protocol
    assert "clear cases use its bounded fast path" in protocol
    assert (
        "collect evidence → surface anomalies → select a diagnosis route → reason from anomalies to cause"
        in protocol
    )
    assert "sweep or targeted mode" in protocol
    assert "whole-file valid JSON" in protocol
    # Per-agent contract details live in the agent prompt files, not in the protocol.
    assert ".opencode/investibook/agents/" in protocol
    assert "single source of truth" in protocol
    assert "finalize immediately" in protocol
    assert "run_in_background=false" in protocol
    assert "Background task launched" in protocol
    assert "Ad-hoc data lookup" in protocol
    assert "Stay in the main agent" in protocol
    assert "`collect_evidence.py`" in protocol
    assert "nutshell `clinic-api`" in protocol
    assert (
        ".opencode/.all-skills/nutshell-skills/platform/clinic-api/SKILL.md" in protocol
    )
    assert "execute through builtin `clinic-cli`" in protocol
    assert "never by evaluation case names" in protocol