宜博 / lemongraph

"""p_validator_root_cause_trace (04-P §4.3): U-015 scope.

"血缘读取"（tracing_lineage: reading SDD/PRD/BRD versions + Coder implementation
records） is the caller's job -- perspective_fns are expected to already have
whatever lineage context they need baked in (e.g. closures over artifact
content read beforehand) by the time judge_root_cause calls them. This module
only drives judging_root_cause's multi-perspective vote and the
bug_recorded/inconclusive_escalate branch, matching 04-P's actual state-table
scope for that transition.

Each perspective_fn takes no arguments and returns {"category": "planning"|
"coder"|"skill_defect", "detail": str} -- one judgment from one of the three
perspectives 04-P names (正确性视角/需求完整性视角/Skill 系统性缺陷视角). Which
concrete question each perspective asks is a prompt-engineering concern for
the caller (see real_validator_perspective below for the real-call version),
not this module's.

**F (2026-08-06, docs/done/verifier-fan-out-jury-design.md)**: judge_root_cause
above is the closure-based implementation -- each perspective is a Python
function call, sequentially, in the SAME process/context. Kept as-is (not
replaced): it's the cheap path for scenarios that don't need worktree/Work
Graph overhead, and its already-proven >=2/3 rule is now shared (via
_tally_perspectives) with judge_root_cause_via_work_graph below, the real
Work Graph fan-out version -- each perspective becomes an independent Work
Node (own claude -p call, own clean context, own execution_event log
entries), scheduled by G's topological_layers()/run_graph() same as any
other graph_manifest. The fan-in vote itself stays plain Python (a
zero_token aggregation per the design-dynamic-work-graph Skill's own
edge_type distinction -- counting doesn't need an Agent call), not a
fourth Work Node.
"""

import json
import uuid
from collections import Counter

from src import bug_record, work_graph_orchestrate


class InconclusiveEscalateError(Exception):
    """04-P: judging_root_cause -> inconclusive_escalate (<2/3 perspectives
    agree). Carries the raw per-perspective judgments for human review
    ("附带三视角原始判断供人工裁决")."""

    def __init__(self, judgments):
        self.judgments = judgments
        super().__init__("perspectives did not reach 2/3 agreement: " + repr(judgments))


def _tally_perspectives(judgments):
    """>=2/3 agreement rule (04-P: "至少 2/3 视角一致才定案") -- shared by both
    judge_root_cause (closures) and judge_root_cause_via_work_graph (Work
    Graph fan-out), so the rule is one piece of code, not two that could
    drift apart. `judgments`: list of {"category", "detail"}."""
    counts = Counter(j["category"] for j in judgments)
    winner, count = counts.most_common(1)[0]
    if count * 2 < len(judgments):  # strictly "at least 2/3", not just plurality
        raise InconclusiveEscalateError(judgments)
    detail = " | ".join(j["detail"] for j in judgments if j["category"] == winner)
    return {"root_cause_category": winner, "root_cause_detail": detail, "judgments": judgments}


def judge_root_cause(perspective_fns):
    """Runs all perspective_fns, requires >=2/3 to agree on the same
    category (04-P guard: "至少 2/3 视角一致才定案"). Returns
    {"root_cause_category", "root_cause_detail", "judgments"} on agreement,
    raises InconclusiveEscalateError otherwise.
    """
    judgments = [fn() for fn in perspective_fns]
    return _tally_perspectives(judgments)


def trace_root_cause_test_failed(conn, task_id, description, perspective_fns, suggested_route, related_artifact_ids=None):
    """04-P: test_failed -> tracing_lineage -> judging_root_cause ->
    bug_recorded. Creates a brand-new bug_record (test_failed origin never
    updates an existing one, unlike human_reported -- see 04-P's own note
    on this distinction). Re-raises InconclusiveEscalateError unchanged if
    judge_root_cause can't reach agreement.
    """
    result = judge_root_cause(perspective_fns)
    created = bug_record.create_bug_record(
        conn, task_id=task_id, description=description,
        root_cause_category=result["root_cause_category"], root_cause_detail=result["root_cause_detail"],
        suggested_route=suggested_route, related_artifact_ids=related_artifact_ids,
    )
    return {**created, **result}


def trace_root_cause_human_reported(conn, bug_id, perspective_fns, suggested_route):
    """04-P: human_reported (already claimed via bug_record.claim_human_reported_bug,
    status=triaging) -> tracing_lineage -> judging_root_cause -> bug_recorded.
    Field-completes the existing bug_record instead of creating a new one.
    """
    result = judge_root_cause(perspective_fns)
    bug_record.complete_human_reported_triage(
        conn, bug_id, result["root_cause_category"], result["root_cause_detail"], suggested_route
    )
    return result


_PERSPECTIVE_FRAMINGS = {
    "correctness": (
        "CORRECTNESS PERSPECTIVE: does the Coder's implementation match what the SDD/PRD actually "
        "specify? If the spec was followed correctly and the bug is still there, this is not your "
        "category to claim."
    ),
    "requirement_completeness": (
        "REQUIREMENT-COMPLETENESS PERSPECTIVE: does the PRD/SDD actually define the behavior the bug "
        "report is about at all? If the requirement was never specified (a genuine planning gap, not "
        "an implementation mistake), that's your category to claim."
    ),
    "skill_defect": (
        "SKILL-DEFECT PERSPECTIVE: is this a one-off mistake in this specific task, or does it look "
        "like the underlying Skill (the reusable methodology/prompt an Org Agent follows) is "
        "systematically wrong in a way that would recur on other tasks?"
    ),
}


