1"""Broadcast the same prompt to every model, audit consensus.
2
3The caller supplies a `query` callable so this package stays provider-agnostic:
4
5    def query(model_id: str, prompt: str) -> tuple[str, float]:
6        # return (answer_text, latency_seconds); raise on failure
7        ...
8
9    audit = broadcast_and_verify(
10        models=["gpt-4o", "claude-3.5", "deepseek-v3"],
11        prompt="What is 2+2?",
12        query=my_query_fn,
13    )
14"""
15import concurrent.futures
16import logging
17import time
18from typing import Callable, Dict, List, Optional
19
20from .audit import ModelAnswer, VerificationAudit, compare_answers
21
22logger = logging.getLogger("multi-model-verifier")
23
24QueryFn = Callable[[str, str], "tuple[str, float]"]
25
26
27def _query_one(model_id: str, prompt: str, query: QueryFn) -> ModelAnswer:
28    t0 = time.time()
29    try:
30        content, latency = query(model_id, prompt)
31        return ModelAnswer(
32            model_id=model_id, provider_name=model_id,
33            content=content, latency_seconds=latency,
34        )
35    except Exception as e:
36        return ModelAnswer(
37            model_id=model_id, provider_name=model_id, content="",
38            latency_seconds=time.time() - t0, error=str(e),
39        )
40
41
42def broadcast_and_verify(
43    models: List[str],
44    prompt: str,
45    query: QueryFn,
46    parallel: bool = True,
47    timeout_per_model: int = 60,
48) -> VerificationAudit:
49    """Send the same prompt to all models and produce a consensus audit.
50
51    Args:
52        models: Distinct model identifiers to query.
53        prompt: The prompt to broadcast.
54        query: (model_id, prompt) -> (answer_text, latency_seconds). Raise to report failure.
55        parallel: Query concurrently via a thread pool when True.
56        timeout_per_model: Per-model timeout in parallel mode.
57
58    Returns:
59        VerificationAudit with per-model answers, divergence score, consensus.
60    """
61    logger.info("Broadcasting to %d models: %s", len(models), ", ".join(models))
62    answers: List[ModelAnswer] = []
63
64    if parallel and len(models) > 1:
65        with concurrent.futures.ThreadPoolExecutor(max_workers=len(models)) as ex:
66            futures = {ex.submit(_query_one, m, prompt, query): m for m in models}
67            for fut in concurrent.futures.as_completed(futures, timeout=timeout_per_model * len(models)):
68                try:
69                    answers.append(fut.result(timeout=timeout_per_model))
70                except Exception as e:
71                    answers.append(ModelAnswer(
72                        model_id=futures[fut], provider_name=futures[fut],
73                        content="", error=f"timeout: {e}",
74                    ))
75    else:
76        for m in models:
77            answers.append(_query_one(m, prompt, query))
78
79    audit = compare_answers(answers)
80    audit.prompt = prompt
81    logger.info("Audit: %d models, divergence=%.2f, unanimous=%s",
82                len(answers), audit.divergence_score, audit.unanimous())
83    return audit
84
85
86def execute_verify_multi(
87    models: List[str],
88    prompt: str,
89    query: QueryFn,
90    threshold: float = 0.3,
91) -> Dict:
92    """Tool-call style wrapper. Returns a structured dict with a `blocked` flag.
93
94    Use as a pre-action gate: if `blocked` is True, the models disagreed beyond
95    `threshold` and a high-stakes action should be held for review.
96    """
97    audit = broadcast_and_verify(models, prompt, query)
98    blocked = audit.divergence_score > threshold
99    per_model = []
100    for a in audit.answers:
101        entry = {"model": a.model_id, "latency": round(a.latency_seconds, 2)}
102        entry["error"] = a.error if a.error else None
103        if not a.error:
104            entry["answer"] = a.content[:500]
105        per_model.append(entry)
106    return {
107        "consensus": audit.consensus[:1000],
108        "divergence_score": round(audit.divergence_score, 2),
109        "unanimous": audit.unanimous(),
110        "blocked": blocked,
111        "recommendations": audit.recommendations,
112        "per_model": per_model,
113        "summary": audit.to_summary(),
114    }
115