citation_verifier

zocrates.agents.library_search.citation_verifier

Fast probabilistic citation verification using decision models.

Evaluates synthesized claims against retrieved evidence chunks via remote (OpenRouter Decisions API) or local (Laya uv tool) decision models, returning calibrated support probabilities for attribution filtering and popover badges.

DEFAULT_COMPOUND_CITATION_INSTRUCTIONS module-attribute

DEFAULT_COMPOUND_CITATION_INSTRUCTIONS = 'This claim cites multiple sources. Does the `source_text` support a substantive part of the `claim`?'

DEFAULT_HEALTH_TIMEOUT_SEC module-attribute

DEFAULT_HEALTH_TIMEOUT_SEC = 0.15

DEFAULT_LOCAL_URL module-attribute

DEFAULT_LOCAL_URL = 'http://127.0.0.1:8005/v1/systemone'

DEFAULT_OPENROUTER_MODEL module-attribute

DEFAULT_OPENROUTER_MODEL = '~typesafe/jev-latest'

DEFAULT_RESPAN_COMPOUND_INSTRUCTIONS module-attribute

DEFAULT_RESPAN_COMPOUND_INSTRUCTIONS = 'This claim cites multiple sources. Does the source text support a substantive part of the claim?'

DEFAULT_RESPAN_SINGLE_INSTRUCTIONS module-attribute

DEFAULT_RESPAN_SINGLE_INSTRUCTIONS = 'Does the source text support the claim?'

DEFAULT_SINGLE_CITATION_INSTRUCTIONS module-attribute

DEFAULT_SINGLE_CITATION_INSTRUCTIONS = 'Does the `source_text` support the `claim`?'

DEFAULT_THRESHOLD module-attribute

DEFAULT_THRESHOLD = 0.6

DEFAULT_TIMEOUT_SEC module-attribute

DEFAULT_TIMEOUT_SEC = 10.0

DEFAULT_TURN_BUDGET_SEC module-attribute

DEFAULT_TURN_BUDGET_SEC = 20.0

MAX_CLAIM_CHARS module-attribute

MAX_CLAIM_CHARS = 800

MAX_LOCAL_SOURCE_TEXT_CHARS module-attribute

MAX_LOCAL_SOURCE_TEXT_CHARS = 1800

MAX_SOURCE_TEXT_CHARS module-attribute

MAX_SOURCE_TEXT_CHARS = 6000

MIN_CLAIM_CHARS module-attribute

MIN_CLAIM_CHARS = 30

OPENROUTER_CHAT_URL module-attribute

OPENROUTER_CHAT_URL = 'https://openrouter.ai/api/v1/chat/completions'

OPENROUTER_DECISIONS_URL module-attribute

OPENROUTER_DECISIONS_URL = 'https://openrouter.ai/api/alpha/decisions'

_ABBREVIATION_RE module-attribute

_ABBREVIATION_RE = re.compile('\\b(?:et al|e\\.g|i\\.e|etc|vs|cf|fig|eq|no|sec|approx|ca|dr|prof|mr|mrs|jr|sr|[a-z])\\.?$', re.IGNORECASE)

_AUTHOR_ONLY_RE module-attribute

_AUTHOR_ONLY_RE = re.compile("^(?:according to|as (?:shown|noted|described|stated|reported) by|in)?\\s*(?:[A-Z][a-zA-Z\\-\\']+(?:,\\s*|\\s+and\\s+|\\s*&\\s*|\\s+et\\s+al\\.?\\s*)*)(?:\\s*\\(\\s*\\d{4}[a-z]?\\s*\\))?\\s*$", re.IGNORECASE)

_CODE_BLOCK_RE module-attribute

_CODE_BLOCK_RE = re.compile('(```[\\s\\S]*?```|~~~[\\s\\S]*?~~~|`[^`\\n]+`)')

_IN_TEXT_CITATION_TOKEN_RE module-attribute

_IN_TEXT_CITATION_TOKEN_RE = re.compile('\\[(?:E\\s?|Chunk\\s?)?(\\d{1,2}(?:\\s*[,;]\\s*\\d{1,2})*)\\]|【(\\d{1,2}(?:\\s*[,;]\\s*\\d{1,2})*)】')

_MD_IMAGE_RE module-attribute

_MD_IMAGE_RE = re.compile('!\\[[^\\]]*\\]\\([^)]*\\)')

_MD_LINK_DEF_RE module-attribute

_MD_LINK_DEF_RE = re.compile('^\\s*\\[\\d+\\]:\\s*\\S+.*$', re.MULTILINE)

_STRUCTURAL_BOUNDARY_RES module-attribute

_STRUCTURAL_BOUNDARY_RES = [re.compile('\\n\\s*\\n'), re.compile('\\n\\s*[-*+]\\s+'), re.compile('\\n\\s*\\d+\\.\\s+'), re.compile('(?:^|\\n)\\s*#{1,6}\\s+[^\\n]*\\n+')]

__all__ module-attribute

__all__ = ['DEFAULT_COMPOUND_CITATION_INSTRUCTIONS', 'DEFAULT_SINGLE_CITATION_INSTRUCTIONS', 'MAX_CLAIM_CHARS', 'MAX_SOURCE_TEXT_CHARS', 'extract_claim_occurrences', 'extract_claims_for_citations', 'prepare_source_text', 'verify_citations_support']

logger module-attribute

logger = logging.getLogger(__name__)

_check_support_local_laya

_check_support_local_laya(claim, source_text, model='laya', local_url=DEFAULT_LOCAL_URL, timeout=DEFAULT_TIMEOUT_SEC, skip_health_probe=False, is_compound=False)

Query local Laya endpoint with fail-open health probe and error handling.

Source code in zocrates/agents/library_search/citation_verifier.py

def _check_support_local_laya(
    claim: str,
    source_text: str,
    model: str = "laya",
    local_url: str = DEFAULT_LOCAL_URL,
    timeout: float = DEFAULT_TIMEOUT_SEC,
    skip_health_probe: bool = False,
    is_compound: bool = False,
) -> float | None:
    """Query local Laya endpoint with fail-open health probe and error handling."""
    parsed = urllib.parse.urlparse(local_url)
    health_url = f"{parsed.scheme}://{parsed.netloc}/health"

    if not skip_health_probe and not _probe_local_laya_health(health_url):
        logger.warning("Local Laya health probe failed (%s). Failing open.", health_url)
        return None

    effective_timeout = max(timeout, 10.0)

    instructions = (
        DEFAULT_COMPOUND_CITATION_INSTRUCTIONS
        if is_compound
        else DEFAULT_SINGLE_CITATION_INSTRUCTIONS
    )
    payload = {
        "model": model,
        "state": {
            "source_text": source_text,
            "claim": claim,
        },
        "questions": {
            "support": {
                "type": "noul",
                "instructions": instructions,
            }
        },
    }
    headers = {"Content-Type": "application/json"}

    try:
        resp = requests.post(local_url, headers=headers, json=payload, timeout=effective_timeout)
        if resp.status_code == 429:
            time.sleep(0.250)
            resp = requests.post(local_url, headers=headers, json=payload, timeout=effective_timeout)

        if resp.status_code != 200:
            if resp.status_code == 500:
                logger.warning(
                    "Local Laya returned HTTP 500: %s. This typically occurs when Laya attempts to run on an incompatible GPU or runs out of VRAM. Ensure Laya is started with LAYA_DEVICE=cpu (e.g. via Zocrates Preferences). Failing open.",
                    resp.text,
                )
            else:
                logger.warning("Local Laya returned HTTP %s. Failing open.", resp.status_code)
            return None

        res_data = resp.json()
        answers = res_data.get("answers", {})
        support_ans = answers.get("support", {})
        noul_val = support_ans.get("noul")

        if noul_val is not None:
            try:
                val = float(noul_val)
                if math.isfinite(val):
                    return min(1.0, max(0.0, val))
            except (ValueError, TypeError):
                pass

        logger.warning("Malformed 200 response from Local Laya (missing or non-numeric noul). Failing open.")
        return None

    except requests.Timeout:
        logger.warning("Local Laya request timed out (%.1fs). Failing open.", effective_timeout)
        return None
    except Exception as exc:
        logger.warning("Local Laya request failed: %s. Failing open.", exc)
        return None

