Spaces:
Sleeping
Sleeping
"""Verification Pass.
Runs after :mod:`app.services.proposal_generator` and before the
proposal is shown to the user. Every factual claim in the draft is
checked against the compact evidence subset that produced the draft.
The default path calls the configured LLM via
:mod:`app.services.llm_client` with task ``verification_pass``. The LLM
returns a structured verdict per claim and a rewritten
``verified_proposal`` with unsupported claims removed or softened.
If the LLM call fails the behaviour depends on the
``ALLOW_LOCAL_PLACEHOLDERS`` flag:
* ``false`` (default) — the report is returned with
``verification_status = "unverified"``, ``meta["status"]`` set to the
provider status and ``cleaned_proposal`` left unchanged. The UI then
shows the draft with a clear "not fact-checked" warning instead of
pretending the proposal is verified. ``"failed"`` is kept for a check
that ran and judged the draft.
* ``true`` — a deterministic claim sweep is used as a clearly-labeled
local fallback.
The function never logs raw dossier text, full prompts, or the API key.
"""
from __future__ import annotations
import json
import logging
import re
import traceback
from dataclasses import dataclass, field
from typing import Any, Iterable, Optional
from app.config import Settings, get_settings
from app.prompts.proposal_prompt import neutralize_tags
from app.prompts.system_prompt import SYSTEM_PROMPT
from app.prompts.verification_prompt import render_prompt as render_verification_prompt
from app.services import llm_client
from app.utils.logging_utils import get_logger
_logger: logging.Logger = get_logger("upwork_strategist.verification")
TASK_NAME = "verification_pass"
# The verifier's reply carries its claim lists AND a full copy of the
# proposal (``verified_proposal``), so it grows with the draft: about
# 1,450-1,900 tokens for a 550-700-word proposal. The output budget scales
# with the draft (about 3 characters a token, plus room for the claim lists)
# between these bounds, and a reply still cut off at the limit is asked for
# once more with twice the room. Output is billed by the tokens written, not
# by the budget.
_VERIFY_MIN_OUTPUT_TOKENS = 4000
_VERIFY_MAX_OUTPUT_TOKENS = 8000
# ---------------------------------------------------------------------------
# Patterns for the local deterministic fallback
# ---------------------------------------------------------------------------
_YEARS_RE = re.compile(
r"(\d+(?:\.\d+)?\+?)\s*(years?|yrs?)\b",
re.IGNORECASE,
)
_PERCENT_RE = re.compile(r"(\d+(?:[\.,]\d+)?)\s*%")
_DOLLAR_RE = re.compile(
r"\$\s?\d[\d,]*(?:\.\d+)?\s?[kKmM]?\b"
r"|\b\d+(?:[\.,]\d+)?\s?[kKmM]\b"
)
_INTEGER_METRIC_RE = re.compile(
r"\b(\d{2,})\s+(features|projects|clients|customers|users|hires|engagements|releases|deals|cases)\b",
re.IGNORECASE,
)
# Evidence-id patterns. Every id (``ev_abc12345``) is first swapped for
# ``_ID_MARK``, so that only the brackets, labels and separators a citation
# leaves behind are cleaned up: ``(ev_a, ev_b)``, ``[ev_a; ev_b]``,
# ``(evidence: ev_a)``, ``(Power BI, ev_a)``. Ordinary brackets such as
# ``(Power BI)``, ``fetch()`` or ``Order[]`` hold no mark and are never
# touched, and nothing is matched across a line break.
_EVIDENCE_ID_BARE_RE = re.compile(r"\bev_[A-Za-z0-9_-]+\b")
_ID_MARK = "\x00"
_ID_SEPARATOR = r"(?:[,;/&]|\band\b)"
_ID_LIST = (
r"(?:\b(?:evidence(?:[_ ]?ids?)?|sources?|see)[ \t]*[:=]?[ \t]*)?"
+ _ID_MARK
+ r"(?:[ \t]*" + _ID_SEPARATOR + r"?[ \t]*(?:and[ \t]+)?" + _ID_MARK + r")*"
)
# A bracket that holds nothing but cited ids goes whole.
_EVIDENCE_ID_INLINE_RE = re.compile(
r"[ \t]*[\(\[][ \t]*" + _ID_LIST + r"[ \t]*[\)\]]", re.IGNORECASE
)
# Ids cited next to other words in a bracket go with their separator.
_EVIDENCE_ID_IN_BRACKET_RE = re.compile(
r"[ \t]*" + _ID_SEPARATOR + r"[ \t]*" + _ID_LIST + r"(?=[ \t]*[\)\]])"
r"|(?<=[\(\[])[ \t]*" + _ID_LIST + r"[ \t]*" + _ID_SEPARATOR + r"[ \t]*",
re.IGNORECASE,
)
# Any other citation ("Evidence: ev_a, ev_b", a bare ev_a) goes with its
# label and the separators between its ids.
_EVIDENCE_ID_LIST_RE = re.compile(r"[ \t]*" + _ID_LIST, re.IGNORECASE)
# ---------------------------------------------------------------------------
# Report
# ---------------------------------------------------------------------------
def _local_meta(note: str, *, status: str = "local_placeholder") -> dict:
return {
"task_name": TASK_NAME,
"used_api": False,
"status": status,
"provider": None,
"model": None,
"error_message": note,
"claims_checked": 0,
"unsupported_claims_count": 0,
}
@dataclass
class VerificationReport:
cleaned_proposal: str
removed_claims: list[str] = field(default_factory=list)
softened_claims: list[dict[str, str]] = field(default_factory=list)
flagged_as_missing: list[str] = field(default_factory=list)
surviving_claims: list[dict[str, Any]] = field(default_factory=list)
supported_claims: list[dict[str, Any]] = field(default_factory=list)
partially_supported_claims: list[dict[str, Any]] = field(default_factory=list)
unsupported_claims: list[dict[str, Any]] = field(default_factory=list)
verification_status: str = "skipped"
summary: str = ""
meta: dict = field(default_factory=lambda: _local_meta(
"Verification pass has not run yet."
))
# ---------------------------------------------------------------------------
# Evidence-id stripping
# ---------------------------------------------------------------------------
def strip_evidence_ids(text: str) -> str:
"""Remove evidence-id citations: ``(ev_xxx)``, ``[ev_xxx; ev_yyy]``,
``(evidence: ev_xxx)`` / bare ``ev_xxx`` tokens.
Evidence IDs are useful internally (citations, claim verification)
but must never appear in the client-facing proposal.
"""
if not text:
return text
cleaned = _EVIDENCE_ID_BARE_RE.sub(_ID_MARK, text.replace(_ID_MARK, ""))
cleaned = _EVIDENCE_ID_INLINE_RE.sub("", cleaned)
cleaned = _EVIDENCE_ID_IN_BRACKET_RE.sub("", cleaned)
cleaned = _EVIDENCE_ID_LIST_RE.sub("", cleaned)
cleaned = re.sub(r"[ \t]{2,}", " ", cleaned)
cleaned = re.sub(r"\s+([.,;:!?])", r"\1", cleaned)
cleaned = re.sub(r"\n{3,}", "\n\n", cleaned)
return cleaned.strip()
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _proof_attr(proof: Any, name: str, default: Any = None) -> Any:
if hasattr(proof, name):
return getattr(proof, name)
if isinstance(proof, dict):
return proof.get(name, default)
return default
# Model replies are JSON of the right syntax but not always of the shape the
# prompt asks for. These coerce a field before it is used, so an odd shape
# never crashes the pass or reaches the client as Python text.
def _text(value: Any) -> str:
"""A model field that should be a string, as stripped text."""
return "" if value is None else str(value).strip()
def _plain_text(value: Any) -> str:
"""Prose from a model field that should be one string.
A list of paragraphs, or a dict of sections (``{"hook": ..., "plan":
[...]}``), is joined in order, one paragraph each, instead of being
shown as Python's ``['...']`` or ``{'...'}`` form.
"""
if isinstance(value, dict):
value = list(value.values())
if isinstance(value, (list, tuple)):
return "\n\n".join(p for p in (_plain_text(v) for v in value) if p)
return _text(value)
def _as_ids(value: Any) -> list[str]:
"""Evidence ids as a list of text: a bare id, a list of ids, or none."""
if isinstance(value, str):
value = [value]
if not isinstance(value, (list, tuple)):
return []
return [v.strip() for v in value if isinstance(v, str) and v.strip()]
def _evidence_text_corpus(evidence: Iterable) -> str:
parts: list[str] = []
for proof in evidence:
parts.append(str(_proof_attr(proof, "claim_text", "")))
parts.append(str(_proof_attr(proof, "normalized_value", "") or ""))
for attr in ("skills", "tools", "industries", "metrics"):
for item in _proof_attr(proof, attr, []) or []:
parts.append(str(item))
return " \n".join(parts).lower()
def _evidence_id_set(evidence: Iterable) -> set[str]:
return {
_proof_attr(p, "evidence_id")
for p in evidence
if _proof_attr(p, "evidence_id")
}
def _drop_sentence_containing(text: str, needle: str) -> tuple[str, Optional[str]]:
"""Drop the sentence containing ``needle``.
Returns ``(new_text, removed_fragment)`` only when the text actually
changed; otherwise ``(text, None)`` so callers never report a claim as
"removed" while leaving it in the proposal.
"""
if not needle or needle not in text:
return text, None
pieces = re.split(r"(?<=[.!?\n])\s+", text)
for idx, piece in enumerate(pieces):
if needle in piece:
removed = piece.strip()
pieces.pop(idx)
return " ".join(p for p in pieces if p).strip(), removed
cleaned = text.replace(needle, "").strip()
if cleaned == text:
return text, None
return cleaned, needle
def _replace_once(text: str, needle: str, replacement: str) -> str:
return text.replace(needle, replacement, 1)
def _squash(text: str) -> str:
"""``text`` in lower case with every run of whitespace as one space."""
return " ".join(str(text or "").split()).lower()
def _claim_pattern(claim: str) -> Optional[re.Pattern[str]]:
"""Match ``claim`` as whole words, in any case and with any spacing or
line break ("AWS" never matches inside "laws")."""
words = str(claim or "").split()
if not words:
return None
body = r"\s+".join(re.escape(w) for w in words)
return re.compile(r"(?<!\w)" + body + r"(?!\w)", re.IGNORECASE)
def _drop_claim_sentences(text: str, pattern: re.Pattern[str]) -> str:
"""Drop every sentence that holds ``pattern``.
Unlike :func:`_drop_sentence_containing`, the paragraph and line breaks
of the rest are kept as they were.
"""
lines: list[str] = []
for line in text.split("\n"):
if pattern.search(line):
line = " ".join(
s for s in re.split(r"(?<=[.!?])\s+", line) if not pattern.search(s)
)
if not line.strip():
continue
lines.append(line)
return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip()
def _enforce_claim_verdicts(
text: str,
*,
rewritten: bool,
removed: list[str],
softenings: list[tuple[str, str]],
unsoftened: list[str],
) -> tuple[str, list[str]]:
"""Make ``text`` agree with the verifier's own verdicts.
A verifier can flag a claim and still hand back text that holds it: an
empty or missing ``verified_proposal`` (the draft stands in), the draft
echoed unchanged, or a rewrite that kept the flagged sentence. A claim
marked for removal that is still in the text loses its sentence, and a
partial claim still in the text is replaced by its suggested softening.
Returns ``(text, unresolved)``: the flagged claims that cannot be
confirmed gone. That is a claim still in the text, or, when the draft
was not rewritten, a claim not found word for word (the verifier
paraphrased it) or one with no wording to soften it to.
"""
unresolved: list[str] = []
for claim in removed:
pattern = _claim_pattern(claim)
if pattern is None:
continue
if pattern.search(text):
text = _drop_claim_sentences(text, pattern)
if pattern.search(text):
unresolved.append(claim)
elif not rewritten:
unresolved.append(claim)
for claim, softening in softenings:
pattern = _claim_pattern(claim)
if pattern is None:
continue
if pattern.search(text):
def _soften(match: re.Match, softening: str = softening) -> str:
if match.group(0)[:1].isupper():
return softening[:1].upper() + softening[1:]
return softening
text = pattern.sub(_soften, text)
elif not rewritten:
unresolved.append(claim)
if not rewritten:
unresolved.extend(c for c in unsoftened if c)
return text, unresolved
def _number_token_in_corpus(number_token: str, corpus: str) -> bool:
"""True if ``number_token`` appears as a whole number in ``corpus``.
Guards against the bare-substring false-positive where a digit (e.g.
"5") is considered "supported" merely because it occurs inside an
unrelated token ("HTML5", "$50k", "Python 3.5"). The token is matched
bounded by non-digit characters so "5" matches "5 years" / "(5)" but
not "50" or "HTML5".
"""
token = (number_token or "").strip().lower()
if not token:
return False
pattern = r"(?<![0-9.,])" + re.escape(token) + r"(?![0-9])"
return re.search(pattern, corpus) is not None
def _compact_evidence_for_prompt(
evidence: Iterable, *, claim_text_cap: int = 240
) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for proof in evidence:
text = str(_proof_attr(proof, "claim_text", "") or "").strip()
if len(text) > claim_text_cap:
text = text[: max(claim_text_cap - 1, 1)].rstrip() + "…"
out.append(
{
"evidence_id": _proof_attr(proof, "evidence_id"),
"source_type": _proof_attr(proof, "source_type"),
"claim_type": _proof_attr(proof, "claim_type"),
"claim": text,
}
)
return out
def _compact_job_for_prompt(confirmed_job: dict) -> dict[str, str]:
out: dict[str, str] = {}
for key, entry in (confirmed_job or {}).items():
if isinstance(entry, dict):
value = str(entry.get("value", "") or "").strip()
else:
value = str(entry or "").strip()
if value and value.lower() != "not visible":
out[key] = value
return out
def _compact_claims_for_prompt(
factual_claims: list[dict[str, Any]],
) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for claim in factual_claims or []:
if not isinstance(claim, dict):
continue
ids = _as_ids(claim.get("evidence_id"))
out.append(
{
"text": str(claim.get("text") or "").strip(),
"kind": claim.get("kind"),
"claim_type": claim.get("claim_type"),
"evidence_id": ids[0] if len(ids) == 1 else (ids or None),
}
)
return out
# ---------------------------------------------------------------------------
# Deterministic local fallback sweeps
# ---------------------------------------------------------------------------
def _sweep_factual_claims(
proposal: str,
factual_claims: list[dict[str, Any]],
evidence_ids: set[str],
) -> tuple[str, list[str], list[dict[str, Any]]]:
removed: list[str] = []
surviving: list[dict[str, Any]] = []
for claim in factual_claims or []:
ev_id = claim.get("evidence_id")
text = (claim.get("text") or "").strip()
if ev_id and ev_id in evidence_ids:
surviving.append(claim)
continue
if text:
new_proposal, dropped = _drop_sentence_containing(proposal, text)
if dropped is not None:
# The proposal text actually changed — report the exact
# fragment that was removed.
proposal = new_proposal
removed.append(dropped)
else:
# The claim wording was not a literal substring, so the
# sentence could not be excised. Report it honestly as an
# unsupported claim that needs manual review rather than
# logging a phantom "removed" while leaving it in place.
removed.append(text)
else:
removed.append(claim.get("kind", "unknown claim"))
return proposal, removed, surviving
def _sweep_unattributed_specifics(
proposal: str, evidence_corpus: str
) -> tuple[str, list[dict[str, str]]]:
softened: list[dict[str, str]] = []
def soften(pattern: re.Pattern[str], replacement: str) -> None:
nonlocal proposal
for match in list(pattern.finditer(proposal)):
snippet = match.group(0)
if snippet.lower() in evidence_corpus:
continue
digit_token = match.group(1) if match.groups() else snippet
# Match the number as a WHOLE token in the corpus, not as a bare
# substring: otherwise "5 years" would be treated as supported
# whenever "5" appears anywhere (e.g. "HTML5", "$50k",
# "Python 3.5"). Require the exact digit token bounded by
# non-alphanumeric characters.
if digit_token and _number_token_in_corpus(digit_token, evidence_corpus):
continue
proposal = _replace_once(proposal, snippet, replacement)
softened.append({"original": snippet, "softened": replacement})
soften(_YEARS_RE, "several years")
soften(_PERCENT_RE, "a meaningful share")
soften(_DOLLAR_RE, "a meaningful amount")
soften(_INTEGER_METRIC_RE, "a number of")
return proposal, softened
def _missing_info_suggestions(
removed: list[str],
softened: list[dict[str, str]],
factual_claims: list[dict[str, Any]],
evidence: list,
) -> list[str]:
suggestions: list[str] = []
if removed:
suggestions.append(
"Some claims were removed because no proof point supported them. "
"Add a matching file to the dossier (testimonial, case study, or "
"portfolio entry) if you want these claims to stand."
)
if any(s["softened"] == "several years" for s in softened):
suggestions.append(
"Years-of-experience figure was softened. Add a resume or LinkedIn "
"summary that states the exact number."
)
if any(s["softened"] in {"a meaningful share", "a meaningful amount"} for s in softened):
suggestions.append(
"A specific metric was softened. Add a case study or testimonial "
"that documents the number you want to cite."
)
if any(s["softened"] == "a number of" for s in softened):
suggestions.append(
"A specific count was softened. Add a portfolio breakdown that "
"shows the exact count."
)
claim_types_present = {
_proof_attr(p, "claim_type") for p in evidence if _proof_attr(p, "claim_type")
}
for needed, hint in (
("testimonial", "No client testimonials in the dossier — add one to support social-proof phrases."),
("portfolio", "No portfolio entries in the dossier — add at least one to support project references."),
("certification", "No certifications in the dossier — add credentials to support cert-related claims."),
):
if needed not in claim_types_present and any(
c.get("kind") == needed for c in factual_claims or []
):
suggestions.append(hint)
return suggestions
def _local_fallback(
proposal: str,
evidence_list: list,
factual_claims: list[dict[str, Any]],
*,
note: str,
status: str,
) -> VerificationReport:
"""Run the deterministic claim sweep as a labelled local fallback."""
evidence_ids = _evidence_id_set(evidence_list)
corpus = _evidence_text_corpus(evidence_list)
proposal_after_claims, removed, surviving = _sweep_factual_claims(
proposal, factual_claims or [], evidence_ids
)
proposal_after_specifics, softened = _sweep_unattributed_specifics(
proposal_after_claims, corpus
)
cleaned = re.sub(r"[ \t]{2,}", " ", proposal_after_specifics)
cleaned = re.sub(r"\n{3,}", "\n\n", cleaned).strip()
cleaned = strip_evidence_ids(cleaned)
suggestions = _missing_info_suggestions(
removed, softened, factual_claims or [], evidence_list
)
meta = _local_meta(note, status=status)
meta["claims_checked"] = len(factual_claims or [])
meta["unsupported_claims_count"] = len(removed)
llm_client.record_local_use(TASK_NAME, note=note)
if removed or softened:
verification_status = "passed_with_softening"
else:
verification_status = "passed"
return VerificationReport(
cleaned_proposal=cleaned,
removed_claims=removed,
softened_claims=softened,
flagged_as_missing=suggestions,
surviving_claims=surviving,
supported_claims=[
{"claim": c.get("text") or c.get("kind") or "", "evidence_ids": [c.get("evidence_id")] if c.get("evidence_id") else []}
for c in surviving
],
partially_supported_claims=[
{
"claim": s.get("original", ""),
"reason": "Specific value not present in evidence.",
"suggested_softening": s.get("softened", ""),
"evidence_ids": [],
}
for s in softened
],
unsupported_claims=[
{"claim": r, "reason": "No matching evidence point.", "action": "remove"}
for r in removed
],
verification_status=verification_status,
summary=(
f"LOCAL FALLBACK — LLM verification not used. "
f"Removed {len(removed)} claim(s); softened {len(softened)} value(s)."
),
meta=meta,
)
def _failure_report(
proposal: str,
*,
note: str,
status: str,
provider: Optional[str] = None,
model: Optional[str] = None,
error_message: Optional[str] = None,
claims_checked: int = 0,
) -> VerificationReport:
"""Return a report that explicitly says verification did not run.
The status is ``"unverified"``, never ``"failed"``: the draft was not
judged, so it is kept and shown as not fact-checked. The proposal text
is returned unchanged but evidence-ids are still stripped so the UI's
main copy box never leaks them.
"""
meta = {
"task_name": TASK_NAME,
"used_api": False,
"status": status,
"provider": provider,
"model": model,
"error_message": llm_client.sanitize_error_message(error_message) or note,
"claims_checked": claims_checked,
"unsupported_claims_count": 0,
}
return VerificationReport(
cleaned_proposal=strip_evidence_ids(proposal),
verification_status="unverified",
summary=note,
meta=meta,
)
# ---------------------------------------------------------------------------
# LLM-backed verification
# ---------------------------------------------------------------------------
def _parse_llm_payload(
payload: Any, fallback_proposal: str
) -> dict[str, Any]:
"""Coerce the LLM response into the expected JSON shape.
A reply that is not an object, or that carries no verdict, no flagged
claim and no rewritten proposal, is no verification at all: it comes
back ``"unverified"`` with the draft, never as a pass. When the
rewrite is missing, ``rewrite_missing`` is true and the draft stands
in for it, so the flagged claims must still be taken out of it.
"""
if not isinstance(payload, dict):
return {
"verification_status": "unverified",
"supported_claims": [],
"partially_supported_claims": [],
"unsupported_claims": [],
"verified_proposal": fallback_proposal,
"rewrite_missing": True,
"missing_information": [],
"summary": "Verification response was not a JSON object.",
}
def _as_list(value: Any) -> list:
if isinstance(value, list):
return value
if value in (None, ""):
return []
return [value]
verified = _plain_text(payload.get("verified_proposal"))
rewrite_missing = not verified
if rewrite_missing:
verified = fallback_proposal
summary = str(payload.get("summary") or "").strip()
status_raw = str(payload.get("verification_status") or "").strip().lower()
if status_raw not in {"passed", "passed_with_softening", "failed"}:
# Infer from claim counts if the model omitted the field.
partials = _as_list(payload.get("partially_supported_claims"))
unsup = _as_list(payload.get("unsupported_claims"))
if not partials and not unsup and rewrite_missing:
# No verdict, nothing flagged and no checked text: nothing shows
# the draft was actually checked, so it is not passed on as
# verified.
status_raw = "unverified"
summary = summary or "Verification response had no verdict and no checked proposal."
elif not partials and not unsup:
status_raw = "passed"
elif unsup and len(unsup) >= 3:
status_raw = "failed"
else:
status_raw = "passed_with_softening"
return {
"verification_status": status_raw,
"supported_claims": _as_list(payload.get("supported_claims")),
"partially_supported_claims": _as_list(payload.get("partially_supported_claims")),
"unsupported_claims": _as_list(payload.get("unsupported_claims")),
"verified_proposal": verified,
"rewrite_missing": rewrite_missing,
"missing_information": _as_list(payload.get("missing_information")),
"summary": summary,
}
def _claims_checked_count(parsed: dict[str, Any]) -> int:
return (
len(parsed.get("supported_claims") or [])
+ len(parsed.get("partially_supported_claims") or [])
+ len(parsed.get("unsupported_claims") or [])
)
def _restore_defanged_tags(verified: str, proposal: str) -> str:
"""Put back the angle-bracket tokens the prompt defanged for the verifier.
The verification prompt shows the draft with simple tags defanged
(``<Suspense>`` -> ``(Suspense)``), so a verifier that rewrites the
draft hands code such as ``Promise<void>`` back as ``Promise(void)``.
A defanged form is restored only when the draft did not contain it.
"""
if not verified or "<" not in (proposal or ""):
return verified
for token in set(re.findall(r"<[^<>\n]{1,80}>", proposal)):
defanged = neutralize_tags(token)
if defanged != token and defanged not in proposal:
verified = verified.replace(defanged, token)
return verified
def _verification_budget(proposal: str) -> int:
"""Output tokens for a verifier reply that echoes ``proposal`` back."""
wanted = len(proposal or "") // 3 + 2500
return min(max(_VERIFY_MIN_OUTPUT_TOKENS, wanted), _VERIFY_MAX_OUTPUT_TOKENS)
def _call_verification_llm(
*,
proposal: str,
evidence_list: list,
factual_claims: list[dict[str, Any]],
confirmed_job_fields: dict,
settings: Settings,
max_tokens: Optional[int] = None,
profile_block: str = "{}",
) -> Any:
compact_evidence = _compact_evidence_for_prompt(evidence_list)
compact_job = _compact_job_for_prompt(confirmed_job_fields)
compact_claims = _compact_claims_for_prompt(factual_claims)
referenced_ids = sorted(
{ev for c in compact_claims for ev in _as_ids(c.get("evidence_id"))}
)
user_prompt = render_verification_prompt(
proposal_block=proposal,
profile_block=profile_block,
factual_claims_block=json.dumps(compact_claims, ensure_ascii=False, indent=2),
evidence_block=json.dumps(compact_evidence, ensure_ascii=False, indent=2),
confirmed_job_block=json.dumps(compact_job, ensure_ascii=False, indent=2),
referenced_ids_block=json.dumps(referenced_ids, ensure_ascii=False),
)
return llm_client.call_text_llm(
task_name=TASK_NAME,
system_prompt=SYSTEM_PROMPT,
user_prompt=user_prompt,
expected_json=True,
max_tokens=max_tokens or _verification_budget(proposal),
settings=settings,
)
def _missing_info_from_payload(
parsed: dict[str, Any], evidence_list: list, factual_claims: list[dict[str, Any]]
) -> list[str]:
suggestions: list[str] = []
for item in parsed.get("missing_information") or []:
if isinstance(item, str) and item.strip():
suggestions.append(item.strip())
elif isinstance(item, dict) and item.get("hint"):
suggestions.append(str(item["hint"]).strip())
# Add dossier-shape suggestions for claim types missing from evidence.
claim_types_present = {
_proof_attr(p, "claim_type") for p in evidence_list if _proof_attr(p, "claim_type")
}
for needed, hint in (
("testimonial", "No client testimonials in the dossier — add one to support social-proof phrases."),
("portfolio", "No portfolio entries in the dossier — add at least one to support project references."),
("certification", "No certifications in the dossier — add credentials to support cert-related claims."),
):
if needed not in claim_types_present and any(
(c.get("kind") == needed or c.get("claim_type") == needed)
for c in factual_claims or []
):
if hint not in suggestions:
suggestions.append(hint)
return suggestions[:8]
def _report_from_llm(
*,
parsed: dict[str, Any],
llm_result: Any,
evidence_list: list,
factual_claims: list[dict[str, Any]],
proposal: str = "",
) -> VerificationReport:
verified = strip_evidence_ids(parsed.get("verified_proposal") or "")
supported = [
c for c in parsed.get("supported_claims") or [] if isinstance(c, (dict, str))
]
partials = [
c for c in parsed.get("partially_supported_claims") or [] if isinstance(c, dict)
]
unsup = [
c for c in parsed.get("unsupported_claims") or [] if isinstance(c, dict)
]
def _softens(c: dict) -> bool:
return _text(c.get("action")).lower() == "soften"
removed = [_text(c.get("claim")) for c in unsup if not _softens(c) and _text(c.get("claim"))]
softened = [
{
"original": _text(c.get("claim")),
"softened": _text(c.get("suggested_softening")),
}
for c in partials
if _text(c.get("claim"))
]
# Unsupported claims marked "soften" count as softenings too.
for c in unsup:
if _softens(c) and _text(c.get("claim")):
softened.append(
{
"original": _text(c.get("claim")),
"softened": "(softened)",
}
)
# The text shown must agree with these verdicts. A flagged claim the
# text still holds is taken out (or softened), and when that cannot be
# confirmed the check is reported as failed, never passed.
verification_status = parsed.get("verification_status") or "passed"
summary = parsed.get("summary") or ""
rewritten = not parsed.get("rewrite_missing") and _squash(verified) != _squash(
strip_evidence_ids(proposal)
)
verified, unresolved = _enforce_claim_verdicts(
verified,
rewritten=rewritten,
removed=removed,
softenings=[
(s["original"], s["softened"])
for s in softened
if s["softened"] not in ("", "(softened)")
],
unsoftened=[
s["original"] for s in softened if s["softened"] in ("", "(softened)")
],
)
if unresolved:
verification_status = "failed"
summary = (
"Some flagged claims could not be taken out of the text. " + summary
).strip()
# Grounding gate: a claim only "survives" if its evidence_id is BOTH
# echoed by the verifier model AND present in the real evidence subset
# that produced the proposal. Both the generator and the verifier are
# untrusted, so the only authoritative source of real ids is
# ``evidence_list`` — without this intersection a fabricated
# evidence_id that the verifier merely repeats would be treated as
# grounded, defeating the "grounded only in real evidence" guarantee.
# (The deterministic local fallback already does this; this aligns the
# LLM path with it.)
real_evidence_ids = _evidence_id_set(evidence_list)
surviving: list[dict[str, Any]] = []
supported_evidence_ids: set[str] = set()
for entry in supported:
if isinstance(entry, dict):
for ev in _as_ids(entry.get("evidence_ids")):
if ev in real_evidence_ids:
supported_evidence_ids.add(ev)
for claim in factual_claims or []:
# A claim that cites several ids survives only if every one does.
ids = _as_ids(claim.get("evidence_id")) if isinstance(claim, dict) else []
if ids and all(ev in supported_evidence_ids for ev in ids):
surviving.append(claim)
missing_info = _missing_info_from_payload(parsed, evidence_list, factual_claims)
claims_checked = _claims_checked_count(parsed)
unsupported_count = len(unsup) + len(partials)
sanitized_error = llm_client.sanitize_error_message(
getattr(llm_result, "error_message", None)
)
meta = {
"task_name": TASK_NAME,
"used_api": True,
"status": getattr(llm_result, "status", "ok"),
"provider": getattr(llm_result, "provider", None),
"model": getattr(llm_result, "model", None),
"error_message": sanitized_error,
"claims_checked": claims_checked,
"unsupported_claims_count": unsupported_count,
}
# Attach the verification metadata to the most recent usage-log
# entry so the API Usage panel sees claim counts without us sending
# raw text. We also re-assert used_api / provider / model so the
# entry is correct even when a test bypasses the central finalize
# step by monkey-patching the provider call directly.
try:
llm_client.extend_last_entry(
TASK_NAME,
{
"used_api": True,
"provider": getattr(llm_result, "provider", None),
"model": getattr(llm_result, "model", None),
"status": getattr(llm_result, "status", "ok"),
"claims_checked": claims_checked,
"unsupported_claims_count": unsupported_count,
},
)
except Exception: # pragma: no cover
pass
return VerificationReport(
cleaned_proposal=verified,
removed_claims=removed,
softened_claims=softened,
flagged_as_missing=missing_info,
surviving_claims=surviving,
supported_claims=[c for c in supported if isinstance(c, dict)],
partially_supported_claims=partials,
unsupported_claims=unsup,
verification_status=verification_status,
summary=summary,
meta=meta,
)
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def verify(
proposal: str,
evidence: Iterable,
factual_claims: Optional[list[dict[str, Any]]] = None,
*,
confirmed_job_fields: Optional[dict] = None,
settings: Optional[Settings] = None,
profile_block: str = "{}",
) -> VerificationReport:
"""Run the verification pass over a draft proposal.
The default path calls the LLM via the central client. If the call
fails the behaviour depends on the ``ALLOW_LOCAL_PLACEHOLDERS``
flag — when false, the report is marked unverified (the draft was not
checked); when true a deterministic sweep is used as a labelled
fallback.
It never raises: a reply whose shape still trips the pass comes back
unverified with the draft, so one odd reply can neither throw the
draft away nor make the caller pay for a second check.
"""
try:
return _verify(
proposal,
evidence,
factual_claims,
confirmed_job_fields=confirmed_job_fields,
settings=settings,
profile_block=profile_block,
)
except Exception as exc: # noqa: BLE001 - reported as an unverified draft
# Only the error's type and place are logged, never its message,
# which can quote proposal or dossier text.
frames = traceback.extract_tb(exc.__traceback__)
_logger.error(
"task=%s status=exception error_type=%s at=%s",
TASK_NAME,
type(exc).__name__,
f"{frames[-1].name}:{frames[-1].lineno}" if frames else "?",
)
return _failure_report(
str(proposal or ""),
note="Verification could not finish because of an unexpected error.",
status=llm_client.STATUS_EXCEPTION,
)
def _verify(
proposal: str,
evidence: Iterable,
factual_claims: Optional[list[dict[str, Any]]] = None,
*,
confirmed_job_fields: Optional[dict] = None,
settings: Optional[Settings] = None,
profile_block: str = "{}",
) -> VerificationReport:
"""The verification pass itself; see :func:`verify`."""
settings = settings or get_settings()
evidence_list = list(evidence or [])
# Profile facts are also grounding for local numeric/claim safeguards.
if profile_block and profile_block != "{}":
evidence_list.append({"evidence_id": "profile", "claim_type": "profile", "claim_text": profile_block})
claims_list = list(factual_claims or [])
confirmed_job_fields = confirmed_job_fields or {}
# No API key available — skip the LLM call entirely.
if not settings.has_api_key:
if settings.allow_local_placeholders:
return _local_fallback(
proposal,
evidence_list,
claims_list,
note=(
"LOCAL FALLBACK — LLM verification not used. "
"ALLOW_LOCAL_PLACEHOLDERS=true; deterministic claim sweep ran."
),
status="local_placeholder",
)
return _failure_report(
proposal,
note=(
"Verification could not run because no LLM API key is configured. "
"Set ANTHROPIC_API_KEY or OPENAI_API_KEY, or enable "
"ALLOW_LOCAL_PLACEHOLDERS=true for the deterministic local "
"claim sweep."
),
status=llm_client.STATUS_NO_API,
claims_checked=len(claims_list),
)
budget = _verification_budget(proposal)
llm_result = _call_verification_llm(
proposal=proposal,
evidence_list=evidence_list,
factual_claims=claims_list,
confirmed_job_fields=confirmed_job_fields,
settings=settings,
max_tokens=budget,
profile_block=profile_block,
)
if getattr(llm_result, "status", None) == llm_client.STATUS_TRUNCATED:
# The reply was cut off at the budget: ask once more with twice the
# room before giving up on the check.
llm_result = _call_verification_llm(
proposal=proposal,
evidence_list=evidence_list,
factual_claims=claims_list,
confirmed_job_fields=confirmed_job_fields,
settings=settings,
max_tokens=budget * 2,
profile_block=profile_block,
)
success = bool(
getattr(llm_result, "success", False)
and isinstance(getattr(llm_result, "response_json", None), dict)
)
if success:
parsed = _parse_llm_payload(
getattr(llm_result, "response_json", None), fallback_proposal=proposal
)
parsed["verified_proposal"] = _restore_defanged_tags(
parsed["verified_proposal"], proposal
)
return _report_from_llm(
parsed=parsed,
llm_result=llm_result,
evidence_list=evidence_list,
factual_claims=claims_list,
proposal=proposal,
)
# LLM call failed.
sanitized = llm_client.sanitize_error_message(
getattr(llm_result, "error_message", None)
)
if settings.allow_local_placeholders:
return _local_fallback(
proposal,
evidence_list,
claims_list,
note=(
"LOCAL FALLBACK — LLM verification call failed; deterministic "
"claim sweep ran instead. "
+ (f"Provider reason: {sanitized}" if sanitized else "")
).strip(),
status="local_placeholder",
)
status = getattr(llm_result, "status", "failed") or "failed"
if getattr(llm_result, "success", False):
# The call itself worked, but the reply was not a JSON object.
status = llm_client.STATUS_PARSE_ERROR
return _failure_report(
proposal,
note="Verification could not run because the LLM API call failed.",
status=status,
provider=getattr(llm_result, "provider", None),
model=getattr(llm_result, "model", None),
error_message=sanitized,
claims_checked=len(claims_list),
)
__all__ = [
"VerificationReport",
"TASK_NAME",
"verify",
"strip_evidence_ids",
]
"""Verification Pass.
Runs after :mod:`app.services.proposal_generator` and before the
proposal is shown to the user. Every factual claim in the draft is
checked against the compact evidence subset that produced the draft.
The default path calls the configured LLM via
:mod:`app.services.llm_client` with task ``verification_pass``. The LLM
returns a structured verdict per claim and a rewritten
``verified_proposal`` with unsupported claims removed or softened.
If the LLM call fails the behaviour depends on the
``ALLOW_LOCAL_PLACEHOLDERS`` flag:
* ``false`` (default) — the report is returned with
``verification_status = "unverified"``, ``meta["status"]`` set to the
provider status and ``cleaned_proposal`` left unchanged. The UI then
shows the draft with a clear "not fact-checked" warning instead of
pretending the proposal is verified. ``"failed"`` is kept for a check
that ran and judged the draft.
* ``true`` — a deterministic claim sweep is used as a clearly-labeled
local fallback.
The function never logs raw dossier text, full prompts, or the API key.
"""
from __future__ import annotations
import json
import logging
import re
import traceback
from dataclasses import dataclass, field
from typing import Any, Iterable, Optional
from app.config import Settings, get_settings
from app.prompts.proposal_prompt import neutralize_tags
from app.prompts.system_prompt import SYSTEM_PROMPT
from app.prompts.verification_prompt import render_prompt as render_verification_prompt
from app.services import llm_client
from app.utils.logging_utils import get_logger
_logger: logging.Logger = get_logger("upwork_strategist.verification")
TASK_NAME = "verification_pass"
# The verifier's reply carries its claim lists AND a full copy of the
# proposal (``verified_proposal``), so it grows with the draft: about
# 1,450-1,900 tokens for a 550-700-word proposal. The output budget scales
# with the draft (about 3 characters a token, plus room for the claim lists)
# between these bounds, and a reply still cut off at the limit is asked for
# once more with twice the room. Output is billed by the tokens written, not
# by the budget.
_VERIFY_MIN_OUTPUT_TOKENS = 4000
_VERIFY_MAX_OUTPUT_TOKENS = 8000
# ---------------------------------------------------------------------------
# Patterns for the local deterministic fallback
# ---------------------------------------------------------------------------
_YEARS_RE = re.compile(
r"(\d+(?:\.\d+)?\+?)\s*(years?|yrs?)\b",
re.IGNORECASE,
)
_PERCENT_RE = re.compile(r"(\d+(?:[\.,]\d+)?)\s*%")
_DOLLAR_RE = re.compile(
r"\$\s?\d[\d,]*(?:\.\d+)?\s?[kKmM]?\b"
r"|\b\d+(?:[\.,]\d+)?\s?[kKmM]\b"
)
_INTEGER_METRIC_RE = re.compile(
r"\b(\d{2,})\s+(features|projects|clients|customers|users|hires|engagements|releases|deals|cases)\b",
re.IGNORECASE,
)
# Evidence-id patterns. Every id (``ev_abc12345``) is first swapped for
# ``_ID_MARK``, so that only the brackets, labels and separators a citation
# leaves behind are cleaned up: ``(ev_a, ev_b)``, ``[ev_a; ev_b]``,
# ``(evidence: ev_a)``, ``(Power BI, ev_a)``. Ordinary brackets such as
# ``(Power BI)``, ``fetch()`` or ``Order[]`` hold no mark and are never
# touched, and nothing is matched across a line break.
_EVIDENCE_ID_BARE_RE = re.compile(r"\bev_[A-Za-z0-9_-]+\b")
_ID_MARK = "\x00"
_ID_SEPARATOR = r"(?:[,;/&]|\band\b)"
_ID_LIST = (
r"(?:\b(?:evidence(?:[_ ]?ids?)?|sources?|see)[ \t]*[:=]?[ \t]*)?"
+ _ID_MARK
+ r"(?:[ \t]*" + _ID_SEPARATOR + r"?[ \t]*(?:and[ \t]+)?" + _ID_MARK + r")*"
)
# A bracket that holds nothing but cited ids goes whole.
_EVIDENCE_ID_INLINE_RE = re.compile(
r"[ \t]*[\(\[][ \t]*" + _ID_LIST + r"[ \t]*[\)\]]", re.IGNORECASE
)
# Ids cited next to other words in a bracket go with their separator.
_EVIDENCE_ID_IN_BRACKET_RE = re.compile(