def real_validator_perspective(perspective_name, bug_description, context, target_folder_path, trigger_fn, timeout=300):
    """Builds a real perspective_fn (for use with judge_root_cause) that calls
    the deployed Validator agent, framed from one of the three named
    perspectives, and parses its 'CATEGORY: ...' / 'DETAIL: ...' response.
    `trigger_fn` is org_node_trigger.trigger_org_node, passed in explicitly
    (not imported/called directly) so tests can inject a fake without
    mocking a module-global.
    """
    framing = _PERSPECTIVE_FRAMINGS[perspective_name]
    prompt = (
        "{framing}\n\nBug: {bug_description}\n\nContext:\n{context}\n\n"
        "Reply with exactly two lines:\n"
        "CATEGORY: planning|coder|skill_defect\n"
        "DETAIL: <one sentence, cite the specific SDD section or code location>"
    ).format(framing=framing, bug_description=bug_description, context=context)

    def perspective_fn():
        response = trigger_fn("validator", prompt, target_folder_path, timeout=timeout)
        category = None
        detail = ""
        for line in response.splitlines():
            stripped = line.strip()
            if stripped.upper().startswith("CATEGORY:"):
                value = stripped.split(":", 1)[1].strip().lower()
                if value in ("planning", "coder", "skill_defect"):
                    category = value
            elif stripped.upper().startswith("DETAIL:"):
                detail = stripped.split(":", 1)[1].strip()
        if category is None:
            raise ValueError("could not parse a CATEGORY line from validator response: " + response)
        return {"category": category, "detail": detail or response}

    return perspective_fn


_PERSPECTIVE_INPUT_SCHEMA = {"type": "object", "required": ["node_id"]}
_PERSPECTIVE_OUTPUT_SCHEMA = {"type": "object", "required": ["category", "detail"]}


def _perspective_prompt(perspective_name, bug_description, context):
    framing = _PERSPECTIVE_FRAMINGS[perspective_name]
    return (
        "{framing}\n\nBug: {bug_description}\n\nContext:\n{context}\n\n"
        "Reply with your judgment: category must be exactly one of 'planning', 'coder', or "
        "'skill_defect'; detail is one sentence citing the specific SDD section or code "
        "location that supports it."
    ).format(framing=framing, bug_description=bug_description, context=context)


def _node_id_for_perspective(name):
    return "perspective-" + name


def _build_jury_graph_manifest(perspective_names):
    """FR-1: a reusable fan-out graph_manifest shape -- N mutually
    independent perspective nodes, generic over any set of perspective
    names (not hardcoded to root-cause's three). No fan-in node: see this
    module's own docstring / design doc §1 for why the vote stays plain
    Python (zero_token aggregation, not worth an Agent call)."""
    return {
        "nodes": [{"node_id": _node_id_for_perspective(name), "depends_on": []} for name in perspective_names],
        "edges": [],
    }


def _jury_prompt_builder(perspective_prompts):
    """FR-2: perspective content is parameterized -- this scheduling code
    has no idea what the perspectives actually ask, it just runs whatever
    prompt each node was given."""
    def build_prompt(node, outputs):
        name = node["node_id"][len("perspective-"):]
        return perspective_prompts[name]
    return build_prompt


def judge_root_cause_via_work_graph(
    conn, task_id, target_folder_path, bug_description, context,
    org_node="validator", skill_id="qa-task-validation", timeout=300,
):
    """Work Graph fan-out version of judge_root_cause (FR-5's real landing
    scenario) -- each perspective is an independent Work Node (own
    `claude -p` call with real schema-constrained structured output, own
    clean context -- work_graph_orchestrate._make_run_fn's dynamic-
    workflow-script mechanism, validated for real 2026-08-01, see that
    module's docstring) instead of a Python closure called in a loop, so
    unlike real_validator_perspective there's no 'CATEGORY:'/'DETAIL:'
    text parsing needed -- the node's output_payload already IS
    {"category", "detail"}. Same return shape as judge_root_cause, and
    raises InconclusiveEscalateError on <2/3 agreement via the same shared
    _tally_perspectives call -- callers don't need to know which
    implementation they're getting.

    No plan_work_graph() call: this graph's shape (N independent nodes) is
    fixed and known ahead of time, same reasoning e2e_pipeline.py's
    implement_code()/validate_implementation_with_own_graph() already use
    for skipping an LLM planning step of their own."""
    perspective_prompts = {
        name: _perspective_prompt(name, bug_description, context) for name in _PERSPECTIVE_FRAMINGS
    }
    manifest = _build_jury_graph_manifest(list(perspective_prompts))

    run_id = str(uuid.uuid4())
    conn.execute(
        "INSERT INTO work_graph_run (run_id, task_id, org_node, graph_manifest, status) "
        "VALUES (?, ?, ?, ?, 'running')",
        (run_id, task_id, org_node, json.dumps(manifest)),
    )
    conn.commit()

    outputs = work_graph_orchestrate.run_graph(
        conn, run_id, org_node, skill_id, target_folder_path, manifest,
        _jury_prompt_builder(perspective_prompts), _PERSPECTIVE_INPUT_SCHEMA, _PERSPECTIVE_OUTPUT_SCHEMA,
        timeout=timeout,
    )
    judgments = [outputs[_node_id_for_perspective(name)] for name in perspective_prompts]
    return _tally_perspectives(judgments)