_check_support_local_laya_batch

_check_support_local_laya_batch(items, local_url=DEFAULT_LOCAL_URL, timeout=15.0, is_compound=False)

Evaluate multiple citation claims concurrently via Laya's native /v1/systemone/batch endpoint.

Supports 3-tuples (chunk_idx, claim, source_text), 4-tuples (chunk_idx, occ_idx, claim, source_text), and 5-tuples (chunk_idx, occ_idx, claim, source_text, is_compound).

Source code in zocrates/agents/library_search/citation_verifier.py

def _check_support_local_laya_batch(
    items: list[tuple[Any, ...]],
    local_url: str = DEFAULT_LOCAL_URL,
    timeout: float = 15.0,
    is_compound: bool = False,
) -> dict[Any, float | None] | None:
    """Evaluate multiple citation claims concurrently via Laya's native /v1/systemone/batch endpoint.

    Supports 3-tuples (chunk_idx, claim, source_text), 4-tuples
    (chunk_idx, occ_idx, claim, source_text), and 5-tuples
    (chunk_idx, occ_idx, claim, source_text, is_compound).
    """
    if not items:
        return {}

    # If items are 5-tuples containing is_compound flags, partition into sub-batches
    if len(items[0]) >= 5:
        single_items = [it[:4] for it in items if not it[4]]
        compound_items = [it[:4] for it in items if it[4]]
        out: dict[Any, float | None] = {}
        if single_items:
            res_s = _check_support_local_laya_batch(
                single_items, local_url=local_url, timeout=timeout, is_compound=False
            )
            if res_s is not None:
                out.update(res_s)
        if compound_items:
            res_c = _check_support_local_laya_batch(
                compound_items, local_url=local_url, timeout=timeout, is_compound=True
            )
            if res_c is not None:
                out.update(res_c)
        return out if out or not (single_items or compound_items) else None

    parsed = urllib.parse.urlparse(local_url)
    batch_url = f"{parsed.scheme}://{parsed.netloc}/v1/systemone/batch"
    states = []
    for item in items:
        if len(item) >= 4:
            states.append({"source_text": item[3], "claim": item[2]})
        else:
            states.append({"source_text": item[2], "claim": item[1]})

    instructions = (
        DEFAULT_COMPOUND_CITATION_INSTRUCTIONS
        if is_compound
        else DEFAULT_SINGLE_CITATION_INSTRUCTIONS
    )
    payload = {
        "states": states,
        "questions": {
            "support": {
                "type": "noul",
                "instructions": instructions,
            }
        },
    }
    headers = {"Content-Type": "application/json"}
    try:
        resp = requests.post(batch_url, headers=headers, json=payload, timeout=max(timeout, 15.0))
        if resp.status_code == 200:
            res_data = resp.json()
            raw_results = res_data.get("results", [])
            out: dict[Any, float | None] = {}
            for item, row in zip(items, raw_results):
                key = (item[0], item[1]) if len(item) == 4 else item[0]
                ans = row.get("answers", {}).get("support", {})
                noul = ans.get("noul")
                if noul is not None:
                    try:
                        val = float(noul)
                        if math.isfinite(val):
                            out[key] = min(1.0, max(0.0, val))
                            continue
                    except (ValueError, TypeError):
                        pass
                out[key] = None
            return out
        logger.warning("Local Laya batch endpoint returned HTTP %s. Failing open.", resp.status_code)
        return None
    except Exception as exc:
        logger.warning("Local Laya batch endpoint query failed: %s. Failing open.", exc)
        return None

_check_support_openrouter

_check_support_openrouter(claim, source_text, model=DEFAULT_OPENROUTER_MODEL, api_key=None, timeout=DEFAULT_TIMEOUT_SEC, is_compound=False)

Query OpenRouter Decisions API with fallback and fail-open handling.

Source code in zocrates/agents/library_search/citation_verifier.py

def _check_support_openrouter(
    claim: str,
    source_text: str,
    model: str = DEFAULT_OPENROUTER_MODEL,
    api_key: str | None = None,
    timeout: float = DEFAULT_TIMEOUT_SEC,
    is_compound: bool = False,
) -> float | None:
    """Query OpenRouter Decisions API with fallback and fail-open handling."""
    effective_key = (
        api_key
        or get_secure_api_key(
            "openrouter_api_key", ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY")
        )
        or os.environ.get("OPENROUTER_API_KEY")
        or os.environ.get("OPEN_ROUTER_API_KEY")
    )
    if not effective_key:
        logger.warning("No OpenRouter API key found for citation verification. Failing open.")
        return None

    headers = {
        "Authorization": f"Bearer {effective_key}",
        "Content-Type": "application/json",
        "HTTP-Referer": "https://github.com/alexplaka/Zocrates",
        "X-OpenRouter-Title": "Zocrates Citation Verifier",
    }

    model_lower = model.lower()
    if "respan" in model_lower:
        instructions = (
            DEFAULT_RESPAN_COMPOUND_INSTRUCTIONS
            if is_compound
            else DEFAULT_RESPAN_SINGLE_INSTRUCTIONS
        )
        payload = {
            "model": model,
            "state": f"Source text: {source_text}\n\nClaim: {claim}",
            "questions": {
                "support": {
                    "type": "noul",
                    "instructions": instructions,
                }
            },
        }
    else:
        instructions = (
            DEFAULT_COMPOUND_CITATION_INSTRUCTIONS
            if is_compound
            else DEFAULT_SINGLE_CITATION_INSTRUCTIONS
        )
        payload = {
            "model": model,
            "state": {
                "source_text": source_text,
                "claim": claim,
            },
            "questions": {
                "support": {
                    "type": "noul",
                    "instructions": instructions,
                }
            },
        }

    try:
        resp = requests.post(
            OPENROUTER_DECISIONS_URL,
            headers=headers,
            json=payload,
            timeout=timeout,
        )

        # Single retry on HTTP 429 with 250ms backoff
        if resp.status_code == 429:
            time.sleep(0.250)
            resp = requests.post(
                OPENROUTER_DECISIONS_URL,
                headers=headers,
                json=payload,
                timeout=timeout,
            )

        # Fallback to chat completions if 404 on alpha decisions endpoint
        if resp.status_code == 404:
            sys_msg = (
                "You are a citation verification model. This claim cites multiple sources. "
                "Determine whether the source text supports a substantive part of the claim. "
                "Respond with JSON in the form: {\"noul\": <probability between 0.0 and 1.0>}."
                if is_compound
                else (
                    "You are a citation verification model. Determine whether the source text supports the claim. "
                    "Respond with JSON in the form: {\"noul\": <probability between 0.0 and 1.0>}."
                )
            )
            chat_payload = {
                "model": model,
                "messages": [
                    {
                        "role": "system",
                        "content": sys_msg,
                    },
                    {
                        "role": "user",
                        "content": f"Source text:\n{source_text}\n\nClaim:\n{claim}",
                    },
                ],
                "temperature": 0.0,
                "max_tokens": 50,
            }
            chat_resp = requests.post(
                OPENROUTER_CHAT_URL,
                headers=headers,
                json=chat_payload,
                timeout=timeout,
            )
            if chat_resp.status_code == 200:
                try:
                    cdata = chat_resp.json()
                    content = cdata["choices"][0]["message"]["content"]
                    m = re.search(r'"noul"\s*:\s*([0-9]*\.?[0-9]+)', content)
                    if m:
                        val = float(m.group(1))
                        if math.isfinite(val):
                            return min(1.0, max(0.0, val))
                except Exception:
                    pass
            logger.warning("OpenRouter chat completions fallback failed. Failing open.")
            return None

        if resp.status_code != 200:
            logger.warning("OpenRouter decisions API returned HTTP %s. Failing open.", resp.status_code)
            return None

        res_data = resp.json()
        answers = res_data.get("answers", {})
        support_ans = answers.get("support", {})
        noul_val = support_ans.get("noul")

        if noul_val is not None:
            try:
                val = float(noul_val)
                if math.isfinite(val):
                    return min(1.0, max(0.0, val))
            except (ValueError, TypeError):
                pass

        logger.warning("Malformed 200 response from OpenRouter decision model (missing or non-numeric noul). Failing open.")
        return None

    except requests.Timeout:
        logger.warning("OpenRouter decision model timed out (%.1fs). Failing open.", timeout)
        return None
    except Exception as exc:
        logger.warning("OpenRouter decision model call failed: %s. Failing open.", exc)
        return None

