diff --git a/composer/pipeline/core.py b/composer/pipeline/core.py
index 97ce1b0b..cdc15598 100644
--- a/composer/pipeline/core.py
+++ b/composer/pipeline/core.py
@@ -36,7 +36,7 @@
from composer.spec.util import string_hash
from composer.input.files import Document
from composer.spec.source.report.build import build_report
-from composer.spec.source.report.collect import ReportComponentInput, Verdict
+from composer.spec.source.report.collect import EvidenceFetcher, ReportComponentInput, Verdict
from composer.spec.source.report.schema import RuleName, ReportBackend
from composer.spec.source.report import build as report_build
from composer.spec.source.task_ids import SYSTEM_ANALYSIS_TASK_ID, REPORT_TASK_ID
@@ -55,7 +55,7 @@ class Formalizer[FormT: BackendResult](ABC):
state — never set post-hoc. `FormT: ReportableResult` is what makes the report a core step."""
formalized_type: type[FormT]
backend_tag: ReportBackend
-
+
@abstractmethod
async def formalize(
self,
@@ -77,6 +77,12 @@ async def fetch_verdicts(self, inp: ReportComponentInput[FormT]) -> dict[RuleNam
off-thread. Foundry: read straight off inp.formalized.result."""
...
+ def findings_evidence(self) -> EvidenceFetcher | None:
+ """The per-rule evidence source for findings synthesis, or None if this backend produces no
+ findings. Returning None is how a backend opts out — the report then builds no findings for
+ it, with no backend-specific branching in the report layer. Default: None."""
+ return None
+
async def finalize(self, outcomes: list[ComponentOutcome[FormT]], run: PipelineRun) -> None:
"""Emit any backend-specific run-level artifacts from the full outcome set (prover:
components_to_prover_runs.json). Default: none."""
@@ -209,7 +215,7 @@ async def _run(batch: _Batch) -> ComponentOutcome[FormT]:
await child.cache_put(result)
else:
result = cached_result
-
+
outcome: Delivered[FormT] | GaveUp = (
result if isinstance(result, GaveUp)
else Delivered(result, backend.artifact_store.write_artifact(result_key, result))
@@ -234,11 +240,15 @@ async def _run(batch: _Batch) -> ComponentOutcome[FormT]:
)
for o in outcomes
] + formalizer.extra_report_inputs()
+ findings_evidence = formalizer.findings_evidence()
try:
report = await run.runner(
job=lambda: build_report(
contract_name=source.contract_name, backend=formalizer.backend_tag,
components=inputs, llm=run.env.llm_lite(), fetch_verdicts=formalizer.fetch_verdicts,
+ # Findings only when the backend supplies evidence — skip the heavy model otherwise.
+ findings_llm=run.env.llm_heavy() if findings_evidence else None,
+ fetch_evidence=findings_evidence,
),
task_info=TaskInfo(REPORT_TASK_ID, label="Report Extraction", phase=backend.core_phases["report"])
)
diff --git a/composer/spec/source/autoprove_common.py b/composer/spec/source/autoprove_common.py
index 24eb0a7e..c7a312be 100644
--- a/composer/spec/source/autoprove_common.py
+++ b/composer/spec/source/autoprove_common.py
@@ -20,6 +20,7 @@
)
from composer.pipeline.cli import cli_pipeline, user_ns
from composer.spec.source.pipeline import ProverBackend, GeneratedCVL
+from composer.spec.source.cex_capture import CexAnalysisStore
from composer.prover.core import make_prover_options
from composer.spec.source.source_env import build_source_env
from composer.spec.source.artifacts import ProverArtifactStore
@@ -115,7 +116,7 @@ async def exit_logger(
async def callback(
handler: HandlerFactory[AutoProvePhase, None]
- ) -> CorePipelineResult[GeneratedCVL]:
+ ) -> CorePipelineResult[GeneratedCVL]:
async with (
cli_pipeline(
args=args, design_doc_phase=design_phase,
@@ -144,7 +145,8 @@ async def callback(
)
backend = ProverBackend(
ProverArtifactStore(staged.source.project_root, staged.source.contract_name),
- make_prover_options(cloud=args.cloud)
+ make_prover_options(cloud=args.cloud),
+ CexAnalysisStore(store=staged.conns.store, namespace=("cex_analyses", thread_id)),
)
return await cont(source_env, backend)
yield callback
diff --git a/composer/spec/source/cex_capture.py b/composer/spec/source/cex_capture.py
new file mode 100644
index 00000000..56242713
--- /dev/null
+++ b/composer/spec/source/cex_capture.py
@@ -0,0 +1,48 @@
+"""Run-scoped capture of per-rule counterexample analysis.
+
+The autoprove prover tool runs an LLM analysis of every violated rule during the run; that text is
+otherwise consumed only as agent feedback and discarded. Capture happens through the prover callback
+(``on_analysis_complete``), which fires regardless of which CEX handler is in use, so nothing here
+assumes a particular handler. The analyses are persisted to the run's ``BaseStore`` (cf.
+``composer.prover.report_store``) so the report phase can reshape the *final* iteration's analysis
+into a finding.
+"""
+from dataclasses import dataclass
+
+from langgraph.store.base import BaseStore
+from pydantic import BaseModel
+
+
+class CexAnalysis(BaseModel):
+ """One rule's captured counterexample analysis: the root-cause / fix explanation and, when
+ available, the counterexample call-trace dump it was derived from."""
+ analysis: str
+ counterexample: str | None = None
+
+
+@dataclass(frozen=True)
+class CexAnalysisStore:
+ """Typed wrapper over a ``BaseStore`` for per-rule `CexAnalysis` capture. Construct once per run
+ with the run's store and a PER-RUN ``namespace`` — rule names are not unique across runs, so the
+ namespace (not the key) provides run isolation — then thread the same wrapper to the prover tool
+ (write side) and the report phase (read side).
+
+ Keyed by the bare rule name, last-write-wins across prover iterations: a rule that stays violated
+ to the end was analyzed on the final run so its final analysis wins, while a rule fixed earlier is
+ GOOD in the report and its (stale) analysis is never looked up."""
+ store: BaseStore
+ namespace: tuple[str, ...]
+
+ async def record(self, rule_name: str, analysis: str, counterexample: str | None = None) -> None:
+ """Store one rule's analysis under ``rule_name``, which must be the bare top-level rule name
+ (``RulePath.rule``) — exactly what the report's ``RuleVerdict.name`` carries (POU's
+ ``context[0]``) — so the findings join is an exact match."""
+ await self.store.aput(
+ self.namespace,
+ rule_name,
+ CexAnalysis(analysis=analysis, counterexample=counterexample).model_dump(mode="json"),
+ )
+
+ async def get(self, rule_name: str) -> CexAnalysis | None:
+ item = await self.store.aget(self.namespace, rule_name)
+ return CexAnalysis.model_validate(item.value) if item is not None else None
diff --git a/composer/spec/source/pipeline.py b/composer/spec/source/pipeline.py
index c6ac5120..9aa1bdbf 100644
--- a/composer/spec/source/pipeline.py
+++ b/composer/spec/source/pipeline.py
@@ -46,8 +46,11 @@
from composer.spec.source.author import batch_cvl_generation
from composer.spec.source.artifacts import ProverArtifactStore, ComponentSpec, InvariantSpec
from composer.spec.source.report_prover import make_prover_fetcher
-from composer.spec.source.report.collect import ReportComponentInput, Verdict, VerdictFetcher
+from composer.spec.source.report.collect import (
+ EvidenceFetcher, ReportComponentInput, RuleEvidence, Verdict, VerdictFetcher,
+)
from composer.spec.source.report.schema import RuleName
+from composer.spec.source.cex_capture import CexAnalysisStore
from composer.spec.source.task_ids import (
HARNESS_TASK_ID, AUTOSETUP_TASK_ID, SUMMARIES_TASK_ID,
INVARIANTS_TASK_ID, INVARIANT_CVL_TASK_ID,
@@ -108,6 +111,7 @@ class ProverRunner(Formalizer[GeneratedCVL]):
_resources: list[CVLResource]
_invariant: tuple[list[PropertyFormulation], Delivered[GeneratedCVL]] | None
_fetch: VerdictFetcher[GeneratedCVL]
+ _analysis_store: CexAnalysisStore
@override
async def formalize(
@@ -148,6 +152,20 @@ async def fetch_verdicts(
) -> dict[RuleName, Verdict]:
return await self._fetch(inp)
+ @override
+ def findings_evidence(self) -> EvidenceFetcher | None:
+ # Prover runs capture per-rule counterexample analysis, so this backend opts into findings.
+ return self._fetch_evidence
+
+ async def _fetch_evidence(self, rule_name: str) -> RuleEvidence | None:
+ # The run captured each violated rule's counterexample analysis as it happened; hand the
+ # final-iteration analysis to the findings synthesizer. None -> the finding degrades to
+ # property/group text.
+ rec = await self._analysis_store.get(rule_name)
+ if rec is None:
+ return None
+ return RuleEvidence(analysis=rec.analysis, counterexample=rec.counterexample)
+
@override
async def finalize(self, outcomes: list[ComponentOutcome[GeneratedCVL]], run: PipelineRun) -> None:
# components_to_prover_runs.json: {run_key (slug): prover /output/ link}.
@@ -173,6 +191,7 @@ class ProverPrepared(PreparedSystem[GeneratedCVL]):
_prover_tool: BaseTool
_prover_opts: ProverOptions
_analyzed: SourceApplication
+ _analysis_store: CexAnalysisStore
@override
async def prepare_formalization(self, run: PipelineRun) -> Formalizer[GeneratedCVL]:
@@ -233,7 +252,7 @@ async def prepare_formalization(self, run: PipelineRun) -> Formalizer[GeneratedC
return ProverRunner(
GeneratedCVL, "prover",
self._store, self._prover_tool, setup_config.prover_config, resources, invariant,
- make_prover_fetcher(),
+ make_prover_fetcher(), self._analysis_store,
)
async def _autosetup(self, run: PipelineRun) -> tuple[SetupSuccess, list[CVLResource]]:
@@ -284,6 +303,7 @@ class ProverBackend:
artifact_store: ProverArtifactStore
_prover_opts: ProverOptions
+ analysis_store: CexAnalysisStore
async def prepare_system(
self, analyzed: SourceApplication, run: PipelineRun[AutoProvePhase, None],
@@ -295,12 +315,12 @@ async def prepare_system(
harnessed = _lift_harnessed(analyzed, sys_desc)
prover_tool = get_prover_tool(
run.env.llm_heavy(), run.source.contract_name, run.source.project_root,
- prover_opts=self._prover_opts,
+ prover_opts=self._prover_opts, analysis_store=self.analysis_store,
)
return ProverPrepared(
main_instance(harnessed, run.source),
self.artifact_store, sys_desc, harnessed, prover_tool,
- self._prover_opts, analyzed,
+ self._prover_opts, analyzed, self.analysis_store,
)
def to_artifact_id(self, c: ContractComponentInstance) -> ComponentSpec:
diff --git a/composer/spec/source/prover.py b/composer/spec/source/prover.py
index f6715372..10e0c1ef 100644
--- a/composer/spec/source/prover.py
+++ b/composer/spec/source/prover.py
@@ -40,6 +40,7 @@
from graphcore.graph import tool_state_update
from composer.spec.util import temp_certora_file
from composer.spec.gen_types import CERTORA_DIR, SPECS_DIR
+from composer.spec.source.cex_capture import CexAnalysisStore
_logger = logging.getLogger("composer.prover")
@@ -99,12 +100,14 @@ def __init__(
tool_call_id: str,
summary: RunSummary,
config: dict,
+ analysis_store: CexAnalysisStore | None = None,
) -> None:
super().__init__(writer, tool_call_id)
self._writer = writer
self._tool_call_id = tool_call_id
self._summary = summary
self._config = config
+ self._analysis_store = analysis_store
self._started_mono: float | None = None
@override
@@ -158,6 +161,19 @@ async def on_prover_result(self, results: dict[str, RuleResult]) -> None:
results
)
+ @override
+ async def on_analysis_complete(self, rule: RuleResult, explanation: str) -> None:
+ # Capture the per-rule counterexample analysis (last-write-wins across iterations) so the
+ # report phase can reshape the final iteration's analysis into a finding without re-running
+ # the analysis. Key by the bare rule name (rule.path.rule).
+ # Never let a capture error disturb the run.
+ if self._analysis_store is not None:
+ try:
+ await self._analysis_store.record(rule.path.rule, explanation, rule.cex_dump)
+ except Exception:
+ _logger.exception("failed to capture cex analysis for %s", rule.name)
+ await super().on_analysis_complete(rule, explanation)
+
class VerifySpecSchema(BaseModel):
"""
@@ -216,6 +232,7 @@ def get_prover_tool(
main_contract: str,
project_root: str,
prover_opts: ProverOptions,
+ analysis_store: CexAnalysisStore | None = None,
) -> BaseTool:
sem = _prover_sem(prover_opts.cloud)
stamper = make_validation_stamper(VALIDATION_KEY)
@@ -268,7 +285,8 @@ async def verify_spec(
[config_path],
tool_call_id,
prover_opts,
- _SpecCallbacks(get_stream_writer(), tool_call_id, summary, config),
+ _SpecCallbacks(get_stream_writer(), tool_call_id, summary, config,
+ analysis_store=analysis_store),
DefaultCexHandler(llm, state, summarization_threshold=10)
)
diff --git a/composer/spec/source/report/build.py b/composer/spec/source/report/build.py
index 70f1ec64..a9df416a 100644
--- a/composer/spec/source/report/build.py
+++ b/composer/spec/source/report/build.py
@@ -13,14 +13,15 @@
from langchain_core.language_models.chat_models import BaseChatModel
from composer.spec.source.report.collect import (
- ReportableResult, ReportComponentInput, VerdictFetcher, collect,
+ EvidenceFetcher, ReportableResult, ReportComponentInput, VerdictFetcher, collect,
)
from composer.spec.source.report.coverage import ValidationError, validate
+from composer.spec.source.report.findings import build_findings
from composer.spec.source.report.grouping import (
build_fallback_grouping, build_groups, call_grouping_llm,
)
from composer.spec.source.report.schema import (
- AutoProverReport, Outcome, PropertyKey, ReportBackend, RuleRef,
+ AutoProverReport, Finding, Outcome, PropertyKey, ReportBackend, RuleRef,
)
_log = logging.getLogger(__name__)
@@ -42,8 +43,15 @@ async def build_report[R: ReportableResult](
components: list[ReportComponentInput[R]],
llm: BaseChatModel,
fetch_verdicts: VerdictFetcher[R],
+ findings_llm: BaseChatModel | None = None,
+ fetch_evidence: EvidenceFetcher | None = None,
) -> AutoProverReport:
- """Build and return the in-memory `AutoProverReport`. Persistence is the caller's job."""
+ """Build and return the in-memory `AutoProverReport`. Persistence is the caller's job.
+
+ When ``findings_llm`` is supplied, violated rules are additionally synthesized into
+ audit-issue `Finding`s (best-effort; a synthesis failure yields no findings
+ rather than failing the report). ``fetch_evidence`` supplies each violation's captured
+ counterexample analysis; it is optional."""
properties, rules, skipped, gave_up, dropped = await collect(
components, fetch_verdicts=fetch_verdicts
)
@@ -84,6 +92,21 @@ async def build_report[R: ReportableResult](
)
coverage.warnings = ["FALLBACK GROUPING APPLIED"] + coverage.warnings
+ # Violated rules -> findings. Its own guard: findings synthesis must never fail the report
+ # (the whole phase is also best-effort in the caller, but this keeps a working report even when
+ # only findings break).
+ findings: list[Finding] = []
+ if findings_llm is not None:
+ try:
+ findings = await build_findings(
+ contract_name=contract_name, rules=rules, properties=properties, groups=groups,
+ fetch_evidence=fetch_evidence, llm=findings_llm,
+ )
+ except Exception as e: # noqa: BLE001
+ if RERAISE_REPORT_FAILURES:
+ raise
+ _log.warning("report: findings synthesis failed (%s); continuing without findings", e)
+
report = AutoProverReport(
backend=backend,
contract_name=contract_name,
@@ -96,5 +119,6 @@ async def build_report[R: ReportableResult](
skipped=skipped,
gave_up_components=gave_up,
coverage=coverage,
+ findings=findings,
)
return report
diff --git a/composer/spec/source/report/collect.py b/composer/spec/source/report/collect.py
index ddff5b8e..b7304bbe 100644
--- a/composer/spec/source/report/collect.py
+++ b/composer/spec/source/report/collect.py
@@ -101,6 +101,25 @@ async def __call__(self, input: ReportComponentInput[R], /) -> dict[RuleName, Ve
...
+@dataclass(frozen=True)
+class RuleEvidence:
+ """Backend-supplied raw material for synthesizing a finding from a violated rule.
+
+ ``analysis`` is the backend's pre-computed root-cause / fix explanation for the violation
+ (prover: the ``analyze_cex_raw`` output captured during the run); ``counterexample`` is a concrete
+ failing trace excerpt (prover: the rule's ``cex_dump``). Both optional — a finding degrades to
+ property/group text when absent."""
+ analysis: str | None = None
+ counterexample: str | None = None
+
+
+class EvidenceFetcher(Protocol):
+ """Backend hook: given a violated rule's name, return its `RuleEvidence`, or ``None`` when the
+ backend has no evidence for that rule. Prover reads the run-scoped CEX-analysis capture."""
+ async def __call__(self, rule_name: str, /) -> RuleEvidence | None:
+ ...
+
+
async def collect[R: ReportableResult](
inputs: list[ReportComponentInput[R]],
*,
diff --git a/composer/spec/source/report/findings.py b/composer/spec/source/report/findings.py
new file mode 100644
index 00000000..3867ae25
--- /dev/null
+++ b/composer/spec/source/report/findings.py
@@ -0,0 +1,159 @@
+"""Synthesize findings from violated rules.
+
+For each violated rule (a `RuleVerdict` with ``outcome == Outcome.BAD``) this asks an LLM to write up
+the issue, grounded in the rule's counterexample analysis (captured during the run and looked up via
+the backend's `EvidenceFetcher`) and the properties the rule formalizes. The model assesses the impact
+and likelihood; this module computes the severity from them via a fixed matrix — the LLM never picks a
+severity directly. Findings are produced only when the backend supplies an evidence fetcher; each
+finding is best-effort, so a synthesis failure drops that one finding rather than the report.
+"""
+import asyncio
+import logging
+
+from langchain_core.language_models.chat_models import BaseChatModel
+from langchain_core.messages import HumanMessage, SystemMessage
+from pydantic import Field
+
+from composer.templates.loader import load_jinja_template
+from composer.spec.source.report.collect import EvidenceFetcher, RuleEvidence
+from composer.spec.source.report.schema import (
+ AuthoredContent, Finding, FindingProvenance, FormalizedProperty, ImpactLevel, IssueContent,
+ LikelihoodLevel, Outcome, PropertyGroup, PropertyKey, RuleRef, RuleVerdict, SeverityTier,
+)
+
+_log = logging.getLogger(__name__)
+
+#: Cap on concurrent findings-synthesis LLM calls (one per violated rule), so a violation-heavy run
+#: doesn't burst dozens of heavy-model requests at once (which rate limits would turn into dropped
+#: findings).
+_MAX_CONCURRENT_FINDING_CALLS = 8
+
+# Impact × Likelihood -> severity. ``none`` impact (no real-world exploit path) is informational,
+# handled in `severity_for` rather than the table. Low impact caps at ``low`` regardless of likelihood:
+# a trivially-triggerable but low-impact break is still only low (a cosmetic issue shouldn't read
+# medium just because anyone can hit it).
+_SEVERITY_MATRIX: dict[tuple[ImpactLevel, LikelihoodLevel], SeverityTier] = {
+ ("high", "high"): "critical", ("high", "medium"): "high", ("high", "low"): "medium",
+ ("medium", "high"): "high", ("medium", "medium"): "medium", ("medium", "low"): "low",
+ ("low", "high"): "low", ("low", "medium"): "low", ("low", "low"): "low",
+}
+
+
+def severity_for(impact: ImpactLevel, likelihood: LikelihoodLevel) -> SeverityTier:
+ """Map an assessed (impact, likelihood) pair to a severity. ``none`` impact -> informational."""
+ if impact == "none":
+ return "informational"
+ return _SEVERITY_MATRIX[(impact, likelihood)]
+
+
+class FindingDraft(AuthoredContent):
+ """Write up a single confirmed vulnerability finding for a formal-verification rule that the
+ Certora Prover refuted with a concrete counterexample."""
+ title: str = Field(description="A one-line title naming the specific broken guarantee.")
+ impact_level: ImpactLevel = Field(description=(
+ "How severe the consequence is if exploited: 'high' (funds lost or stolen, protocol "
+ "insolvency, permanently frozen assets, or unauthorized privileged control), 'medium' "
+ "(limited, conditional, or recoverable loss, or temporary denial of service), 'low' (a minor "
+ "deviation with no funds at risk), or 'none' (no real-world exploit path — a specification or "
+ "code-quality observation)."
+ ))
+ likelihood_level: LikelihoodLevel = Field(description=(
+ "How reachable the counterexample is: 'high' (any actor, no special preconditions), 'medium' "
+ "(a specific but reachable state, ordering, or setup), or 'low' (privileged access, an unusual "
+ "configuration, or a narrow window)."
+ ))
+ risk_reasoning: str = Field(description=(
+ "One to three sentences justifying the impact and likelihood you assigned, grounded in the "
+ "counterexample."
+ ))
+
+
+async def build_findings(
+ *,
+ contract_name: str,
+ rules: list[RuleVerdict],
+ properties: list[FormalizedProperty],
+ groups: list[PropertyGroup],
+ fetch_evidence: EvidenceFetcher | None,
+ llm: BaseChatModel,
+) -> list[Finding]:
+ """One `Finding` per violated rule (concurrent, best-effort). Findings are produced only when the
+ backend supplies an evidence fetcher; returns ``[]`` otherwise or when nothing is violated."""
+ if fetch_evidence is None:
+ return []
+ bad = [r for r in rules if r.outcome == Outcome.BAD]
+ if not bad:
+ return []
+
+ # A violated rule -> every property it formalizes (a rule may jointly formalize several) -> the
+ # audit groups those properties sit in. Properties partition into groups (coverage-validated), so
+ # `group_by_key` has one entry per key.
+ props_by_ref: dict[RuleRef, list[FormalizedProperty]] = {}
+ for p in properties:
+ for ref in p.rule_refs:
+ props_by_ref.setdefault(ref, []).append(p)
+ group_by_key: dict[PropertyKey, PropertyGroup] = {k: g for g in groups for k in g.members}
+
+ system = load_jinja_template("autoprove_report_findings_system.j2")
+ bound = llm.with_structured_output(FindingDraft)
+
+ async def _one(rule: RuleVerdict) -> Finding | None:
+ try:
+ ev = await fetch_evidence(rule.name)
+ # Every rule in the report is referenced by >=1 property (collect drops orphans), so pass
+ # ALL of them: the rule may jointly formalize several, and only the counterexample says
+ # which one actually broke — that is the LLM's job, not ours to pre-guess.
+ props = props_by_ref.get(rule.ref, [])
+ # A rule's properties may share a group (some may be in none); dedupe by slug, first-seen order.
+ rule_groups = list({g.slug: g for p in props if (g := group_by_key.get(p.key)) is not None}.values())
+ user = load_jinja_template(
+ "autoprove_report_findings_prompt.j2",
+ contract_name=contract_name,
+ rule_name=rule.name,
+ properties=props,
+ groups=rule_groups,
+ analysis=ev.analysis if ev else None,
+ counterexample=ev.counterexample if ev else None,
+ )
+ draft = await bound.ainvoke([SystemMessage(system), HumanMessage(user)])
+ assert isinstance(draft, FindingDraft)
+ return _compose(rule, draft, ev, rule_groups)
+ except Exception: # noqa: BLE001 — one finding failing must never fail the report
+ _log.warning("report: finding synthesis failed for rule %r; skipping", rule.name, exc_info=True)
+ return None
+
+ sem = asyncio.Semaphore(_MAX_CONCURRENT_FINDING_CALLS)
+
+ async def _bounded(rule: RuleVerdict) -> Finding | None:
+ async with sem:
+ return await _one(rule)
+
+ findings = await asyncio.gather(*[_bounded(r) for r in bad])
+ return [f for f in findings if f is not None]
+
+
+def _compose(
+ rule: RuleVerdict,
+ draft: FindingDraft,
+ ev: RuleEvidence | None,
+ groups: list[PropertyGroup],
+) -> Finding:
+ return Finding(
+ title=draft.title,
+ severity=severity_for(draft.impact_level, draft.likelihood_level),
+ content=IssueContent(
+ **{f: getattr(draft, f) for f in AuthoredContent.model_fields},
+ proof_of_concept=ev.counterexample if ev else None,
+ references=[rule.prover_link] if rule.prover_link else None,
+ ),
+ provenance=FindingProvenance(
+ rule_name=rule.name,
+ spec_file=rule.spec_file,
+ outcome=rule.outcome,
+ group_slugs=[g.slug for g in groups],
+ prover_link=rule.prover_link,
+ impact=draft.impact_level,
+ likelihood=draft.likelihood_level,
+ risk_reasoning=draft.risk_reasoning,
+ ),
+ )
diff --git a/composer/spec/source/report/render.py b/composer/spec/source/report/render.py
index 25986c51..b25bee70 100644
--- a/composer/spec/source/report/render.py
+++ b/composer/spec/source/report/render.py
@@ -26,8 +26,8 @@
from composer.spec.gen_types import TypedTemplate
from composer.templates.loader import load_jinja_template
from composer.spec.source.report.schema import (
- AutoProverReport, CoverageReport, FormalizedProperty, GaveUpComponent, GroupStatus, Outcome,
- PropertyGroup, PropertyKey, ReportBackend, RuleRef, RuleVerdict, SkippedClaim,
+ AutoProverReport, CoverageReport, Finding, FormalizedProperty, GaveUpComponent, GroupStatus,
+ Outcome, PropertyGroup, PropertyKey, ReportBackend, RuleRef, RuleVerdict, SkippedClaim,
)
@@ -45,6 +45,14 @@
GroupStatus.PARTIAL: "warn",
GroupStatus.UNKNOWN: "muted",
}
+# Finding severity -> CSS badge kind. Unknown severities render muted.
+_SEVERITY_KIND: dict[str, str] = {
+ "critical": "bad",
+ "high": "bad",
+ "medium": "warn",
+ "low": "info",
+ "informational": "muted",
+}
# Per-backend human labels: the data carries the neutral `Outcome`; these turn it into the words an
# auditor reads ("Verified" for a proof, "Successful test" for a forge run).
@@ -134,6 +142,23 @@ class GroupView(TypedDict):
rows: list[RowView]
+class FindingView(TypedDict):
+ title: str
+ severity: str
+ severity_kind: str
+ impact_level: str | None
+ likelihood_level: str | None
+ risk_reasoning: str | None
+ summary: str
+ impact: str
+ description: str
+ attack_path: str | None
+ assumptions_and_uncertainties: str | None
+ link: LinkView
+ rule_name: str | None
+ spec_file: str | None
+
+
class ReportTemplateParams(TypedDict):
"""The full, typed context of ``autoprove_report.html.j2``."""
report: AutoProverReport
@@ -143,6 +168,7 @@ class ReportTemplateParams(TypedDict):
prover_runs: list[RunView]
rule_counts: list[ChipView]
group_counts: list[ChipView]
+ findings: list[FindingView]
groups: list[GroupView]
skipped: list[SkippedClaim]
gave_up: list[GaveUpComponent]
@@ -227,6 +253,28 @@ def _group_view(
}
+def _finding_view(f: Finding) -> FindingView:
+ """Project a `Finding` into the template shape: the finding fields the page shows, plus a
+ severity->CSS kind and the prover-run link pulled from provenance."""
+ prov = f.provenance
+ return {
+ "title": f.title,
+ "severity": f.severity,
+ "severity_kind": _SEVERITY_KIND.get(f.severity, "muted"),
+ "impact_level": prov.impact if prov else None,
+ "likelihood_level": prov.likelihood if prov else None,
+ "risk_reasoning": prov.risk_reasoning if prov else None,
+ "summary": f.content.summary,
+ "impact": f.content.impact,
+ "description": f.content.description,
+ "attack_path": f.content.attack_path,
+ "assumptions_and_uncertainties": f.content.assumptions_and_uncertainties,
+ "link": _link_view(prov.prover_link if prov else None),
+ "rule_name": prov.rule_name if prov else None,
+ "spec_file": prov.spec_file if prov else None,
+ }
+
+
def _build_context(report: AutoProverReport) -> ReportTemplateParams:
props_by_key = {p.key: p for p in report.properties}
rules_by_ref = {r.ref: r for r in report.rules}
@@ -244,6 +292,7 @@ def _build_context(report: AutoProverReport) -> ReportTemplateParams:
],
"rule_counts": _outcome_counts([r.outcome for r in report.rules], unit_labels),
"group_counts": _group_counts([g.status for g in report.groups], group_labels),
+ "findings": [_finding_view(f) for f in report.findings],
"groups": [
_group_view(g, props_by_key, rules_by_ref, unit_labels, group_labels)
for g in report.groups
diff --git a/composer/spec/source/report/schema.py b/composer/spec/source/report/schema.py
index a791ac7e..e838ab1c 100644
--- a/composer/spec/source/report/schema.py
+++ b/composer/spec/source/report/schema.py
@@ -157,6 +157,67 @@ class CoverageReport(BaseModel):
for a report.json it reads cold."""
+# ---------------------------------------------------------------------------
+# Findings — violated rules surfaced as audit issues.
+#
+# A `Finding` records a violated rule (a `RuleVerdict` with ``outcome == Outcome.BAD``) as an audit
+# issue: a ``title``, a ``severity``, and the ``content`` write-up, synthesized from the rule's
+# counterexample analysis and the properties/groups it breaks (see ``report/findings.py``). The field
+# set follows the "Submit Issue" body of the Sherlock Audit Engine API — schema at
+# https://api-audit-engine.sherlock.xyz/v1/docs/public — so a finding maps cleanly onto a submission,
+# but the source ``locations`` a submission needs
+# (``{owner}/{repo}`` / file / line) are NOT produced here: a run knows only local paths and CVL-spec
+# lines, so the submission layer reconstructs locations from the engagement scope + the counterexample.
+# The report-time locator is on ``provenance`` (rule name, spec file, prover-run link).
+# ---------------------------------------------------------------------------
+
+type SeverityTier = Literal["critical", "high", "medium", "low", "informational"]
+type ImpactLevel = Literal["high", "medium", "low", "none"]
+"""Impact axis. ``none`` marks a property break with no real-world consequence (a specification or
+code-quality observation); it resolves to ``informational`` severity."""
+type LikelihoodLevel = Literal["high", "medium", "low"]
+
+
+class AuthoredContent(BaseModel):
+ """The written sections of a finding. These are what the findings LLM produces, so the field
+ descriptions double as its instructions."""
+ summary: str = Field(description="A one to three sentence summary of the finding.")
+ description: str = Field(description="The full technical explanation of the vulnerability and how it manifests, grounded in the counterexample.")
+ impact: str = Field(description="The concrete consequence if the issue is exploited — funds at risk, denial of service, data exposure, and so on.")
+ attack_path: str | None = Field(default=None, description="The step-by-step path from the counterexample that triggers the issue, when one applies.")
+ assumptions_and_uncertainties: str | None = Field(default=None, description="Assumptions the finding relies on, and anything you are uncertain about.")
+
+
+class IssueContent(AuthoredContent):
+ """A finding's ``content``: the authored sections plus the evidence the report attaches."""
+ proof_of_concept: str | None = None
+ references: list[str] | None = None
+
+
+class FindingProvenance(BaseModel):
+ """Report-only trace from a finding back to the verdict that produced it (not part of a
+ submission payload)."""
+ rule_name: RuleName
+ spec_file: str
+ outcome: Outcome
+ group_slugs: list[str] = Field(default_factory=list)
+ prover_link: str | None = None
+ impact: ImpactLevel | None = None
+ likelihood: LikelihoodLevel | None = None
+ #: The findings LLM's justification for the assessed impact and likelihood.
+ risk_reasoning: str | None = None
+
+
+class Finding(BaseModel):
+ """A violated rule rendered as an audit issue. ``severity`` is computed from the assessed impact
+ and likelihood (both recorded on ``provenance``); ``content`` holds the write-up. Source
+ ``locations`` are added by the submission layer, not stored here."""
+ title: str
+ severity: SeverityTier
+ content: IssueContent
+ provenance: FindingProvenance | None = None
+
+
class AutoProverReport(BaseModel):
"""Top-level report document — written to ``certora/ap_report/report.json``."""
schema_version: Literal["3.0"] = "3.0"
@@ -172,3 +233,7 @@ class AutoProverReport(BaseModel):
skipped: list[SkippedClaim] = Field(default_factory=list)
gave_up_components: list[GaveUpComponent] = Field(default_factory=list)
coverage: CoverageReport
+ #: Violated rules surfaced as audit issues (one per BAD rule; empty
+ #: when nothing is violated, when synthesis was unavailable, or for a non-prover backend). Prose
+ #: is synthesized at report time — see ``report/findings.py``.
+ findings: list[Finding] = Field(default_factory=list)
diff --git a/composer/templates/autoprove_report.html.j2 b/composer/templates/autoprove_report.html.j2
index 894c6c9c..12b75f4f 100644
--- a/composer/templates/autoprove_report.html.j2
+++ b/composer/templates/autoprove_report.html.j2
@@ -134,6 +134,58 @@ section.prop .desc {
.badge-warn { color: var(--warn-fg); background: var(--warn-bg); }
.badge-info { color: var(--info-fg); background: var(--info-bg); }
.badge-muted { color: var(--muted-fg); background: var(--muted-bg); }
+.findings-head { margin: 20px 0 12px; }
+.findings-head h2 {
+ margin: 0 0 4px;
+ font-size: 18px;
+ font-weight: 600;
+ display: flex;
+ align-items: center;
+ gap: 10px;
+}
+section.finding {
+ background: var(--bg-card);
+ border: 1px solid var(--border);
+ border-left: 3px solid var(--bad-fg);
+ border-radius: 8px;
+ padding: 16px 20px;
+ margin-bottom: 14px;
+}
+section.finding h3 {
+ margin: 0 0 8px;
+ font-size: 15px;
+ font-weight: 600;
+ display: flex;
+ align-items: center;
+ gap: 10px;
+ flex-wrap: wrap;
+}
+.finding-meta {
+ display: flex;
+ flex-wrap: wrap;
+ gap: 12px;
+ align-items: baseline;
+ font-size: 12px;
+ color: var(--muted-fg);
+ margin-bottom: 10px;
+}
+.finding-meta code {
+ font-family: ui-monospace, SFMono-Regular, "SF Mono", Menlo, monospace;
+ font-size: 11.5px;
+ word-break: break-all;
+}
+.finding-meta a { color: var(--link); text-decoration: none; }
+.finding-meta a:hover { text-decoration: underline; }
+.finding-summary { margin: 0 0 12px; }
+dl.finding-body {
+ display: grid;
+ grid-template-columns: max-content 1fr;
+ gap: 6px 16px;
+ margin: 0;
+ font-size: 13px;
+}
+dl.finding-body dt { color: var(--muted-fg); font-weight: 600; }
+dl.finding-body dd { margin: 0; white-space: pre-wrap; }
table.rules {
width: 100%;
border-collapse: collapse;
@@ -247,6 +299,30 @@ footer.report-foot {
{%- endif %}
+{%- if findings %}
+
+ Findings {{ findings | length }}
+ Violated {{ terms.unit_plural }} surfaced as audit issues — fields follow the standard issue-submission format.
+
+{%- for f in findings %}
+
+ {{ f.severity }} {{ f.title }}
+
+ {{ f.summary }}
+
+ {%- if f.risk_reasoning %}- Severity rationale
- {% if f.impact_level %}Impact {{ f.impact_level }} × Likelihood {{ f.likelihood_level }} — {% endif %}{{ f.risk_reasoning }}
{% endif %}
+ - Impact
- {{ f.impact }}
+ - Description
- {{ f.description }}
+ {%- if f.attack_path %}- Attack path
- {{ f.attack_path }}
{% endif %}
+ {%- if f.assumptions_and_uncertainties %}- Assumptions & uncertainties
- {{ f.assumptions_and_uncertainties }}
{% endif %}
+
+
+{%- endfor %}
+{%- endif %}
+
{%- for g in groups %}
{{ g.slug }} {{ g.title }} {{ g.label }}
diff --git a/composer/templates/autoprove_report_findings_prompt.j2 b/composer/templates/autoprove_report_findings_prompt.j2
new file mode 100644
index 00000000..33661841
--- /dev/null
+++ b/composer/templates/autoprove_report_findings_prompt.j2
@@ -0,0 +1,26 @@
+Contract: {{ contract_name }}
+
+Violated rule: {{ rule_name }}
+
+{% if properties -%}
+Properties this rule formalizes (the counterexample breaks at least one of them — determine which):
+{% for p in properties %}
+ - ({{ p.sort }}) {{ p.title }}: {{ p.description }}
+{%- endfor %}
+{% endif -%}
+{% if groups -%}
+Audit-level claim(s) it belongs to:
+{% for g in groups %}
+ - {{ g.title }}: {{ g.description }}
+{%- endfor %}
+{% endif %}
+{% if analysis -%}
+The prover's counterexample analysis (root cause + suggested fix), produced during the run:
+{{ analysis }}
+{%- else -%}
+(No counterexample analysis was captured for this rule — reason conservatively from the properties.)
+{%- endif %}
+{% if counterexample %}
+Counterexample trace:
+{{ counterexample }}
+{% endif %}
diff --git a/composer/templates/autoprove_report_findings_system.j2 b/composer/templates/autoprove_report_findings_system.j2
new file mode 100644
index 00000000..b14a8d87
--- /dev/null
+++ b/composer/templates/autoprove_report_findings_system.j2
@@ -0,0 +1,33 @@
+You are a smart-contract security auditor writing up a single, confirmed finding for an audit
+report. A formal-verification rule was VIOLATED: the Certora Prover found a concrete counterexample
+refuting a property the code was expected to satisfy. A violation is hard evidence the property does
+not hold, so treat this as a real, confirmed issue — not a maybe.
+
+You are given the violated rule, the natural-language properties it formalizes, the audit-level
+claim(s) it belongs to, the prover's own root-cause analysis of the counterexample, and — when
+available — the counterexample trace. A single rule may jointly formalize several properties, but the
+counterexample violates one specific behavior: ground the write-up in what the counterexample actually
+shows, and address the property it truly breaks rather than assuming a particular one.
+
+You assess two axes; the report computes the severity from them, so judge each on its own merits.
+
+Impact — what an attacker gains or users lose when the property break is exploited:
+ - high: direct loss or theft of funds, protocol insolvency, permanently frozen assets, or
+ unauthorized control of privileged functionality.
+ - medium: limited, conditional, or recoverable value loss; temporary denial of service; incorrect
+ accounting that does not by itself drain funds.
+ - low: no funds at risk; a minor deviation from the intended guarantee.
+ - none: no real-world exploit path at all — a specification or code-quality observation.
+
+Likelihood — how reachable the counterexample is in practice:
+ - high: exploitable by any actor with no special preconditions or privileged access.
+ - medium: needs a specific but reachable state, ordering, or non-trivial-yet-feasible setup.
+ - low: needs privileged access, an unusual configuration, or a narrow / costly window.
+
+A prover counterexample proves the property CAN be broken, so do not rate a genuine break as impact
+'none' merely because it looks hard to reach — lower the Likelihood instead. Reserve impact 'none' for
+a break with no real-world consequence.
+
+Ground every claim in the supplied analysis and counterexample. Do not invent addresses, values,
+function names, or exploit steps beyond what the material supports; if the material is thin, keep the
+finding modest rather than fabricating specifics.
diff --git a/tests/test_autoprove_report.py b/tests/test_autoprove_report.py
index 9849699a..1298c260 100644
--- a/tests/test_autoprove_report.py
+++ b/tests/test_autoprove_report.py
@@ -10,7 +10,7 @@
which is how a caller hands a gap to the report layer).
"""
from types import SimpleNamespace
-from typing import cast
+from typing import Any, cast
import pathlib
import pytest
@@ -35,10 +35,15 @@
)
from composer.spec.source.report.render import render_html
from composer.spec.source.report.schema import (
- AutoProverReport, CoverageReport, FormalizedProperty, GaveUpComponent, GroupStatus,
- Outcome, PropertyGroup, RuleVerdict, SkippedClaim,
+ AutoProverReport, CoverageReport, Finding, FindingProvenance, FormalizedProperty,
+ GaveUpComponent, GroupStatus, ImpactLevel, IssueContent, LikelihoodLevel, Outcome,
+ PropertyGroup, RuleVerdict, SeverityTier, SkippedClaim,
)
from composer.spec.source.report_prover import make_prover_fetcher
+from composer.spec.source.report.collect import RuleEvidence
+from composer.spec.source.report.findings import FindingDraft, build_findings
+from composer.spec.source.cex_capture import CexAnalysisStore
+from composer.diagnostics.timing import RunSummary
# ---------------------------------------------------------------------------
@@ -114,22 +119,21 @@ def _pg(slug, members, status=GroupStatus.GOOD) -> PropertyGroup:
return PropertyGroup(slug=slug, title="T", description="d", status=status, members=members)
-class _GroupingStubModel(BaseChatModel):
- """A `BaseChatModel` whose structured-output binding returns a preset `GroupingResult`.
- Lets the build tests drive the *real* `call_grouping_llm` — template rendering + the
- `isinstance` check — without a live model, only stubbing the model's output."""
- result: GroupingResult
+class _StructuredStubModel(BaseChatModel):
+ """A `BaseChatModel` whose structured-output binding returns a preset object, so tests drive the
+ real caller (template rendering + the `isinstance` check) without a live model."""
+ output: Any
def with_structured_output(self, schema, **kwargs) -> Runnable: # type: ignore[override]
- result = self.result
- return RunnableLambda(lambda _messages: result)
+ out = self.output
+ return RunnableLambda(lambda _messages: out)
def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
raise NotImplementedError("stub is structured-output only")
@property
def _llm_type(self) -> str:
- return "grouping-stub"
+ return "structured-stub"
# ---------------------------------------------------------------------------
@@ -451,7 +455,7 @@ def test_artifact_store_write_report_round_trips(tmp_path):
async def test_build_groups_properties(tmp_path):
gen = _gen({"p1": ["r1"], "p2": ["r2"]})
fetch = _fetcher({"L1": [_fake_check("r1", NodeStatus.VERIFIED), _fake_check("r2", NodeStatus.VERIFIED)]})
- llm = _GroupingStubModel(result=GroupingResult(groups=[PropertyGroupDraft(
+ llm = _StructuredStubModel(output=GroupingResult(groups=[PropertyGroupDraft(
slug="g", title="G", description="d", members=[("C", "p1"), ("C", "p2")])]))
report = await build.build_report(
@@ -470,7 +474,7 @@ async def test_build_groups_properties(tmp_path):
async def test_build_empty_grouping_falls_back(tmp_path):
gen = _gen({"p1": ["r1"], "p2": ["r2"]})
fetch = _fetcher({"L1": [_fake_check("r1", NodeStatus.VERIFIED), _fake_check("r2", NodeStatus.VIOLATED)]})
- llm = _GroupingStubModel(result=GroupingResult(groups=[])) # empty grouping -> fallback
+ llm = _StructuredStubModel(output=GroupingResult(groups=[])) # empty grouping -> fallback
report = await build.build_report(
contract_name="C",
@@ -490,7 +494,7 @@ async def test_build_empty_grouping_falls_back(tmp_path):
async def test_build_surfaces_skipped_and_gave_up_gaps(tmp_path):
gen = _gen({"p_ok": ["r1"]}, skipped={"p_skip": "needs a ghost"})
fetch = _fetcher({"L1": [_fake_check("r1", NodeStatus.VERIFIED)]})
- llm = _GroupingStubModel(result=GroupingResult(groups=[PropertyGroupDraft(
+ llm = _StructuredStubModel(output=GroupingResult(groups=[PropertyGroupDraft(
slug="g", title="G", description="d", members=[("C", "p_ok")])]))
report = await build.build_report(
@@ -506,3 +510,210 @@ async def test_build_surfaces_skipped_and_gave_up_gaps(tmp_path):
assert [(s.component, s.title) for s in report.skipped] == [("C", "p_skip")]
assert [g.component for g in report.gave_up_components] == ["D"]
assert report.coverage.skipped_count == 1 and report.coverage.gave_up_component_count == 1
+
+
+# ---------------------------------------------------------------------------
+# findings (violated rules -> audit-issue findings)
+# ---------------------------------------------------------------------------
+
+
+def _draft(*, impact_level: ImpactLevel = "high", likelihood_level: LikelihoodLevel = "medium",
+ title: str = "Reentrancy drains vault") -> FindingDraft:
+ return FindingDraft(
+ title=title, impact_level=impact_level, likelihood_level=likelihood_level,
+ risk_reasoning="High impact (fund loss); medium likelihood (needs a specific state).",
+ summary="s", description="d", impact="funds at risk", attack_path="1..2..3",
+ )
+
+
+def _evidence(by_rule: dict[str, RuleEvidence]):
+ async def fetch(rule_name):
+ return by_rule.get(rule_name)
+ return fetch
+
+
+def _finding(severity: SeverityTier = "high") -> Finding:
+ return Finding(
+ title="Reentrancy drains vault", severity=severity,
+ content=IssueContent(summary="s", description="d", impact="funds at risk",
+ assumptions_and_uncertainties="assumes attacker deploys a contract",
+ proof_of_concept=""),
+ provenance=FindingProvenance(rule_name="r_bad", spec_file="c.spec", outcome=Outcome.BAD,
+ group_slugs=["g"], prover_link="https://prover.example/run/abc",
+ impact="high", likelihood="medium",
+ risk_reasoning="High impact; medium likelihood."),
+ )
+
+
+def test_severity_for_matrix():
+ """Severity is computed from the impact × likelihood matrix, not chosen by the LLM."""
+ from composer.spec.source.report.findings import severity_for
+ assert severity_for("high", "high") == "critical"
+ assert severity_for("high", "medium") == "high"
+ assert severity_for("high", "low") == "medium"
+ assert severity_for("medium", "high") == "high"
+ assert severity_for("medium", "medium") == "medium"
+ assert severity_for("low", "high") == "low" # low impact caps at low regardless of likelihood
+ assert severity_for("low", "low") == "low"
+ assert severity_for("none", "high") == "informational" # no exploit path -> informational
+ assert severity_for("none", "low") == "informational"
+
+
+@pytest.mark.asyncio
+async def test_build_report_synthesizes_one_finding_per_violation():
+ """Only the violated rule becomes a finding; severity is computed from the model's impact/
+ likelihood, the counterexample rides proof_of_concept, the run link is a reference, and provenance
+ traces back to the rule. No locations are produced at report time (submission builds those)."""
+ gen = _gen({"p_good": ["r_ok"], "p_bad": ["r_bad"]})
+ fetch = _fetcher({"L1": [
+ _fake_check("r_ok", NodeStatus.VERIFIED, file="autospec_C.spec"),
+ _fake_check("r_bad", NodeStatus.VIOLATED, line=42, file="autospec_C.spec"),
+ ]})
+ grouping = _StructuredStubModel(output=GroupingResult(groups=[PropertyGroupDraft(
+ slug="g", title="G", description="gd", members=[("C", "p_good"), ("C", "p_bad")])]))
+ evidence = _evidence({"r_bad": RuleEvidence(analysis="root cause X", counterexample="")})
+
+ report = await build.build_report(
+ contract_name="Vault", backend="prover",
+ components=[_input("C", "autospec_C.spec",
+ [_prop("p_good", "d"), _prop("p_bad", "d")], gen)],
+ llm=grouping, fetch_verdicts=fetch,
+ findings_llm=_StructuredStubModel(output=_draft()), fetch_evidence=evidence,
+ )
+
+ assert len(report.findings) == 1
+ f = report.findings[0]
+ assert f.severity == "high" # computed: impact high × likelihood medium
+ prov = f.provenance
+ assert prov is not None
+ assert prov.impact == "high" and prov.likelihood == "medium"
+ assert prov.risk_reasoning # the axis justification is captured
+ assert prov.rule_name == "r_bad" and prov.outcome == Outcome.BAD
+ assert prov.group_slugs == ["g"] and prov.spec_file == "autospec_C.spec"
+ assert not hasattr(f, "locations") # locations are a submission-layer concern, not on the report
+ assert f.content.proof_of_concept == ""
+ assert f.content.references == ["L1"]
+
+
+@pytest.mark.asyncio
+async def test_build_report_no_findings_without_findings_llm():
+ """The findings pass is opt-in: omitting ``findings_llm`` leaves findings empty (back-compat)."""
+ gen = _gen({"p_bad": ["r_bad"]})
+ fetch = _fetcher({"L1": [_fake_check("r_bad", NodeStatus.VIOLATED)]})
+ grouping = _StructuredStubModel(output=GroupingResult(groups=[PropertyGroupDraft(
+ slug="g", title="G", description="d", members=[("C", "p_bad")])]))
+ report = await build.build_report(
+ contract_name="C", backend="prover",
+ components=[_input("C", "autospec_C.spec", [_prop("p_bad", "d")], gen)],
+ llm=grouping, fetch_verdicts=fetch,
+ )
+ assert report.findings == []
+
+
+@pytest.mark.asyncio
+async def test_build_findings_degrades_without_analysis():
+ """A violated rule whose evidence fetch yields nothing (fetcher present, no analysis for that rule)
+ still produces a finding from the property/group text; proof_of_concept is simply absent."""
+ rules = [_rv("c.spec", "r_bad", Outcome.BAD)]
+ props = [_fp("C", "p_bad", [("c.spec", "r_bad")], desc="balances stay solvent")]
+ groups = [_pg("g", [("C", "p_bad")], status=GroupStatus.BAD)]
+ findings = await build_findings(
+ contract_name="Vault", rules=rules, properties=props, groups=groups,
+ fetch_evidence=_evidence({}), # fetcher present, but no evidence recorded for r_bad
+ llm=_StructuredStubModel(output=_draft(impact_level="medium", likelihood_level="medium")),
+ )
+ assert len(findings) == 1
+ assert findings[0].severity == "medium" # medium × medium
+ assert findings[0].content.proof_of_concept is None
+ prov = findings[0].provenance
+ assert prov is not None and prov.spec_file == "c.spec"
+
+
+@pytest.mark.asyncio
+async def test_build_findings_empty_without_evidence_fetcher():
+ """Findings are produced only when the backend supplies an evidence fetcher — no backend-id
+ check. A backend that opts out (fetch_evidence is None) yields no findings."""
+ rules = [_rv("c.spec", "r_bad", Outcome.BAD)]
+ findings = await build_findings(
+ contract_name="V", rules=rules, properties=[], groups=[],
+ fetch_evidence=None, llm=_StructuredStubModel(output=_draft()),
+ )
+ assert findings == []
+
+
+def test_findings_prompt_lists_all_properties_a_rule_formalizes():
+ """A rule may jointly formalize several properties; the prompt presents ALL of them so the model
+ grounds the write-up in the one the counterexample actually breaks (not a pre-picked one)."""
+ from composer.templates.loader import load_jinja_template
+ user = load_jinja_template(
+ "autoprove_report_findings_prompt.j2",
+ contract_name="Vault", rule_name="allowDeposits_revert_characteristic",
+ properties=[_prop("revert_on_non_admin", "reverts if caller is not an admin"),
+ _prop("revert_on_zero_actor", "reverts if allowedActor is address(0)")],
+ groups=[], analysis=None, counterexample="",
+ )
+ assert "reverts if caller is not an admin" in user
+ assert "reverts if allowedActor is address(0)" in user
+
+
+def test_render_html_findings_section():
+ report = _mini_report().model_copy(update={"findings": [_finding()]})
+ h = render_html(report)
+ assert '' in h
+ assert "Reentrancy drains vault" in h
+ assert "badge-bad" in h # high severity
+ assert "r_bad" in h and "c.spec" in h # provenance locator (rule in spec file)
+ assert "Severity rationale" in h
+ assert "Impact high" in h and "Likelihood medium" in h
+ assert "Assumptions" in h and "assumes attacker deploys a contract" in h
+
+
+def test_render_html_omits_findings_section_when_empty():
+ h = render_html(_mini_report()) # _mini_report has no findings
+ assert '' not in h
+ assert '' not in h
+
+
+def test_report_round_trips_with_findings(tmp_path):
+ report = _mini_report().model_copy(update={"findings": [_finding()]})
+ ProverArtifactStore(str(tmp_path), "Counter").write_report(report)
+
+ out = tmp_path / "certora" / "ap_report" / "report.json"
+ reloaded = AutoProverReport.model_validate_json(out.read_text())
+ assert len(reloaded.findings) == 1
+ rf = reloaded.findings[0]
+ assert rf.content.impact == "funds at risk"
+ prov = rf.provenance
+ assert prov is not None
+ assert prov.rule_name == "r_bad"
+ assert prov.impact == "high"
+ assert prov.risk_reasoning == "High impact; medium likelihood."
+ assert not hasattr(rf, "locations")
+
+
+@pytest.mark.asyncio
+async def test_spec_callbacks_captures_cex_analysis():
+ """The source-pipeline prover callback records each violated rule's analysis into the run-scoped
+ store, keyed by the bare rule name, while still emitting the stream event."""
+ from langgraph.store.memory import InMemoryStore
+ from composer.spec.source.prover import _SpecCallbacks
+ from composer.prover.ptypes import RulePath, RuleResult
+
+ store = CexAnalysisStore(store=InMemoryStore(), namespace=("cex_analyses", "t"))
+ events: list = []
+ cb = _SpecCallbacks(events.append, "tc1", cast(RunSummary, SimpleNamespace()), {},
+ analysis_store=store)
+ rule = RuleResult(path=RulePath(rule="no_reentrancy", method="withdraw"),
+ cex_dump="", status="VIOLATED")
+
+ await cb.on_analysis_complete(rule, "root cause: external call before state update")
+
+ # Keyed by the bare rule name (RulePath.rule) — exactly what the report's RuleVerdict.name carries
+ # (POU derives it from the prover tree's context[0]) — so the findings join is an exact match.
+ rec = await store.get("no_reentrancy")
+ assert rec is not None and rec.analysis.startswith("root cause")
+ assert rec.counterexample == ""
+ # NOT keyed by the pretty-printed form ("no_reentrancy for withdraw"); the report never looks that up.
+ assert await store.get(rule.name) is None
+ # the UI stream event still fires
+ assert any(e.get("type") == "rule_analysis" for e in events)