woogi
d37ac8c1c7
- ENABLE_CODE_REVIEW flag (default off): skips code/ADA/jurisdiction review path in both pipelines; nothing deleted, one env flag to restore. - Per-sheet Drawing Integrity QA wave (agent + classic, default on): dangling refs, on-sheet contradictions, dimension sanity, missing sheet essentials, tag hygiene. New DrawingIntegrityAgent + classic stage. - Broadened conflict critic: intra-sheet + same-discipline contradictions, not just cross-discipline. - Wave 6.5 Brain-directed clarification (bounded hub-and-spoke): Brain names uncertain findings, verify_evidence requests route through the wave-5b verifier; refuted findings suppressed. One planning call + capped verifies, single iteration. Shared _build_verify_scopes across 5b and 6.5. - Config knobs, .env.example, frontend copy, tests (182 passing).
656 lines
25 KiB
Python
656 lines
25 KiB
Python
"""Public entry point for the scoped Agent-mode pipeline."""
|
|
|
|
import base64
|
|
import json
|
|
import os
|
|
from typing import Callable, Dict, Optional
|
|
|
|
from backend import config
|
|
from backend.agents.base import AgentScope, AgentUsage
|
|
from backend.agents.brain import BrainAgent
|
|
from backend.agents.code_agent import CodeAgent, build_code_scopes
|
|
from backend.agents.completeness import CompletenessAgent, build_sheet_summaries
|
|
from backend.agents.conflict_critic import ConflictCriticAgent
|
|
from backend.agents.construct_agent import ConstructabilityAgent, build_construct_scopes
|
|
from backend.agents.disputes import annotate_clusters
|
|
from backend.agents.extractors import (
|
|
JurisdictionAgent,
|
|
SheetExtractorAgent,
|
|
SheetIndexAgent,
|
|
)
|
|
from backend.agents.integrity_agent import (
|
|
DrawingIntegrityAgent, build_integrity_scopes,
|
|
)
|
|
from backend.agents.linker import LinkerAgent, build_link_scopes, build_object_graph
|
|
from backend.agents.memory import ProjectMemory
|
|
from backend.agents.orchestrator import Orchestrator
|
|
from backend.agents.rfi_writer import RFIWriterAgent
|
|
from backend.agents.verifier import (
|
|
EvidenceVerifierAgent, apply_verdicts, select_findings,
|
|
)
|
|
from backend.llm import reset_cost
|
|
from backend.pipeline.pdf_processor import convert_pdf_to_images
|
|
from backend.pipeline.report import build_report, to_markdown
|
|
from backend.pipeline.sheet_index import derive_project_meta_from_cover
|
|
from backend.review.gate import build_review_queue
|
|
from backend.review.store import ReviewStore
|
|
from backend.sheet_reconcile import declared_sheet_list, reconcile_sheets
|
|
from backend.text_layer import (
|
|
attach_text_layers, coverage_gaps, find_evidence_bbox, render_crop,
|
|
)
|
|
|
|
|
|
def run_agent_pipeline(
|
|
pdf_path: str,
|
|
out_dir: Optional[str] = None,
|
|
on_stage: Optional[Callable[[str], None]] = None,
|
|
project_input: Optional[Dict] = None,
|
|
source_name: Optional[str] = None,
|
|
require_review: bool = True,
|
|
) -> Dict:
|
|
"""Run all scoped specialist waves and return a Classic-compatible report."""
|
|
if not os.path.isfile(pdf_path):
|
|
raise FileNotFoundError(pdf_path)
|
|
|
|
# Keep the llm module's counters job-local (matches Classic): the job
|
|
# log's failure-path cost estimate in jobs.py reads llm.get_cost().
|
|
reset_cost()
|
|
|
|
agent_dir = os.path.join(out_dir, "agent") if out_dir else None
|
|
memory = ProjectMemory(artifact_dir=agent_dir)
|
|
orchestrator = Orchestrator(memory=memory, on_stage=on_stage)
|
|
usage = AgentUsage()
|
|
orchestrator.initialize()
|
|
|
|
orchestrator.stage("Agent ingest: PDF -> images")
|
|
pages = convert_pdf_to_images(pdf_path)
|
|
page_to_b64 = {page["page_number"]: page["base64"] for page in pages}
|
|
text_dir = os.path.join(agent_dir, "text") if agent_dir else None
|
|
page_words = attach_text_layers(pdf_path, pages, text_dir=text_dir)
|
|
page_to_text = {page["page_number"]: page.get("text_layer") for page in pages}
|
|
|
|
orchestrator.stage("Agent wave 1: extract sheets")
|
|
extract_scopes = [
|
|
AgentScope(
|
|
scope_id=f"sheet:{page['page_number']}",
|
|
payload={"page": page},
|
|
)
|
|
for page in pages
|
|
]
|
|
extract_results = orchestrator.run_scopes(
|
|
SheetExtractorAgent(usage), extract_scopes, config.EXTRACT_CONCURRENCY
|
|
)
|
|
sheets = [
|
|
artifact
|
|
for result in extract_results
|
|
for artifact in result.artifacts
|
|
]
|
|
sheets.sort(key=lambda sheet: sheet.get("page_number") or 0)
|
|
memory.replace("sheets", sheets)
|
|
memory.dump("01-extract.json")
|
|
|
|
# Deterministic reconciliation: the cover sheet's own sheet index
|
|
# declares what the set should contain; compare against what wave 1
|
|
# identified (catches missed sheets AND phantom/misread sheet numbers).
|
|
sheet_recon = reconcile_sheets(sheets, declared_sheet_list(page_to_text))
|
|
if sheet_recon["declared_total"]:
|
|
print(f"[SheetIndex] cover declares {sheet_recon['declared_total']} "
|
|
f"sheets; {sheet_recon['found_total']} identified in set")
|
|
if sheet_recon["declared_not_in_set"]:
|
|
print(f"[SheetIndex] declared but not in set: "
|
|
f"{', '.join(sheet_recon['declared_not_in_set'][:20])}")
|
|
if sheet_recon["in_set_not_declared"]:
|
|
print(f"[SheetIndex] in set but not declared: "
|
|
f"{', '.join(sheet_recon['in_set_not_declared'][:20])}")
|
|
# Coverage signal: text layer present but extraction failed/empty reuses
|
|
# the failed-scopes gap-finding path (finding built below wave 6).
|
|
for gap_page in coverage_gaps(pages, sheets):
|
|
orchestrator.stats.failed_scopes.append(
|
|
f"sheet_extractor:sheet:{gap_page}: extraction gap "
|
|
f"(text layer present, no objects extracted)"
|
|
)
|
|
|
|
cover_meta = derive_project_meta_from_cover(
|
|
sheets, source_name or os.path.basename(pdf_path)
|
|
)
|
|
merged_input = {**cover_meta, **(project_input or {})}
|
|
orchestrator.stage("Agent wave 2: sheet index and jurisdiction")
|
|
index_results = orchestrator.run_scopes(
|
|
SheetIndexAgent(usage),
|
|
[AgentScope("sheet-index", {"sheets": sheets})],
|
|
1,
|
|
)
|
|
jurisdiction_results = orchestrator.run_scopes(
|
|
JurisdictionAgent(usage),
|
|
[AgentScope("jurisdiction", {"project_input": merged_input})],
|
|
1,
|
|
)
|
|
sheet_index = (
|
|
index_results[0].artifacts[0]
|
|
if index_results and index_results[0].artifacts else {}
|
|
)
|
|
jurisdiction = (
|
|
jurisdiction_results[0].artifacts[0]
|
|
if jurisdiction_results and jurisdiction_results[0].artifacts else {}
|
|
)
|
|
memory.replace("sheet_index", sheet_index)
|
|
memory.replace("jurisdiction", jurisdiction)
|
|
memory.dump("02-orient.json")
|
|
|
|
orchestrator.stage("Agent wave 3: scoped semantic linking")
|
|
link_results = orchestrator.run_scopes(
|
|
LinkerAgent(usage),
|
|
build_link_scopes(sheets),
|
|
config.AGENT_LINK_CONCURRENCY,
|
|
)
|
|
clusters = [
|
|
artifact
|
|
for result in link_results
|
|
for artifact in result.artifacts
|
|
][:config.CLUSTER_MAX]
|
|
object_graph = build_object_graph(clusters)
|
|
disputed_count = annotate_clusters(clusters)
|
|
if disputed_count:
|
|
orchestrator.stage(f"[Link] {disputed_count} clusters carry disputed extracted values")
|
|
memory.replace("clusters", clusters)
|
|
memory.replace("object_graph", object_graph)
|
|
memory.dump("03-link.json")
|
|
|
|
orchestrator.stage("Agent wave 4: per-cluster conflict critics")
|
|
conflict_scopes = [
|
|
AgentScope(
|
|
scope_id=f"conflict:{cluster.get('key')}",
|
|
payload={"cluster": cluster, "page_to_b64": page_to_b64},
|
|
)
|
|
for cluster in clusters
|
|
]
|
|
conflict_results = orchestrator.run_scopes(
|
|
ConflictCriticAgent(usage),
|
|
conflict_scopes,
|
|
config.AGENT_CONFLICT_CONCURRENCY,
|
|
)
|
|
conflict_findings = [
|
|
artifact
|
|
for result in conflict_results
|
|
for artifact in result.artifacts
|
|
]
|
|
memory.extend("findings", conflict_findings)
|
|
|
|
orchestrator.stage("Agent wave 5: scoped specialists")
|
|
if config.ENABLE_CODE_REVIEW:
|
|
code_results = orchestrator.run_scopes(
|
|
CodeAgent(usage),
|
|
build_code_scopes(sheets, jurisdiction, sheet_index),
|
|
config.AGENT_SPECIALIST_CONCURRENCY,
|
|
)
|
|
else:
|
|
orchestrator.stage("[wave 5] code/ADA review disabled (ENABLE_CODE_REVIEW=0)")
|
|
code_results = []
|
|
if config.ENABLE_DRAWING_INTEGRITY:
|
|
integrity_results = orchestrator.run_scopes(
|
|
DrawingIntegrityAgent(usage),
|
|
build_integrity_scopes(sheets, page_to_b64, page_to_text),
|
|
config.AGENT_INTEGRITY_CONCURRENCY,
|
|
)
|
|
else:
|
|
integrity_results = []
|
|
construct_results = orchestrator.run_scopes(
|
|
ConstructabilityAgent(usage),
|
|
build_construct_scopes(clusters, conflict_findings),
|
|
config.AGENT_SPECIALIST_CONCURRENCY,
|
|
)
|
|
completeness_scope = AgentScope("completeness", {
|
|
"sheet_index": sheet_index,
|
|
"sheet_summaries": build_sheet_summaries(sheets),
|
|
"cluster_summary": {
|
|
"count": len(clusters),
|
|
"by_kind": _counts(clusters, "kind"),
|
|
},
|
|
})
|
|
completeness_results = orchestrator.run_scopes(
|
|
CompletenessAgent(usage), [completeness_scope], 1
|
|
)
|
|
specialist_findings = [
|
|
artifact
|
|
for result in (code_results + integrity_results
|
|
+ construct_results + completeness_results)
|
|
for artifact in result.artifacts
|
|
]
|
|
|
|
orchestrator.stage("Agent wave 5b: evidence verification")
|
|
sheet_to_page = {str(s.get("sheet_number")): s.get("page_number")
|
|
for s in sheets}
|
|
verify_targets = select_findings(
|
|
specialist_findings, clusters,
|
|
max_checks=config.AGENT_VERIFY_MAX_CHECKS,
|
|
severities=config.AGENT_VERIFY_SEVERITIES,
|
|
)
|
|
target_indexes = {id(f): i for i, f in enumerate(specialist_findings)}
|
|
verify_scopes = _build_verify_scopes(
|
|
verify_targets,
|
|
index_for=lambda f: target_indexes[id(f)],
|
|
sheet_to_page=sheet_to_page, page_to_b64=page_to_b64,
|
|
page_to_text=page_to_text, page_words=page_words, pdf_path=pdf_path,
|
|
)
|
|
verify_results = orchestrator.run_scopes(
|
|
EvidenceVerifierAgent(usage), verify_scopes, config.AGENT_VERIFY_CONCURRENCY)
|
|
suppressed = apply_verdicts(specialist_findings, verify_results)
|
|
if suppressed:
|
|
suppressed_ids = {id(f) for f in suppressed}
|
|
specialist_findings = [f for f in specialist_findings if id(f) not in suppressed_ids]
|
|
memory.replace("suppressed", suppressed)
|
|
|
|
memory.extend("findings", specialist_findings)
|
|
gap_findings = [
|
|
{
|
|
"issue_id": f"AGENT-GAP-{index + 1:03d}",
|
|
"source_stage": "qaqc",
|
|
"category": "analysis_gap",
|
|
"severity": "low",
|
|
"confidence": "high",
|
|
"location": failed_scope.split(":", 2)[1] if ":" in failed_scope else "",
|
|
"disciplines": [],
|
|
"sheets": [],
|
|
"description": f"Agent analysis scope did not complete: {failed_scope}",
|
|
"evidence": [],
|
|
"recommended_resolution": "Review this scope manually or rerun the job.",
|
|
"code_reference": None,
|
|
"agent": "completeness",
|
|
"scope_id": "failed-scopes",
|
|
}
|
|
for index, failed_scope in enumerate(orchestrator.stats.failed_scopes)
|
|
]
|
|
memory.extend("findings", gap_findings)
|
|
memory.dump("05-specialists.json")
|
|
|
|
orchestrator.stage("Agent wave 6: Brain merge, judge, prioritize")
|
|
all_findings = memory.snapshot()["findings"]
|
|
if all_findings:
|
|
prioritized, decisions = BrainAgent(usage).run(
|
|
all_findings, sheet_index, jurisdiction
|
|
)
|
|
else:
|
|
prioritized, decisions = [], []
|
|
memory.extend("decisions", decisions)
|
|
orchestrator.stats.merges = sum(
|
|
1 for decision in decisions if decision.get("action") == "merged"
|
|
)
|
|
|
|
# Wave 6.5 — Brain-directed clarification (bounded hub-and-spoke). The Brain
|
|
# names kept findings it is unsure about; verify_evidence requests route
|
|
# back through the wave-5b verifier (fresh images + text-layer oracle).
|
|
# Refuted findings are demoted, dropped from `prioritized`, and moved to
|
|
# memory["suppressed"]. One planning call, one bounded verify wave, no loop.
|
|
if config.ENABLE_BRAIN_CLARIFY and prioritized:
|
|
prioritized = _brain_clarification_pass(
|
|
orchestrator, usage, memory, prioritized,
|
|
sheet_to_page=sheet_to_page, page_to_b64=page_to_b64,
|
|
page_to_text=page_to_text, page_words=page_words, pdf_path=pdf_path,
|
|
)
|
|
|
|
if require_review:
|
|
orchestrator.stage("Agent review gate: build human-review queue")
|
|
memory_snapshot = memory.snapshot()
|
|
queue = build_review_queue(memory_snapshot, prioritized, decisions,
|
|
limit=config.AGENT_REVIEW_AUDIT_SAMPLE)
|
|
store = ReviewStore(out_dir)
|
|
store.write_queue(queue)
|
|
candidate_conflicts = [_finding_as_conflict(item) for item in conflict_findings]
|
|
report = build_report(
|
|
conflicts=candidate_conflicts,
|
|
sheets=sheets,
|
|
clusters=clusters,
|
|
source=source_name or os.path.basename(pdf_path),
|
|
)
|
|
report.update({
|
|
"project_input": merged_input,
|
|
"jurisdiction": jurisdiction,
|
|
"sheet_index": sheet_index,
|
|
"sheet_reconciliation": sheet_recon,
|
|
"project_intelligence": object_graph,
|
|
"validated_issues": prioritized,
|
|
"rfis": [],
|
|
"suppressed_issues": memory.snapshot().get("suppressed") or [],
|
|
})
|
|
progress = store.progress(queue)
|
|
# Same usage/stats summary block as the wave-7 path (rfis: 0 — they
|
|
# are drafted only after human review finalizes the run).
|
|
cost = usage.snapshot()
|
|
orchestrator.stats.calls = cost["calls"]
|
|
stats = orchestrator.stats.as_dict()
|
|
report["summary"].update({
|
|
"pipeline_mode": "agent",
|
|
"agent_status": "needs_review",
|
|
"review": progress,
|
|
"agent_stats": stats,
|
|
"by_stage": {
|
|
"conflicts": len(conflict_findings),
|
|
"qaqc": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "qaqc"
|
|
),
|
|
"code": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "code"
|
|
),
|
|
"drawing_integrity": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "drawing_integrity"
|
|
),
|
|
"constructability": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "constructability"
|
|
),
|
|
"validated": len(prioritized),
|
|
"rfis": 0,
|
|
},
|
|
"cost_usd": round(cost["usd"], 4),
|
|
"llm_calls": cost["calls"],
|
|
"cached_calls": cost["cached"],
|
|
"cost_by_stage": cost["by_stage"],
|
|
"text_backend": "openrouter",
|
|
"models_used": cost["models"],
|
|
})
|
|
if out_dir:
|
|
_dump(out_dir, "conflicts.json", report)
|
|
_dump(out_dir, "validated_issues.json", prioritized)
|
|
# Snapshot for the review finalizer's targeted clarification reruns.
|
|
memory.dump("memory.json")
|
|
return report
|
|
|
|
orchestrator.stage("Agent wave 7: per-finding RFI writers")
|
|
rfi_scopes = [
|
|
AgentScope(
|
|
scope_id=f"rfi:{finding.get('issue_id') or index + 1}",
|
|
payload={"finding": finding},
|
|
)
|
|
for index, finding in enumerate(prioritized)
|
|
]
|
|
rfi_results = orchestrator.run_scopes(
|
|
RFIWriterAgent(usage), rfi_scopes, config.AGENT_RFI_CONCURRENCY
|
|
)
|
|
rfis = [
|
|
artifact for result in rfi_results for artifact in result.artifacts
|
|
]
|
|
memory.extend("rfis", rfis)
|
|
memory.dump("memory.json")
|
|
|
|
orchestrator.stage("Build agent report")
|
|
conflicts = [_finding_as_conflict(item) for item in conflict_findings]
|
|
report = build_report(
|
|
conflicts=conflicts,
|
|
sheets=sheets,
|
|
clusters=clusters,
|
|
source=source_name or os.path.basename(pdf_path),
|
|
)
|
|
report.update({
|
|
"project_input": merged_input,
|
|
"jurisdiction": jurisdiction,
|
|
"sheet_index": sheet_index,
|
|
"sheet_reconciliation": sheet_recon,
|
|
"project_intelligence": object_graph,
|
|
"validated_issues": prioritized,
|
|
"rfis": rfis,
|
|
"suppressed_issues": memory.snapshot().get("suppressed") or [],
|
|
})
|
|
cost = usage.snapshot()
|
|
orchestrator.stats.calls = cost["calls"]
|
|
stats = orchestrator.stats.as_dict()
|
|
report["summary"].update({
|
|
"pipeline_mode": "agent",
|
|
"agent_status": "complete",
|
|
"agent_stats": stats,
|
|
"by_stage": {
|
|
"conflicts": len(conflict_findings),
|
|
"qaqc": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "qaqc"
|
|
),
|
|
"code": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "code"
|
|
),
|
|
"drawing_integrity": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "drawing_integrity"
|
|
),
|
|
"constructability": sum(
|
|
1 for item in specialist_findings
|
|
if item.get("source_stage") == "constructability"
|
|
),
|
|
"validated": len(prioritized),
|
|
"rfis": len(rfis),
|
|
},
|
|
"cost_usd": round(cost["usd"], 4),
|
|
"llm_calls": cost["calls"],
|
|
"cached_calls": cost["cached"],
|
|
"cost_by_stage": cost["by_stage"],
|
|
"text_backend": "openrouter",
|
|
"models_used": cost["models"],
|
|
})
|
|
|
|
if out_dir:
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
_dump(out_dir, "assertions.json", sheets)
|
|
_dump(out_dir, "clusters.json", [_without_base64(item) for item in clusters])
|
|
_dump(out_dir, "sheet_index.json", sheet_index)
|
|
_dump(out_dir, "jurisdiction.json", jurisdiction)
|
|
_dump(out_dir, "project_intelligence.json", object_graph)
|
|
_dump(out_dir, "validated_issues.json", prioritized)
|
|
_dump(out_dir, "rfis.json", rfis)
|
|
_dump(out_dir, "conflicts.json", report)
|
|
with open(os.path.join(out_dir, "report.md"), "w", encoding="utf-8") as f:
|
|
f.write(to_markdown(report))
|
|
|
|
return report
|
|
|
|
|
|
def _dump(out_dir: str, name: str, value) -> None:
|
|
with open(os.path.join(out_dir, name), "w", encoding="utf-8") as f:
|
|
json.dump(value, f, indent=2)
|
|
|
|
|
|
def _brain_clarification_pass(
|
|
orchestrator,
|
|
usage,
|
|
memory,
|
|
prioritized,
|
|
sheet_to_page,
|
|
page_to_b64,
|
|
page_to_text,
|
|
page_words,
|
|
pdf_path,
|
|
):
|
|
"""Wave 6.5: let the Brain request targeted clarifications, execute the
|
|
verify_evidence ones through the wave-5b verifier, and prune refuted
|
|
findings out of `prioritized` into memory["suppressed"].
|
|
|
|
Bounded and non-looping: one Brain planning call, at most
|
|
BRAIN_CLARIFY_MAX_REQUESTS verifications, a single pass. Returns the
|
|
(possibly shortened) prioritized list. Any request type other than
|
|
verify_evidence is logged as planned-but-not-executed and left untouched.
|
|
"""
|
|
requests = BrainAgent(usage).plan_clarifications(prioritized)
|
|
if not requests:
|
|
return prioritized
|
|
by_id = {f.get("issue_id"): f for f in prioritized}
|
|
verify_findings = []
|
|
unsupported = 0
|
|
for req in requests:
|
|
if req.get("request_type") != "verify_evidence":
|
|
unsupported += 1
|
|
continue
|
|
finding = by_id.get(req.get("issue_id"))
|
|
if finding is not None and finding not in verify_findings:
|
|
verify_findings.append(finding)
|
|
orchestrator.stage(
|
|
f"Agent wave 6.5: Brain-directed clarification "
|
|
f"({len(verify_findings)} verify, {unsupported} other)"
|
|
)
|
|
if unsupported:
|
|
for req in requests:
|
|
if req.get("request_type") != "verify_evidence":
|
|
orchestrator.stats.failed_scopes.append(
|
|
f"brain_clarify:{req.get('issue_id')}: "
|
|
f"request_type '{req.get('request_type')}' planned, "
|
|
f"not executed (v1 supports verify_evidence only)"
|
|
)
|
|
if not verify_findings:
|
|
return prioritized
|
|
index_of = {id(f): i for i, f in enumerate(prioritized)}
|
|
verify_scopes = _build_verify_scopes(
|
|
verify_findings,
|
|
index_for=lambda f: index_of[id(f)],
|
|
sheet_to_page=sheet_to_page, page_to_b64=page_to_b64,
|
|
page_to_text=page_to_text, page_words=page_words, pdf_path=pdf_path,
|
|
)
|
|
if not verify_scopes:
|
|
return prioritized
|
|
verify_results = orchestrator.run_scopes(
|
|
EvidenceVerifierAgent(usage), verify_scopes,
|
|
config.AGENT_VERIFY_CONCURRENCY,
|
|
)
|
|
suppressed = apply_verdicts(prioritized, verify_results)
|
|
if suppressed:
|
|
suppressed_ids = {id(f) for f in suppressed}
|
|
prioritized = [f for f in prioritized if id(f) not in suppressed_ids]
|
|
existing = memory.snapshot().get("suppressed") or []
|
|
memory.replace("suppressed", existing + suppressed)
|
|
memory.extend("decisions", [
|
|
{
|
|
"finding_refs": [f.get("issue_id")],
|
|
"action": "dropped",
|
|
"reason": "Brain-directed clarification: evidence refuted on re-check",
|
|
"kept_issue_id": None,
|
|
}
|
|
for f in suppressed
|
|
])
|
|
return prioritized
|
|
|
|
|
|
def _build_verify_scopes(
|
|
targets,
|
|
index_for,
|
|
sheet_to_page,
|
|
page_to_b64,
|
|
page_to_text,
|
|
page_words,
|
|
pdf_path,
|
|
):
|
|
"""Build EvidenceVerifierAgent scopes for a set of findings.
|
|
|
|
Shared by wave 5b (severity-gated) and wave 6.5 (Brain-directed): each
|
|
finding's cited sheets are mapped to page images (hi-DPI evidence crops
|
|
when enabled, else full pages) plus a capped text-layer oracle. Findings
|
|
whose sheets resolve to NO loadable image are skipped (I2 guard) — never
|
|
judge evidence against images we could not load. index_for(finding) yields
|
|
the finding_index the verifier echoes back for apply_verdicts alignment.
|
|
"""
|
|
scopes = []
|
|
for finding in targets:
|
|
cited_pages = [
|
|
sheet_to_page[str(name)]
|
|
for name in (finding.get("sheets") or [])
|
|
if sheet_to_page.get(str(name)) in page_to_b64
|
|
]
|
|
images = [
|
|
page_to_b64[p]
|
|
for p in cited_pages[:config.AGENT_CONFLICT_MAX_IMAGES]
|
|
]
|
|
if not images:
|
|
continue # never judge evidence against images we could not load
|
|
# Text oracle: concatenated text layer of the cited sheets, capped.
|
|
excerpt = "\n\n".join(
|
|
f"--- Page {p} ---\n{page_to_text[p]}"
|
|
for p in cited_pages
|
|
if page_to_text.get(p)
|
|
)[:config.VERIFY_TEXT_MAX_CHARS]
|
|
if config.VERIFY_HI_DPI_CROPS:
|
|
images = _evidence_crops(finding, cited_pages, sheet_to_page,
|
|
page_words, page_to_b64, pdf_path,
|
|
fallback=images)
|
|
finding_index = index_for(finding)
|
|
scopes.append(AgentScope(
|
|
scope_id=f"verify:{finding_index}",
|
|
payload={
|
|
"finding_index": finding_index,
|
|
"finding": finding,
|
|
"images_b64": images,
|
|
"text_layer_excerpt": excerpt,
|
|
},
|
|
))
|
|
return scopes
|
|
|
|
|
|
def _evidence_crops(
|
|
finding: Dict,
|
|
cited_pages: list,
|
|
sheet_to_page: Dict,
|
|
page_words: Dict,
|
|
page_to_b64: Dict,
|
|
pdf_path: str,
|
|
fallback: list,
|
|
) -> list:
|
|
"""High-DPI crops around each evidence item's source_text, located via the
|
|
page text layer. Crops REPLACE full-page images when at least one evidence
|
|
location resolves confidently; otherwise the full-page fallback is kept.
|
|
Never returns an empty list when fallback is non-empty (I2 guard)."""
|
|
crops: list = []
|
|
for item in finding.get("evidence") or []:
|
|
if len(crops) >= config.AGENT_CONFLICT_MAX_IMAGES:
|
|
break
|
|
if not isinstance(item, dict):
|
|
continue
|
|
source_text = item.get("source_text") or ""
|
|
if not source_text:
|
|
continue
|
|
# Prefer the page named on the evidence item, then any cited page.
|
|
candidates = []
|
|
named_page = sheet_to_page.get(str(item.get("sheet") or ""))
|
|
if named_page in cited_pages:
|
|
candidates.append(named_page)
|
|
candidates.extend(p for p in cited_pages if p not in candidates)
|
|
for page in candidates:
|
|
bbox = find_evidence_bbox(page_words.get(page) or [], source_text)
|
|
if bbox is None:
|
|
continue
|
|
crop = render_crop(pdf_path, page, bbox)
|
|
if not crop:
|
|
continue
|
|
crops.append(base64.b64encode(crop).decode("utf-8"))
|
|
break
|
|
return crops or fallback
|
|
|
|
|
|
def _counts(items, key: str) -> Dict[str, int]:
|
|
counts: Dict[str, int] = {}
|
|
for item in items:
|
|
value = str(item.get(key) or "unknown")
|
|
counts[value] = counts.get(value, 0) + 1
|
|
return counts
|
|
|
|
|
|
def _finding_as_conflict(finding: Dict) -> Dict:
|
|
return {
|
|
"category": finding.get("category") or "uncategorized",
|
|
"severity": finding.get("severity") or "medium",
|
|
"disciplines": finding.get("disciplines") or [],
|
|
"location": finding.get("location") or "",
|
|
"sheets": finding.get("sheets") or [],
|
|
"description": finding.get("description") or "",
|
|
"evidence": finding.get("evidence") or [],
|
|
"recommended_resolution": finding.get("recommended_resolution") or "",
|
|
"confidence": finding.get("confidence") or "medium",
|
|
"cluster_key": finding.get("scope_id"),
|
|
}
|
|
|
|
|
|
def _without_base64(cluster: Dict) -> Dict:
|
|
return {
|
|
**cluster,
|
|
"assertions": [
|
|
{key: value for key, value in assertion.items() if key != "base64"}
|
|
for assertion in cluster.get("assertions") or []
|
|
],
|
|
}
|