_clean_and_join_spans

_clean_and_join_spans(backward, forward, total_chunks=0)

Join backward and forward spans into a clean, complete claim.

Source code in zocrates/agents/library_search/citation_verifier.py

def _clean_and_join_spans(backward: str, forward: str, total_chunks: int = 0) -> str:
    """Join backward and forward spans into a clean, complete claim."""
    backward = backward.strip()
    forward = forward.strip()

    if not backward:
        claim = forward
    elif not forward:
        claim = backward
    else:
        if forward.startswith((",", ";", ":", ".")):
            claim = f"{backward}{forward}"
        else:
            claim = f"{backward} {forward}"

    # Normalize symbols: ≈ -> "about", ≥ -> "at least", ≤ -> "at most", → -> "to", — -> " - "
    claim = claim.replace("≈", "about")
    claim = claim.replace("≥", "at least")
    claim = claim.replace("≤", "at most")
    claim = claim.replace("→", "to")
    claim = claim.replace("—", " - ")

    # Strip inline bold/italic markers **, __, and backticks anywhere in the claim
    claim = claim.replace("**", "").replace("__", "").replace("`", "")

    claim = re.sub(r"\s+", " ", claim).strip()

    def _strip_valid_citation(m: re.Match) -> str:
        if total_chunks <= 0:
            return ""
        raw_content = m.group(1) if m.group(1) is not None else m.group(2)
        if not raw_content:
            return ""
        parts = [p.strip() for p in re.split(r"[,;]", raw_content) if p.strip()]
        if any(p.isascii() and p.isdigit() and 1 <= int(p) <= total_chunks for p in parts):
            return ""
        return m.group(0)

    claim = _IN_TEXT_CITATION_TOKEN_RE.sub(_strip_valid_citation, claim)
    claim = re.sub(r"\(\s*\)", "", claim)
    claim = re.sub(r"\[\s*\]", "", claim)
    claim = re.sub(r"\s+", " ", claim).strip()
    claim = re.sub(r"^(?:[\.\,\;\:\!\?\*\-\#\>\)\}\]\s]|\d+\.\s*)+", "", claim).strip()

    if claim and not claim.endswith((".", "!", "?")):
        claim = claim + "."

    return claim

_find_backward_claim_boundary

_find_backward_claim_boundary(prefix, total_chunks=0, max_chars=MAX_CLAIM_CHARS)

Find start index of the paragraph/block preceding a citation token.

Step 1: Scans prefix for earlier valid citation tokens (1 <= idx <= total_chunks); if found, min_boundary is the end of the last citation match so citations in the same block do not bleed. Out-of-range author references (e.g. [44, 69]) are ignored. Step 2: In search_region = prefix[min_boundary:], searches for the latest structural boundary (paragraph breaks, list bullets, numbered items, or markdown headers). Step 3: Bounds backward claim length to max_chars, advancing past the first sentence boundary in the window to avoid chopping mid-sentence.

Source code in zocrates/agents/library_search/citation_verifier.py

def _find_backward_claim_boundary(
    prefix: str,
    total_chunks: int = 0,
    max_chars: int = MAX_CLAIM_CHARS,
) -> int:
    """Find start index of the paragraph/block preceding a citation token.

    Step 1: Scans prefix for earlier valid citation tokens (1 <= idx <= total_chunks);
    if found, min_boundary is the end of the last citation match so citations in the
    same block do not bleed. Out-of-range author references (e.g. [44, 69]) are ignored.
    Step 2: In search_region = prefix[min_boundary:], searches for the latest structural
    boundary (paragraph breaks, list bullets, numbered items, or markdown headers).
    Step 3: Bounds backward claim length to max_chars, advancing past the first
    sentence boundary in the window to avoid chopping mid-sentence.
    """
    n = len(prefix)
    if n == 0:
        return 0

    # Step 1: Find the last valid in-text citation match in prefix
    min_boundary = 0
    if total_chunks > 0:
        for m in _IN_TEXT_CITATION_TOKEN_RE.finditer(prefix):
            raw_content = m.group(1) if m.group(1) is not None else m.group(2)
            if not raw_content:
                continue
            parts = [p.strip() for p in re.split(r"[,;]", raw_content) if p.strip()]
            if any(p.isascii() and p.isdigit() and 1 <= int(p) <= total_chunks for p in parts):
                min_boundary = m.end()

    # Step 2: Search for the latest structural boundary end in search_region
    search_region = prefix[min_boundary:]
    all_matches = []
    for pat in _STRUCTURAL_BOUNDARY_RES:
        all_matches.extend(pat.finditer(search_region))

    if all_matches:
        start_idx = min_boundary + max(m.end() for m in all_matches)
    else:
        start_idx = min_boundary

    # Step 3: Bound claim length if exceeding max_chars
    if (n - start_idx) > max_chars:
        start_idx = n - max_chars
        fwd_offset = _find_forward_sentence_boundary(prefix[start_idx:])
        if fwd_offset > 0 and (n - (start_idx + fwd_offset)) >= MIN_CLAIM_CHARS:
            start_idx += fwd_offset

    return start_idx

_find_backward_sentence_boundary

_find_backward_sentence_boundary(prefix)

Find start index of the sentence preceding a citation token.

Rejects candidate periods preceded by abbreviations or followed by digits. Treats ** or __ following sentence-ending punctuation (. ! ?) as a boundary so lead-in headings like - Lead-in. Claim yield Claim without trailing syntax. Returns the character index in prefix where the sentence starts.

Source code in zocrates/agents/library_search/citation_verifier.py

def _find_backward_sentence_boundary(prefix: str) -> int:
    """Find start index of the sentence preceding a citation token.

    Rejects candidate periods preceded by abbreviations or followed by digits.
    Treats ** or __ following sentence-ending punctuation (. ! ?) as a boundary
    so lead-in headings like - **Lead-in.** Claim yield Claim without trailing syntax.
    Returns the character index in prefix where the sentence starts.
    """
    n = len(prefix)
    if n == 0:
        return 0

    idx = n - 1
    while idx >= 0:
        ch = prefix[idx]
        if ch in ("\n", "\r"):
            return idx + 1
        elif ch in ("!", "?"):
            # Check for trailing bold/italic markers: e.g. !** or ?__
            if idx + 2 < n and prefix[idx + 1:idx + 3] in ("**", "__"):
                if idx + 3 == n or prefix[idx + 3].isspace():
                    return idx + 3
            if idx + 1 == n or prefix[idx + 1].isspace():
                return idx + 1
        elif ch == ".":
            # Check if followed by ** or __ (e.g. - **Lead-in.** Claim)
            if idx + 2 < n and prefix[idx + 1:idx + 3] in ("**", "__"):
                if idx + 3 == n or prefix[idx + 3].isspace():
                    before_period = prefix[:idx].rstrip()
                    if not _ABBREVIATION_RE.search(before_period):
                        return idx + 3
            is_followed_by_space = (idx + 1 == n) or prefix[idx + 1].isspace()
            is_followed_by_digit = (idx + 1 < n) and prefix[idx + 1].isdigit()
            if not is_followed_by_digit and is_followed_by_space:
                before_period = prefix[:idx].rstrip()
                if not _ABBREVIATION_RE.search(before_period):
                    return idx + 1
        idx -= 1

    return 0

_find_forward_sentence_boundary

_find_forward_sentence_boundary(suffix)

Find end index of the clause/sentence following a mid-sentence citation token.

Rejects candidate periods preceded by abbreviations or followed by digits. Stops at sentence boundaries, newlines, semicolons, or the next citation token.

Source code in zocrates/agents/library_search/citation_verifier.py

def _find_forward_sentence_boundary(suffix: str) -> int:
    """Find end index of the clause/sentence following a mid-sentence citation token.

    Rejects candidate periods preceded by abbreviations or followed by digits.
    Stops at sentence boundaries, newlines, semicolons, or the next citation token.
    """
    n = len(suffix)
    if n == 0:
        return 0

    cit_match = _IN_TEXT_CITATION_TOKEN_RE.search(suffix)
    max_search = cit_match.start() if cit_match else n

    idx = 0
    while idx < max_search:
        ch = suffix[idx]
        if ch in ("\n", "\r"):
            return idx
        elif ch in ("!", "?"):
            return idx + 1
        elif ch == ";":
            return idx
        elif ch == ".":
            is_followed_by_space = (idx + 1 == n) or suffix[idx + 1].isspace()
            is_followed_by_digit = (idx + 1 < n) and suffix[idx + 1].isdigit()
            if not is_followed_by_digit and is_followed_by_space:
                before_period = suffix[:idx].rstrip()
                if not _ABBREVIATION_RE.search(before_period):
                    return idx + 1
        idx += 1

    return max_search

_probe_local_laya_health

_probe_local_laya_health(health_url, timeout=DEFAULT_HEALTH_TIMEOUT_SEC)

Fast health check probe against local Laya server.

Source code in zocrates/agents/library_search/citation_verifier.py

def _probe_local_laya_health(health_url: str, timeout: float = DEFAULT_HEALTH_TIMEOUT_SEC) -> bool:
    """Fast health check probe against local Laya server."""
    try:
        resp = requests.get(health_url, timeout=timeout)
        return resp.status_code == 200
    except Exception:
        return False

_safe_set

_safe_set(span, key, value)

Safely set an attribute on an active OpenTelemetry span, no-oping on error or None.

Source code in zocrates/agents/library_search/citation_verifier.py

def _safe_set(span: Any, key: str, value: Any) -> None:
    """Safely set an attribute on an active OpenTelemetry span, no-oping on error or None."""
    if span is None:
        return
    try:
        if hasattr(span, "set_attribute"):
            if isinstance(value, (dict, list)):
                span.set_attribute(key, json.dumps(value))
            else:
                span.set_attribute(key, value)
        elif hasattr(span, "_attributes") and isinstance(span._attributes, dict):
            span._attributes[key] = value
    except Exception as exc:
        logger.debug("Failed setting span attribute %s: %s", key, exc)

_verification_span

_verification_span(name)

Context manager yielding an active OpenTelemetry span or None on fallback.

Parameters:

Name	Type	Description	Default
name	str	Name of the OpenTelemetry span.	required

Yields:

Type	Description
Any	Active span if tracing is available, otherwise None.

Source code in zocrates/agents/library_search/citation_verifier.py

@contextlib.contextmanager
def _verification_span(name: str) -> Iterator[Any]:
    """Context manager yielding an active OpenTelemetry span or None on fallback.

    Args:
        name: Name of the OpenTelemetry span.

    Yields:
        Active span if tracing is available, otherwise None.
    """
    if _otel_trace is None:
        yield None
        return

    span_cm = None
    try:
        tracer = _otel_trace.get_tracer("zocrates.citation_verifier")
        span_cm = tracer.start_as_current_span(name)
    except Exception as exc:
        logger.debug("Failed starting verification span: %s", exc)
        yield None
        return

    span = None
    if hasattr(span_cm, "__enter__") and hasattr(span_cm, "__exit__"):
        try:
            with span_cm as s:
                span = s
                yield span
        except Exception as exc:
            if span is not None and _StatusCode is not None:
                try:
                    span.set_status(_StatusCode.ERROR, str(exc))
                except Exception:
                    pass
            raise
    else:
        yield span_cm

extract_claim_occurrences

extract_claim_occurrences(text, total_chunks)

Extract claim text for each occurrence of cited chunk indices in document order.

Parameters:

Name	Type	Description	Default
text	str	Synthesized markdown text containing citation tokens.	required
total_chunks	int	Number of available retrieved chunks for bounds validation.	required

Returns:

Type	Description
list[dict[str, Any]]	List of dicts with keys 'chunk_index', 'occurrence', and 'claim', where
list[dict[str, Any]]	occurrence is the 1-based occurrence index for that specific chunk_index.

Source code in zocrates/agents/library_search/citation_verifier.py

def extract_claim_occurrences(
    text: str,
    total_chunks: int,
) -> list[dict[str, Any]]:
    """Extract claim text for each occurrence of cited chunk indices in document order.

    Args:
        text: Synthesized markdown text containing citation tokens.
        total_chunks: Number of available retrieved chunks for bounds validation.

    Returns:
        List of dicts with keys 'chunk_index', 'occurrence', and 'claim', where
        occurrence is the 1-based occurrence index for that specific chunk_index.
    """
    if not text or total_chunks <= 0:
        return []

    masked_text, placeholders = mask_text_for_extraction(text)
    occurrences: list[dict[str, Any]] = []
    occ_counters: dict[int, int] = {}

    for m in _IN_TEXT_CITATION_TOKEN_RE.finditer(masked_text):
        raw_content = m.group(1) if m.group(1) is not None else m.group(2)
        if not raw_content:
            continue
        parts = [p.strip() for p in re.split(r"[,;]", raw_content) if p.strip()]
        valid_indices: list[int] = []
        seen_in_token: set[int] = set()
        for p in parts:
            if p.isdigit():
                val = int(p)
                if 1 <= val <= total_chunks and val not in seen_in_token:
                    seen_in_token.add(val)
                    valid_indices.append(val)

        if not valid_indices:
            continue

        token_start = m.start()
        token_end = m.end()

        # Backward scan
        prefix = masked_text[:token_start]
        b_idx = _find_backward_claim_boundary(prefix, total_chunks=total_chunks)
        backward_span = prefix[b_idx:].strip()

        # Forward scan check
        forward_span = ""
        if len(backward_span) < MIN_CLAIM_CHARS or _AUTHOR_ONLY_RE.match(backward_span):
            suffix = masked_text[token_end:]
            f_idx = _find_forward_sentence_boundary(suffix)
            forward_span = suffix[:f_idx].strip()

        claim = _clean_and_join_spans(backward_span, forward_span, total_chunks=total_chunks)
        if len(claim) < 15:
            start_w = max(0, token_start - 100)
            end_w = min(len(masked_text), token_end + 100)
            fallback_span = masked_text[start_w:end_w].strip()
            claim = _clean_and_join_spans(fallback_span, "", total_chunks=total_chunks)

        claim = unmask_text(claim, placeholders)

        for idx in valid_indices:
            occ = occ_counters.get(idx, 0) + 1
            occ_counters[idx] = occ
            occurrences.append({
                "chunk_index": idx,
                "occurrence": occ,
                "claim": claim,
                "is_compound": len(valid_indices) > 1,
            })

    return occurrences

extract_claims_for_citations

extract_claims_for_citations(text, total_chunks)

Extract claim text for each cited chunk index in the synthesis text.

Maintained for backward compatibility; selects the longest claim for each chunk.

Parameters:

Name	Type	Description	Default
text	str	Synthesized markdown text containing citation tokens.	required
total_chunks	int	Number of available retrieved chunks for bounds validation.	required

Returns:

Type	Description
dict[int, str]	Dict mapping chunk index (1-based int) to the extracted claim text.

Source code in zocrates/agents/library_search/citation_verifier.py

def extract_claims_for_citations(
    text: str,
    total_chunks: int,
) -> dict[int, str]:
    """Extract claim text for each cited chunk index in the synthesis text.

    Maintained for backward compatibility; selects the longest claim for each chunk.

    Args:
        text: Synthesized markdown text containing citation tokens.
        total_chunks: Number of available retrieved chunks for bounds validation.

    Returns:
        Dict mapping chunk index (1-based int) to the extracted claim text.
    """
    occurrences = extract_claim_occurrences(text, total_chunks)
    claims_by_index: dict[int, str] = {}
    for item in occurrences:
        idx = item["chunk_index"]
        claim = item["claim"]
        if idx not in claims_by_index or len(claim) > len(claims_by_index[idx]):
            claims_by_index[idx] = claim
    return claims_by_index

get_secure_api_key

get_secure_api_key(key_name='openrouter_api_key', env_vars=('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY'))

Load a named API key from in-memory session environment variables.

Zotero Preferences / Login Manager serves as the secure persistent vault. Backend keys are in-memory session variables (os.environ) synced from Zotero.

Parameters:

Name	Type	Description	Default
key_name	str	Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').	'openrouter_api_key'
env_vars	list[str] | tuple[str, ...] | None	Environment variable names to check.	('OPENROUTER_API_KEY', 'OPEN_ROUTER_API_KEY')

Returns:

Type	Description
str | None	The API key string or None if not found.

Source code in zocrates/config.py

def get_secure_api_key(
    key_name: str = "openrouter_api_key",
    env_vars: list[str] | tuple[str, ...] | None = ("OPENROUTER_API_KEY", "OPEN_ROUTER_API_KEY"),
) -> str | None:
    """Load a named API key from in-memory session environment variables.

    Zotero Preferences / Login Manager serves as the secure persistent vault.
    Backend keys are in-memory session variables (os.environ) synced from Zotero.

    Args:
        key_name: Entry identifier (e.g., 'openrouter_api_key', 'openai_api_key').
        env_vars: Environment variable names to check.

    Returns:
        The API key string or None if not found.
    """
    if env_vars:
        for var in env_vars:
            val = os.getenv(var)
            if val:
                return val
    return None

load_zocrates_config

load_zocrates_config(zotero_data_dir=None)

Load the Zocrates config from the data directory.

If zotero_data_dir is None, attempts to obtain it from the active LibrarySyncer service or the ZOTERO_DATA_DIR environment variable.

Returns config merged with standard defaults.

Parameters:

Name	Type	Description	Default
zotero_data_dir	Path | str | None	Root Zotero data directory.	None

Returns:

Type	Description
dict	Parsed JSON config dict merged with defaults.

Source code in zocrates/config.py

def load_zocrates_config(zotero_data_dir: Path | str | None = None) -> dict:
    """Load the Zocrates config from the data directory.

    If ``zotero_data_dir`` is ``None``, attempts to obtain it from the active
    ``LibrarySyncer`` service or the ``ZOTERO_DATA_DIR`` environment variable.

    Returns config merged with standard defaults.

    Args:
        zotero_data_dir: Root Zotero data directory.

    Returns:
        Parsed JSON config dict merged with defaults.
    """
    defaults = copy.deepcopy(DEFAULT_CONFIG)

    resolved_dir = resolve_zotero_data_dir(zotero_data_dir)
    if resolved_dir is None:
        return defaults

    config_path = get_config_path(resolved_dir)
    resolved_path = config_path.resolve()
    if not config_path.exists():
        with _CONFIG_CACHE_LOCK:
            _CONFIG_CACHE.pop(resolved_path, None)
            _CONFIG_CACHE_FINGERPRINT.pop(resolved_path, None)
        return defaults

    try:
        stat_res = config_path.stat()
        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
    except OSError:
        return defaults

    with _CONFIG_CACHE_LOCK:
        if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
            return copy.deepcopy(_CONFIG_CACHE[resolved_path])

    if config_path.exists():
        try:
            lock_path = str(config_path) + ".lock"
            with FileLock(lock_path):
                try:
                    stat_res = config_path.stat()
                    current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                except OSError:
                    return defaults

                with _CONFIG_CACHE_LOCK:
                    if resolved_path in _CONFIG_CACHE and _CONFIG_CACHE_FINGERPRINT.get(resolved_path) == current_fingerprint:
                        return copy.deepcopy(_CONFIG_CACHE[resolved_path])

                with open(config_path) as f:
                    loaded = json.load(f)

                # Backfill override_agent_models if missing or incomplete
                needs_save = False

                if "override_agent_models" not in loaded or not isinstance(loaded["override_agent_models"], dict):
                    loaded["override_agent_models"] = copy.deepcopy(DEFAULT_OVERRIDE_AGENT_MODELS)
                    needs_save = True
                else:
                    for role, default_role_cfg in DEFAULT_OVERRIDE_AGENT_MODELS.items():
                        if role not in loaded["override_agent_models"]:
                            loaded["override_agent_models"][role] = copy.deepcopy(default_role_cfg)
                            needs_save = True

                # Backfill search_harness if missing or incomplete in existing config file on disk
                if "search_harness" not in loaded or not isinstance(loaded["search_harness"], dict):
                    loaded["search_harness"] = copy.deepcopy(DEFAULT_SEARCH_HARNESS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_SEARCH_HARNESS.items():
                        if k not in loaded["search_harness"]:
                            loaded["search_harness"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill graph_relations if missing or incomplete in existing config file on disk
                if "graph_relations" not in loaded or not isinstance(loaded["graph_relations"], dict):
                    loaded["graph_relations"] = copy.deepcopy(DEFAULT_GRAPH_RELATIONS)
                    needs_save = True
                else:
                    for k, default_val in DEFAULT_GRAPH_RELATIONS.items():
                        if k not in loaded["graph_relations"]:
                            loaded["graph_relations"][k] = copy.deepcopy(default_val)
                            needs_save = True

                # Backfill llm_num_retries if missing from existing config file on disk
                if "llm_num_retries" not in loaded:
                    loaded["llm_num_retries"] = DEFAULT_LLM_NUM_RETRIES
                    needs_save = True

                # Backfill llm_base_url if missing from existing config file on disk
                if "llm_base_url" not in loaded:
                    loaded["llm_base_url"] = DEFAULT_LLM_BASE_URL
                    needs_save = True

                # Backfill citations options if missing from existing config file on disk
                if "citations_enabled" not in loaded:
                    loaded["citations_enabled"] = DEFAULT_CITATIONS_ENABLED
                    needs_save = True
                if "references_section_enabled" not in loaded:
                    loaded["references_section_enabled"] = DEFAULT_REFERENCES_SECTION_ENABLED
                    needs_save = True
                if "considered_sources_disclosure" not in loaded:
                    loaded["considered_sources_disclosure"] = DEFAULT_CONSIDERED_SOURCES_DISCLOSURE
                    needs_save = True
                if "auto_render_mermaid_diagrams" not in loaded:
                    loaded["auto_render_mermaid_diagrams"] = DEFAULT_AUTO_RENDER_MERMAID_DIAGRAMS
                    needs_save = True
                if "enable_mcp_server" not in loaded:
                    loaded["enable_mcp_server"] = DEFAULT_ENABLE_MCP_SERVER
                    needs_save = True
                if "auto_spawn_backend" not in loaded:
                    loaded["auto_spawn_backend"] = DEFAULT_AUTO_SPAWN_BACKEND
                    needs_save = True
                if "enable_predictive_ghost_input" not in loaded:
                    loaded["enable_predictive_ghost_input"] = DEFAULT_ENABLE_PREDICTIVE_GHOST_INPUT
                    needs_save = True
                if "predictive_prompt_model" not in loaded:
                    loaded["predictive_prompt_model"] = DEFAULT_PREDICTIVE_PROMPT_MODEL
                    needs_save = True
                if "sentence_attribution_model" not in loaded:
                    loaded["sentence_attribution_model"] = DEFAULT_SENTENCE_ATTRIBUTION_MODEL
                    needs_save = True
                if "scholar_polite_email" not in loaded:
                    loaded["scholar_polite_email"] = DEFAULT_SCHOLAR_POLITE_EMAIL
                    needs_save = True
                if "conversion_timeout_seconds" not in loaded:
                    loaded["conversion_timeout_seconds"] = DEFAULT_CONVERSION_TIMEOUT_SECONDS
                    needs_save = True
                if "datalab_merge_cross_page" not in loaded:
                    loaded["datalab_merge_cross_page"] = DEFAULT_DATALAB_MERGE_CROSS_PAGE
                    needs_save = True
                if "user_research_profile" not in loaded:
                    loaded["user_research_profile"] = DEFAULT_USER_RESEARCH_PROFILE
                    needs_save = True
                if "max_research_profile_chars" not in loaded:
                    loaded["max_research_profile_chars"] = DEFAULT_MAX_RESEARCH_PROFILE_CHARS
                    needs_save = True
                if "max_saved_sessions" not in loaded:
                    loaded["max_saved_sessions"] = DEFAULT_MAX_SAVED_SESSIONS
                    needs_save = True
                if "session_retention_days" not in loaded:
                    loaded["session_retention_days"] = DEFAULT_SESSION_RETENTION_DAYS
                    needs_save = True
                if "session_db_auto_vacuum_on_startup" not in loaded:
                    loaded["session_db_auto_vacuum_on_startup"] = DEFAULT_SESSION_DB_AUTO_VACUUM_ON_STARTUP
                    needs_save = True
                if "prune_empty_sessions" not in loaded:
                    loaded["prune_empty_sessions"] = DEFAULT_PRUNE_EMPTY_SESSIONS
                    needs_save = True
                if "backend_port" not in loaded:
                    loaded["backend_port"] = DEFAULT_BACKEND_PORT
                    needs_save = True

                # Backfill Phase 7 onboarding SSOT keys if missing from existing
                # config files on disk (embedding_dimension / sync_auto_abort_failures)
                if "embedding_dimension" not in loaded:
                    loaded["embedding_dimension"] = DEFAULT_CONFIG["embedding_dimension"]
                    needs_save = True
                if "sync_auto_abort_failures" not in loaded:
                    loaded["sync_auto_abort_failures"] = DEFAULT_SYNC_AUTO_ABORT_FAILURES
                    needs_save = True

                if "vlm_captioning_backup_model" not in loaded:
                    loaded["vlm_captioning_backup_model"] = DEFAULT_CONFIG["vlm_captioning_backup_model"]
                    needs_save = True
                if "vlm_captioning_model" not in loaded:
                    loaded["vlm_captioning_model"] = DEFAULT_CONFIG["vlm_captioning_model"]
                    needs_save = True
                if "vlm_captioning_provider" not in loaded:
                    loaded["vlm_captioning_provider"] = DEFAULT_CONFIG["vlm_captioning_provider"]
                    needs_save = True
                if "enable_vlm_captioning" not in loaded:
                    loaded["enable_vlm_captioning"] = DEFAULT_CONFIG["enable_vlm_captioning"]
                    needs_save = True
                if "vlm_max_images_per_doc" not in loaded:
                    loaded["vlm_max_images_per_doc"] = DEFAULT_CONFIG["vlm_max_images_per_doc"]
                    needs_save = True

                # Detect and migrate deprecated compaction keys from disk
                if "compaction_token_threshold" in loaded and "adaptive_compaction_token_ceiling" not in loaded:
                    loaded["adaptive_compaction_token_ceiling"] = loaded["compaction_token_threshold"]
                    needs_save = True
                if "enable_context_compaction" in loaded and "enable_adaptive_compaction" not in loaded:
                    loaded["enable_adaptive_compaction"] = loaded["enable_context_compaction"]
                    needs_save = True
                if "adaptive_decision_model" in loaded and "adaptive_compaction_decision_model" not in loaded:
                    loaded["adaptive_compaction_decision_model"] = loaded.pop("adaptive_decision_model")
                    needs_save = True
                elif "adaptive_decision_model" in loaded:
                    loaded.pop("adaptive_decision_model", None)
                    needs_save = True

                # Purge all 5 deprecated compaction keys
                deprecated_compaction_keys = (
                    "compaction_interval",
                    "compaction_overlap_size",
                    "compaction_token_threshold",
                    "compaction_event_retention_size",
                    "enable_context_compaction",
                )
                for dep_key in deprecated_compaction_keys:
                    if dep_key in loaded:
                        loaded.pop(dep_key, None)
                        needs_save = True

                # Backfill missing adaptive compaction options
                adaptive_compaction_defaults = {
                    "enable_adaptive_compaction": DEFAULT_ENABLE_ADAPTIVE_COMPACTION,
                    "adaptive_compaction_retention_turns": DEFAULT_ADAPTIVE_COMPACTION_RETENTION_TURNS,
                    "adaptive_compaction_min_turns": DEFAULT_ADAPTIVE_COMPACTION_MIN_TURNS,
                    "adaptive_compaction_token_floor": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_FLOOR,
                    "adaptive_compaction_token_ceiling": DEFAULT_ADAPTIVE_COMPACTION_TOKEN_CEILING,
                    "adaptive_compaction_threshold": DEFAULT_ADAPTIVE_COMPACTION_THRESHOLD,
                    "adaptive_compaction_decision_model": DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL,
                    "compaction_summarizer_model": DEFAULT_COMPACTION_SUMMARIZER_MODEL,
                }
                for k, default_val in adaptive_compaction_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True

                # Disallow local Laya for compaction decisions on disk; reset to default remote model
                if is_local_decision_model(loaded.get("adaptive_compaction_decision_model")):
                    loaded["adaptive_compaction_decision_model"] = DEFAULT_ADAPTIVE_COMPACTION_DECISION_MODEL
                    needs_save = True

                # Backfill citation verification options if missing from existing config file on disk
                verification_defaults = {
                    "citation_verification_enabled": DEFAULT_CITATION_VERIFICATION_ENABLED,
                    "citation_verification_provider": DEFAULT_CITATION_VERIFICATION_PROVIDER,
                    "citation_verification_model": DEFAULT_CITATION_VERIFICATION_MODEL,
                    "citation_verification_threshold": DEFAULT_CITATION_VERIFICATION_THRESHOLD,
                    "citation_verification_local_url": DEFAULT_CITATION_VERIFICATION_LOCAL_URL,
                    "citation_verification_local_port": DEFAULT_CITATION_VERIFICATION_LOCAL_PORT,
                    "citation_verification_auto_start_local": DEFAULT_CITATION_VERIFICATION_AUTO_START_LOCAL,
                    "citation_verification_timeout_sec": DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC,
                    "citation_verification_turn_budget_sec": DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC,
                }
                for k, default_val in verification_defaults.items():
                    if k not in loaded:
                        loaded[k] = default_val
                        needs_save = True
                # Migrate legacy default 3.0s/10.0s timeout and budget to 10.0s/20.0s for CPU inference headroom
                try:
                    if float(loaded.get("citation_verification_timeout_sec", 0)) == 3.0:
                        loaded["citation_verification_timeout_sec"] = DEFAULT_CITATION_VERIFICATION_TIMEOUT_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass
                try:
                    if float(loaded.get("citation_verification_turn_budget_sec", 0)) == 10.0:
                        loaded["citation_verification_turn_budget_sec"] = DEFAULT_CITATION_VERIFICATION_TURN_BUDGET_SEC
                        needs_save = True
                except (ValueError, TypeError):
                    pass

                if needs_save:
                    try:
                        with open(config_path, "w") as f_out:
                            json.dump(loaded, f_out, indent=2)
                        stamp_config_mtime(resolved_dir)
                    except Exception as save_err:
                        logger.warning(
                            "Failed to backfill default dialogue agent models to %s: %s",
                            config_path,
                            save_err,
                        )
                    try:
                        stat_res = config_path.stat()
                        current_fingerprint = (stat_res.st_mtime_ns, stat_res.st_size)
                    except OSError:
                        pass

                res = dict(defaults)
                res.update(loaded)
                with _CONFIG_CACHE_LOCK:
                    _CONFIG_CACHE[resolved_path] = copy.deepcopy(res)
                    _CONFIG_CACHE_FINGERPRINT[resolved_path] = current_fingerprint
                return copy.deepcopy(res)
        except Exception as e:
            logger.warning("Failed to load config from %s: %s", config_path, e)
    return defaults

mask_text_for_extraction

mask_text_for_extraction(text)

Mask code blocks, images, and markdown link reference definitions.

Source code in zocrates/agents/library_search/citation_verifier.py

def mask_text_for_extraction(text: str) -> tuple[str, dict[str, str]]:
    """Mask code blocks, images, and markdown link reference definitions."""
    placeholders: dict[str, str] = {}
    counter = 0

    def _mask(m: re.Match) -> str:
        """Replace a matched span with an extraction mask placeholder.

        Args:
            m: Regular expression match object containing the matched text.

        Returns:
            Placeholder string mapped to the matched content.
        """
        nonlocal counter
        ph = f"@@ZOCRATES_VERIFY_MASK_{counter}@@"
        counter += 1
        placeholders[ph] = m.group(0)
        return ph

    masked = _CODE_BLOCK_RE.sub(_mask, text)
    masked = _MD_IMAGE_RE.sub(_mask, masked)
    masked = _MD_LINK_DEF_RE.sub(_mask, masked)
    return masked, placeholders

prepare_source_text

prepare_source_text(chunk, claim=None, max_chars=MAX_SOURCE_TEXT_CHARS)

Format and bound chunk source text around claim keywords.

Source code in zocrates/agents/library_search/citation_verifier.py

def prepare_source_text(
    chunk: dict[str, Any],
    claim: str | None = None,
    max_chars: int = MAX_SOURCE_TEXT_CHARS,
) -> str:
    """Format and bound chunk source text around claim keywords."""
    raw_text = str(chunk.get("text") or "").strip()
    if len(raw_text) <= max_chars:
        return raw_text

    if claim:
        words = set(re.findall(r"\b[a-zA-Z0-9]{4,}\b", claim.lower()))
        if words:
            step = min(500, max(100, max_chars // 3))
            best_score = -1
            best_pos = 0
            for pos in range(0, len(raw_text) - max_chars + 1, step):
                window = raw_text[pos : pos + max_chars].lower()
                score = sum(1 for w in words if w in window)
                if score > best_score:
                    best_score = score
                    best_pos = pos
            if best_score > 0:
                return raw_text[best_pos : best_pos + max_chars]

    return raw_text[:max_chars]

unmask_text

unmask_text(text, placeholders)

Restore masked spans in extracted claims.

Source code in zocrates/agents/library_search/citation_verifier.py

def unmask_text(text: str, placeholders: dict[str, str]) -> str:
    """Restore masked spans in extracted claims."""
    for ph, orig in placeholders.items():
        text = text.replace(ph, orig)
    return text

verify_citations_support async

verify_citations_support(text, ranked_chunks, config=None, telemetry_out=None)

Verify citations in text against retrieved chunks using decision models.

Parameters:

Name	Type	Description	Default
text	str	Synthesized markdown response text containing citation tokens.	required
ranked_chunks	list[dict[str, Any]]	Complete list of retrieved evidence chunks.	required
config	dict[str, Any] | None	Optional configuration dict from zocrates_config.json.	None
telemetry_out	dict[str, Any] | None	Optional dict to receive verification telemetry data.	None

Returns:

Type	Description
dict[int, list[float | None]]	Dict mapping chunk index (1-based int) to list of support probabilities
dict[int, list[float | None]]	for each occurrence of that chunk index in document order ([p_occ1, p_occ2, ...]).

Source code in zocrates/agents/library_search/citation_verifier.py

async def verify_citations_support(
    text: str,
    ranked_chunks: list[dict[str, Any]],
    config: dict[str, Any] | None = None,
    telemetry_out: dict[str, Any] | None = None,
) -> dict[int, list[float | None]]:
    """Verify citations in text against retrieved chunks using decision models.

    Args:
        text: Synthesized markdown response text containing citation tokens.
        ranked_chunks: Complete list of retrieved evidence chunks.
        config: Optional configuration dict from zocrates_config.json.
        telemetry_out: Optional dict to receive verification telemetry data.

    Returns:
        Dict mapping chunk index (1-based int) to list of support probabilities
        for each occurrence of that chunk index in document order ([p_occ1, p_occ2, ...]).
    """
    if not text or not ranked_chunks:
        return {}

    start_time = time.perf_counter()

    with _verification_span("citation_verification") as span:
        cfg = dict(config) if config is not None else load_zocrates_config()

        provider = str(cfg.get("citation_verification_provider") or "openrouter")
        if provider == "local_laya":
            model = "laya"
        else:
            model = str(cfg.get("citation_verification_model") or DEFAULT_OPENROUTER_MODEL)
        local_url = str(cfg.get("citation_verification_local_url") or DEFAULT_LOCAL_URL)
        timeout_sec = float(cfg.get("citation_verification_timeout_sec", DEFAULT_TIMEOUT_SEC))
        turn_budget_sec = float(cfg.get("citation_verification_turn_budget_sec", DEFAULT_TURN_BUDGET_SEC))
        verification_threshold = float(cfg.get("citation_verification_threshold", DEFAULT_THRESHOLD))
        execution_mode = "batch" if provider == "local_laya" else "thread_pool"
        endpoint = local_url if provider == "local_laya" else OPENROUTER_DECISIONS_URL

        occurrences = extract_claim_occurrences(text, len(ranked_chunks))
        if not occurrences:
            return {}

        # Initialize results mapping chunk_index -> list of None for each occurrence
        results: dict[int, list[float | None]] = {}
        for item in occurrences:
            c_idx = item["chunk_index"]
            if c_idx not in results:
                results[c_idx] = []
            results[c_idx].append(None)

        item_latencies: dict[tuple[int, int], float | None] = {}
        budget_exceeded = False

        # Build input metadata and assessments for span / telemetry
        input_assessments: list[dict[str, Any]] = []
        for item in occurrences:
            c_idx = item["chunk_index"]
            occ_idx = item["occurrence"]
            is_comp = bool(item.get("is_compound", False))
            chunk = ranked_chunks[c_idx - 1] if (1 <= c_idx <= len(ranked_chunks)) else {}
            src = prepare_source_text(chunk, item["claim"])
            input_assessments.append({
                "chunk_index": c_idx,
                "occurrence": occ_idx,
                "is_compound": is_comp,
                "claim": item["claim"],
                "source_text": src,
                "title": str(chunk.get("title") or ""),
                "section": str(chunk.get("section") or chunk.get("heading") or ""),
                "authors": chunk.get("authors") if chunk.get("authors") is not None else [],
                "year": chunk.get("year") or "",
                "item_key": str(chunk.get("item_key") or ""),
                "attachment_key": chunk.get("attachment_key"),
            })

        metadata_dict = {
            "provider": provider,
            "model": model,
            "endpoint": endpoint,
            "threshold": verification_threshold,
            "execution_mode": execution_mode,
            "turn_budget_sec": turn_budget_sec,
            "batch_size": len(occurrences),
        }

        # Set span input attributes
        try:
            _safe_set(span, "openinference.span.kind", "GUARDRAIL")
            _safe_set(span, "input.mime_type", "application/json")
            _safe_set(span, "input.value", json.dumps(input_assessments))
            _safe_set(span, "metadata", json.dumps(metadata_dict))
        except Exception as span_err:
            logger.debug("Failed recording input span attributes: %s", span_err)

        # Progressively initialize telemetry_out if provided
        if telemetry_out is not None:
            try:
                telemetry_out.clear()
                telemetry_out.update({
                    "schema_version": 1,
                    "provider": provider,
                    "model": model,
                    "endpoint": endpoint,
                    "threshold": verification_threshold,
                    "execution_mode": execution_mode,
                    "turn_budget_sec": turn_budget_sec,
                    "batch_size": len(occurrences),
                    "assessments": [],
                })
            except Exception as tel_err:
                logger.debug("Failed initializing telemetry_out: %s", tel_err)

        if provider == "local_laya":
            effective_budget = max(turn_budget_sec, 20.0, len(occurrences) * 2.5)
            parsed = urllib.parse.urlparse(local_url)
            health_url = f"{parsed.scheme}://{parsed.netloc}/health"
            if not _probe_local_laya_health(health_url):
                logger.warning("Local Laya health probe failed (%s). Failing open.", health_url)
            else:
                # Fast path: Native tensor-batched inference across all citation claim occurrences
                batch_items = [
                    (
                        item["chunk_index"],
                        item["occurrence"],
                        item["claim"],
                        prepare_source_text(
                            ranked_chunks[item["chunk_index"] - 1],
                            item["claim"],
                            max_chars=MAX_LOCAL_SOURCE_TEXT_CHARS,
                        ),
                        bool(item.get("is_compound", False)),
                    )
                    for item in occurrences
                ]
                batch_res = _check_support_local_laya_batch(
                    batch_items,
                    local_url=local_url,
                    timeout=effective_budget,
                )
                if batch_res is not None:
                    for item in occurrences:
                        c_idx = item["chunk_index"]
                        occ_idx = item["occurrence"]
                        prob = batch_res.get((c_idx, occ_idx))
                        results[c_idx][occ_idx - 1] = prob
                else:
                    if (time.perf_counter() - start_time) >= effective_budget:
                        budget_exceeded = True
                    # If batch inference timed out or failed, fail open cleanly without causing a cascading queue jam
                    logger.warning(
                        "Local Laya batch verification failed or timed out (budget: %.1fs). Failing open for %d citations.",
                        effective_budget,
                        len(occurrences),
                    )
        else:
            def _verify_one(
                chunk_idx: int,
                occ_idx: int,
                claim: str,
                is_compound: bool = False,
            ) -> tuple[int, int, float | None, float]:
                """Verify a single claim against its referenced chunk evidence.

                Args:
                    chunk_idx: 1-based index of the candidate evidence chunk.
                    occ_idx: 1-based occurrence number of the citation within the text.
                    claim: Extracted claim sentence to evaluate.
                    is_compound: Whether this citation occurrence originates from a multi-citation bracket.

                Returns:
                    Tuple of (chunk_idx, occ_idx, support_probability, elapsed_ms).
                """
                t0 = time.perf_counter()
                try:
                    chunk = ranked_chunks[chunk_idx - 1]
                    source_text = prepare_source_text(chunk, claim)
                    if provider == "local_laya":
                        prob = _check_support_local_laya(
                            claim=claim,
                            source_text=source_text,
                            model=model,
                            local_url=local_url,
                            timeout=timeout_sec,
                            skip_health_probe=True,
                            is_compound=is_compound,
                        )
                    else:
                        prob = _check_support_openrouter(
                            claim=claim,
                            source_text=source_text,
                            model=model,
                            timeout=timeout_sec,
                            is_compound=is_compound,
                        )
                    elapsed_item_ms = (time.perf_counter() - t0) * 1000.0
                    return chunk_idx, occ_idx, prob, round(elapsed_item_ms, 2)
                except Exception as exc:
                    logger.warning(
                        "Unexpected error verifying chunk %s occ#%s: %s. Failing open.",
                        chunk_idx,
                        occ_idx,
                        exc,
                    )
                    elapsed_item_ms = (time.perf_counter() - t0) * 1000.0
                    return chunk_idx, occ_idx, None, round(elapsed_item_ms, 2)

            loop = asyncio.get_running_loop()
            num_workers = min(len(occurrences), 16)
            if num_workers > 0:
                with concurrent.futures.ThreadPoolExecutor(max_workers=num_workers) as executor:
                    futures = [
                        loop.run_in_executor(
                            executor,
                            _verify_one,
                            item["chunk_index"],
                            item["occurrence"],
                            item["claim"],
                            bool(item.get("is_compound", False)),
                        )
                        for item in occurrences
                    ]
                    try:
                        completed = await asyncio.wait_for(
                            asyncio.gather(*futures),
                            timeout=turn_budget_sec,
                        )
                        for chunk_idx, occ_idx, prob, item_lat in completed:
                            results[chunk_idx][occ_idx - 1] = prob
                            item_latencies[(chunk_idx, occ_idx)] = item_lat
                    except asyncio.TimeoutError:
                        budget_exceeded = True
                        logger.warning(
                            "Citation verification turn budget exceeded (%.1fs). Failing open for remaining citations.",
                            turn_budget_sec,
                        )

        # Build completed assessments and summary
        assessments: list[dict[str, Any]] = []
        kept_count = 0
        demoted_count = 0
        fail_open_count = 0

        for i, item in enumerate(occurrences):
            c_idx = item["chunk_index"]
            occ_idx = item["occurrence"]
            prob = results.get(c_idx, [])[occ_idx - 1] if (c_idx in results and len(results[c_idx]) >= occ_idx) else None

            if prob is None:
                verdict = "FAIL_OPEN"
                fail_open_count += 1
            elif prob >= verification_threshold:
                verdict = "KEPT"
                kept_count += 1
            else:
                verdict = "DEMOTED"
                demoted_count += 1

            assessment_entry = dict(input_assessments[i])
            assessment_entry.update({
                "support_probability": prob,
                "threshold": verification_threshold,
                "verdict": verdict,
                "latency_ms": item_latencies.get((c_idx, occ_idx)) if execution_mode == "thread_pool" else None,
            })
            assessments.append(assessment_entry)

        probs = [float(a["support_probability"]) for a in assessments if a.get("support_probability") is not None]
        mean_support = round(statistics.fmean(probs), 3) if probs else None
        std_support = round(statistics.stdev(probs), 3) if len(probs) >= 2 else (0.0 if len(probs) == 1 else None)

        wall_latency_ms = round((time.perf_counter() - start_time) * 1000.0, 2)
        summary = {
            "batch_size": len(occurrences),
            "kept": kept_count,
            "demoted": demoted_count,
            "fail_open": fail_open_count,
            "mean_support": mean_support,
            "std_support": std_support,
            "latency_ms": wall_latency_ms,
            "budget_exceeded": budget_exceeded,
        }

        # Progressively finalize telemetry_out
        if telemetry_out is not None:
            try:
                telemetry_out["assessments"] = assessments
                telemetry_out["summary"] = summary
                telemetry_out["latency_ms"] = wall_latency_ms
            except Exception as tel_err:
                logger.debug("Failed finalizing telemetry_out: %s", tel_err)

        # Set output attributes on span
        try:
            _safe_set(span, "output.mime_type", "application/json")
            _safe_set(span, "output.value", json.dumps({"summary": summary, "assessments": assessments}))
            _safe_set(span, "citation_verification.provider", provider)
            _safe_set(span, "citation_verification.model", model)
            _safe_set(span, "citation_verification.threshold", verification_threshold)
            _safe_set(span, "citation_verification.batch_size", len(occurrences))
            _safe_set(span, "citation_verification.kept", kept_count)
            _safe_set(span, "citation_verification.demoted", demoted_count)
            _safe_set(span, "citation_verification.fail_open", fail_open_count)
            _safe_set(span, "citation_verification.mean_support", mean_support)
            _safe_set(span, "citation_verification.std_support", std_support)
            _safe_set(span, "citation_verification.latency_ms", wall_latency_ms)
        except Exception as span_err:
            logger.debug("Failed setting span output attributes: %s", span_err)

        # Per-occurrence logging
        for idx in sorted(results.keys()):
            for occ_idx, p in enumerate(results[idx], start=1):
                p_str = f"{p:.4f}" if isinstance(p, (float, int)) else str(p)
                status = "KEPT" if (p is None or p >= verification_threshold) else "DEMOTED"
                logger.info(
                    "Citation verification [%d] occ#%d: p=%s (threshold=%.2f, status=%s)",
                    idx,
                    occ_idx,
                    p_str,
                    verification_threshold,
                    status,
                )

        # Summary line logging
        logger.info(
            "Citation verification complete: %d assessments in %.1fms (provider=%s, model=%s, mode=%s, threshold=%.2f, kept=%d, demoted=%d, fail_open=%d)",
            len(assessments),
            wall_latency_ms,
            provider,
            model,
            execution_mode,
            verification_threshold,
            kept_count,
            demoted_count,
            fail_open_count,
        )

        return results