diff --git a/composer/spec/source/agent_groups.py b/composer/spec/source/agent_groups.py new file mode 100644 index 00000000..7ce96801 --- /dev/null +++ b/composer/spec/source/agent_groups.py @@ -0,0 +1,203 @@ +"""Agent-declared verification groups over properties. + +The transparent, agent-controlled splitting policy. Rather than infer a partition, +the CVL author *declares* it: a set of groups, each naming the properties it +verifies and the summaries it installs (per function). This +rides the structure autoprover already has — the property -> rule mapping and its +coverage guarantee (every non-skipped property is mapped to rules) — so a group is +expressed in the agent's native unit (properties), and coverage composes: every +property lands in exactly one group, every group's owned rules are verified +(per-group completion), therefore every property is covered. + +Groups here are NOT opaque. The agent names them, sees their membership, and +controls each group's summaries (`summaries`). (The substrate group also carries a +per-group conf overlay, but the agent-facing tool does not expose it yet — a +per-group prover-config knob is new, unvalidated, and unreviewed by the judge, so it +is deferred to a follow-up with proper guardrails.) +The machinery only expands properties to rules, enforces a disjoint rule partition, +and validates coverage; the group count is capped by rejecting an over-cap declaration +(:func:`over_cap_message`), not by merging. The per-group spec is the +shared base spec plus a `methods{}` block of the summaries that group declared +(:func:`composer.spec.source.verification_groups.append_summaries`); a function a +group does not summarize is verified precise there. +""" + +from pydantic import BaseModel, Field + +from composer.spec.cvl_generation import PropertyRuleMapping +from composer.spec.source.verification_groups import ( + VerificationGroup, append_summaries, cap_groups, +) + + +def validate_declared_coverage( + specs: list["VerificationGroupSpec"], + *, + all_properties: set[str], + skipped: set[str], +) -> str | None: + """Whether the declared groups cover the property space exactly once. + + Reuses autoprover's coverage contract: every non-skipped property must be + assigned to exactly one group; no unknown or skipped property may be assigned. + Returns None when valid, else one message enumerating every problem — the shape + an agent tool hands back so the author can fix its declaration.""" + assigned: list[str] = [str(m.property_title) for s in specs for m in s.property_rules] + seen: set[str] = set() + duplicated: set[str] = set() + for p in assigned: + (duplicated if p in seen else seen).add(p) + required = all_properties - skipped + problems: list[str] = [] + if duplicated: + problems.append(f"properties assigned to more than one group: {sorted(duplicated)}") + if missing := required - seen: + problems.append(f"non-skipped properties assigned to no group: {sorted(missing)}") + if unknown := seen - all_properties: + problems.append(f"unknown property titles: {sorted(unknown)}") + if skipped_assigned := seen & skipped: + problems.append(f"skipped properties should not be assigned to a group: {sorted(skipped_assigned)}") + return "; ".join(problems) if problems else None + + +# --- Agent-facing declaration (the tool input / state shape) ---------------- + + +class VerificationGroupSpec(BaseModel): + """One verification group as the CVL author declares it — the transparent, + agent-controlled unit. Carries its own property->rule mapping so the rules are + known during authoring (the publish-time mapping is their union) and the functions + it keeps precise. (Per-group conf overrides are supported by the substrate but not + exposed here yet — see the module docstring.)""" + name: str = Field(description="A short, unique, human-readable name for this group (used in run/spec names).") + property_rules: list[PropertyRuleMapping] = Field( + description="The properties this group verifies and, for each, the rule/invariant names in " + "the spec that verify it. A group may cover multiple properties. Across all groups every " + "non-skipped property must appear in exactly one group." + ) + summaries: dict[str, str] = Field( + default_factory=dict, + description="The summaries THIS group installs: each key a hostile function, each value the full " + "CVL methods{} entry to summarize it here (e.g. \"function C.f(uint) external => NONDET;\", a ghost " + "mirror, a model). A function absent from this map is verified as the base spec has it in this " + "group — precise only if the base spec (incl. its imports) does not already summarize it. The same " + "function may be summarized differently in different groups — choose, per group, " + "the weakest summary sound for that group's rules, and reuse the same entry across groups where " + "it is sound (consistency).", + ) + + +def owned_rules_per_group(specs: list[VerificationGroupSpec]) -> list[frozenset[str]]: + """Each spec's owned rules under first-declaration-wins, aligned to ``specs`` by index. + + A group's owned rules are the union of its properties' rules; a rule declared by more than + one group is owned by the FIRST that declares it, so the partition stays disjoint — every + rule has exactly one owner.""" + claimed: set[str] = set() + owned_per: list[frozenset[str]] = [] + for s in specs: + owned = {str(r) for m in s.property_rules for r in m.rules} - claimed + claimed |= owned + owned_per.append(frozenset(owned)) + return owned_per + + +def over_cap_message(specs: list[VerificationGroupSpec], cap: int) -> str | None: + """A rejection message when the agent declared MORE groups than the cap, else ``None``. + + Each group is a separate prover run, so the count is bounded. Rather than silently auto-merge the + declaration (which would undo the split the agent deliberately chose), the declaring tool rejects an + over-cap declaration and asks the agent to refactor — and SUGGESTS a concrete valid merge (the greedy, + most-agreeing-summaries merge :func:`cap_groups` computes), which the agent can adopt or improve.""" + if len(specs) <= cap: + return None + # A lightweight sim of the partition, run through cap_groups to name one valid merge to suggest. + sim = [ + VerificationGroup(name=s.name, owned_rules=owned, summaries=dict(s.summaries)) + for s, owned in zip(specs, owned_rules_per_group(specs)) + ] + suggested = "; ".join(g.name for g in cap_groups(sim, cap)) + return ( + f"You declared {len(specs)} verification groups but at most {cap} are allowed — each group is a " + f"separate prover run (raise the limit via AUTOPROVER_MAX_VERIFICATION_GROUPS). Merge groups until " + f"there are at most {cap}: combine the ones whose rules can share the same summaries — a merged " + f"group keeps a summary only where both groups agree, else that function drops to precise. One valid " + f"merge to adopt or improve: {suggested}." + ) + + +def groups_from_specs( + base_spec: str, + specs: list[VerificationGroupSpec], + *, + cap: int, +) -> list[VerificationGroup]: + """Expand the agent's declared group specs into runnable :class:`VerificationGroup`s + (coverage assumed already validated with :func:`validate_declared_coverage`). + + Owned rules are partitioned first-declaration-wins (:func:`owned_rules_per_group`). Each + group's spec installs the summaries it declared (:func:`append_summaries`); a function it + does not summarize is verified precise. The declaration must already be within ``cap`` — the + declaring tool rejects an over-cap declaration (:func:`over_cap_message`) rather than merging + — so this asserts the bound instead of capping.""" + assert len(specs) <= cap, ( + f"{len(specs)} groups exceeds cap {cap}; over-cap declarations are rejected at declare time" + ) + return [ + VerificationGroup( + name=s.name, + owned_rules=owned, + spec_contents=append_summaries(base_spec, s.summaries), + summaries=dict(s.summaries), + # conf_overlay left at its substrate default: not exposed to the agent. + ) + for s, owned in zip(specs, owned_rules_per_group(specs)) + ] + + +def render_group_plan_for_judge(specs: list["VerificationGroupSpec"]) -> str | None: + """A judge-facing note describing the verification-group plan, or ``None`` when + no groups are declared. + + The feedback judge reviews the *base* spec (``curr_spec``), which deliberately + leaves the hostile summaries OUT of its ``methods{}`` block — each group installs + its own summaries at prover time via :func:`append_summaries`. Without this note + the judge sees hostile functions used-but-not-summarized and false-flags them as + unsound/HAVOCing. The note makes each group's install concrete: its properties, + rules, and exactly which functions it summarizes (with the summary text) — so the + judge evaluates the spec as it is actually verified, not as a monolith. A function + a group does NOT list is verified as the base spec has it there — precise unless the + base spec itself summarizes it.""" + if not specs: + return None + lines: list[str] = [ + "// ============================================================================", + "// Verification-group plan (informational — NOT part of the base spec above)", + "// ============================================================================", + "// This spec is NOT verified as a monolith. It is split into parallel prover", + "// runs ('verification groups'), each with its OWN methods{} block installing the", + "// summaries listed below. A hostile function that appears un-summarized in the base", + "// spec above IS summarized in every group that lists it here — treat those as", + "// installed (not HAVOCing) when judging soundness and coverage; a function a group", + "// does not list is verified as the base spec above has it (precise unless the base", + "// spec above already summarizes it). A summary a group DOES list for a function the base", + "// spec above already summarizes is a more-specific override (exact beats wildcard), so that", + "// group verifies under the group's entry, not the base's.", + "//", + ] + for s in specs: + props = [str(m.property_title) for m in s.property_rules] + rules = [str(r) for m in s.property_rules for r in m.rules] + lines.append(f"// Group \"{s.name}\":") + lines.append(f"// properties: {', '.join(props) if props else '(none)'}") + lines.append(f"// rules: {', '.join(rules) if rules else '(none)'}") + if s.summaries: + lines.append("// installs summaries:") + for func in sorted(s.summaries): + lines.append(f"// {func}: {s.summaries[func].strip()}") + else: + lines.append("// installs summaries: (none — all functions precise)") + # No per-group conf is shown (the agent-facing group carries none). If groups gain + # agent-set conf and the judge starts reviewing .conf files, list each group's conf diff here. + lines.append("//") + return "\n".join(lines) diff --git a/composer/spec/source/author.py b/composer/spec/source/author.py index e16a37e6..4f74f013 100644 --- a/composer/spec/source/author.py +++ b/composer/spec/source/author.py @@ -26,6 +26,8 @@ GeneratedCVL, PropertyRuleMapping, AppliedEdit, FeedbackToolBase, ) from composer.prover.core import run_prover, CexHandler, ProverCallbacks, ProverReport +from composer.spec.source.agent_groups import VerificationGroupSpec, over_cap_message, render_group_plan_for_judge +from composer.spec.source.verification_groups import resolved_max_groups from composer.spec.source.live_explorer import VersionedHistory, LiveEditTools, WIPE_HISTORY from composer.spec.source.prover import setup_prover_config_in from composer.spec.context import WorkflowContext, CVLGeneration, CacheKey, SourceCode @@ -170,6 +172,91 @@ async def run(self) -> Command | str: ) +@tool_display(lambda p: f"Declaring {len(p['groups'])} verification group(s)", None) +class DeclareVerificationGroups( + WithAsyncDependencies[Command | str, list[PropertyTitle]], + WithInjectedState[SourceCVLGenerationState], + WithInjectedId, +): + """ + Split verification into independent, PARALLEL prover runs ("verification groups"). + + Use this when one combined run is (or would be) intractable — typically a timeout from + having to keep too much of the code precise at once. Each group verifies a subset of the + properties under its OWN summarization and configuration, so different groups can keep + DIFFERENT functions precise: a function summarized in one group can stay exact in another. + This breaks the "one global methods block forces the intersection of every rule's precision + needs" bottleneck. + + A valid declaration: + - Every non-skipped property appears in exactly ONE group (via that group's `property_rules`). + Coverage is checked exactly as at publish time. + - Each group's `summaries` maps each function IT summarizes to the CVL methods-block entry to + summarize it with, HERE. A function a group omits is verified as the base spec has it — precise only + if the base spec (incl. its imports) does not already summarize it. The same + function may be summarized differently in different groups (a rule may allow `foo` monotone + while another needs it injective) — choose, per group, the weakest summary sound for that + group's rules. + - The base spec you put on the VFS must define any ghosts/CVL functions those entries use, and + must itself leave the summarized functions UNsummarized (each group's spec adds its own). + - Your `summaries` are APPENDED to the base spec's methods block (autosetup's imported summaries + included). To change how an ALREADY-base-summarized function behaves in a group, your entry must + be MORE SPECIFIC than the base's — an exact `Contract.f(...)` overrides a wildcard `_.f(...)`. If + the base already summarizes it EXACTLY for that contract you cannot override or remove it (a second + exact entry is a duplicate → typecheck error), and there is no way to make it precise in one group; + the workaround is to summarize a CALLER of that function instead (usually not base-summarized). + + Note: a rule reachable from two properties in DIFFERENT groups is verified ONCE — in the FIRST + group that declares it — under THAT group's summaries (the run partitions rules disjointly). So + make the first group's summaries sound for any rule it shares with a later group, or keep that + rule's functions precise there. + + Groups run in parallel; already-verified rules are not re-run. Call again to REPLACE the whole + partition; pass an empty `groups` list to revert to a single combined run. There is a cap on the + number of groups (each is a separate prover run); declaring more is REJECTED with the merge the run + would otherwise force, so you refactor the partition yourself rather than have it silently merged. + """ + groups: list[VerificationGroupSpec] = Field( + description="The verification groups to split into. Empty list reverts to one combined run." + ) + + @override + async def run(self) -> Command | str: + specs = self.groups + if not specs: + return tool_state_update( + self.tool_call_id, + "Reverted to a single combined verification run.", + verification_groups=[], + ) + names = [s.name for s in specs] + if len(set(names)) != len(names): + return "Group names must be unique." + # Coverage: the union of the groups' property->rule mappings must cover every + # non-skipped property — the same check applied at publish. + combined = [m for s in specs for m in s.property_rules] + with self.tool_deps() as titles: + if (err := validate_property_rules(combined, self.state["skipped"], titles)) is not None: + return err + # Partition: a property must not be claimed by more than one group. + seen: set[str] = set() + dup: set[str] = set() + for s in specs: + for m in s.property_rules: + (dup if m.property_title in seen else seen).add(m.property_title) + if dup: + return f"Each property must belong to exactly one group; these appear in more than one: {sorted(dup)}" + # Reject an over-cap declaration; over_cap_message suggests a valid merge to adopt. + if (over := over_cap_message(specs, resolved_max_groups())) is not None: + return over + return tool_state_update( + self.tool_call_id, + f"Declared {len(specs)} verification group(s): {', '.join(names)}. " + "Subsequent verify_spec runs split the rules across them and run in parallel.", + verification_groups=specs, + ) + + _GIVE_UP_DESCRIPTION = """ Call this tool to give up on the CVL generation for this task. @@ -619,7 +706,13 @@ async def _get_feedback( vfs=self.state["vfs"], version_history=self.state["version_history"], ) - return await judge(snap, spec, skipped, self.rebuttals, self.tool_call_id) + # The judge reviews the base spec, whose methods{} block deliberately omits + # the hostile summaries — each verification group installs its own at prover + # time (append_summaries). Surface that plan so the judge does not false-flag + # those functions as un-summarized / HAVOCing. + plan = render_group_plan_for_judge(self.state.get("verification_groups") or []) + judged_spec = spec if plan is None else f"{spec.rstrip()}\n\n{plan}\n" + return await judge(snap, judged_spec, skipped, self.rebuttals, self.tool_call_id) @override def _version_history(self) -> Sequence[str]: @@ -864,6 +957,7 @@ async def propose( [prover_tool.lg_tool, ExpectRulePassage.as_tool("expect_rule_passage"), ExpectRuleFailure.as_tool("expect_rule_failure"), + DeclareVerificationGroups.bind(titles).as_tool("declare_verification_groups"), give_up_tool(name="give_up", description=_GIVE_UP_DESCRIPTION, label="CVL generation"), PublishResultTool.bind(titles).as_tool("result"), ctx.get_memory_tool()] diff --git a/composer/spec/source/prover.py b/composer/spec/source/prover.py index e03ae0b9..58e7342f 100644 --- a/composer/spec/source/prover.py +++ b/composer/spec/source/prover.py @@ -12,6 +12,7 @@ import json import logging import time +from dataclasses import dataclass from contextlib import contextmanager, asynccontextmanager, ExitStack, nullcontext from pathlib import Path from typing import ( @@ -35,7 +36,7 @@ from graphcore.graph import LLM from composer.prover.core import ( - ProverOptions, SpecCompilationError, declared_rules_list, run_prover, + ProverOptions, SpecCompilationError, ProverReport, declared_rules_list, run_prover, DefaultCexHandler ) from composer.prover.callbacks import ProverEventCallbacks @@ -53,6 +54,14 @@ from composer.spec.gen_types import CERTORA_DIR, SPECS_DIR from composer.spec.util import string_hash from composer.spec.source.cex_capture import CexAnalysisStore +from composer.spec.source.verification_groups import ( + VerificationGroup, + merge_group_results, + prune_phantom_owned_rules, + resolved_max_groups, + single_group, +) +from composer.spec.source.agent_groups import VerificationGroupSpec, groups_from_specs _logger = logging.getLogger("composer.prover") @@ -113,6 +122,8 @@ class ProverRunLog(TypedDict): sort: Literal["run"] declared_rules: list[str] state_digest: str + # The verification group this run belongs to; absent (None) for an ungrouped run. + group: NotRequired[str | None] class NagMarker(TypedDict): nagged_rules: list[RulePath] @@ -135,6 +146,10 @@ def _executed_rules( #: about a rule. Counts the run being processed, so 3 means "this run plus the two before it". STUCK_RULE_NAG_THRESHOLD = 3 +#: Statuses that make a rule "stuck" — a failure worth warning/nagging about (a genuine +#: VIOLATED is a real verdict, not stuck). +STUCK_STATUSES: frozenset[StatusCodes] = frozenset({"TIMEOUT", "ERROR", "SANITY_FAILED"}) + def stuck_rule_warnings( # Values are compared for equality only, so the looser ``str`` keeps callers free of @@ -246,6 +261,116 @@ def _is_completion_history( return True return False +def _history_for_group( + l: list[ProverHistoryItem], group: str | None +) -> list[ProverHistoryItem]: + """The history entries relevant to one verification group's completion: that + group's own runs, plus every nag marker (nags are group-agnostic and are + transparent to ``_iterate_history``). Runs belonging to a *different* group are + dropped, so a group's contiguous same-digest streak is not truncated by another + group's interleaved run at a different digest. An untagged run (``group`` absent) + belongs to the default group ``None`` — the single-group / backward-compatible case.""" + return [it for it in l if it["sort"] != "run" or it.get("group") == group] + + +def group_is_complete( + l: list[ProverHistoryItem], + *, + group: str | None, + curr_digest: str, + expected_to_fail: set[str], + curr_status: list[tuple[RulePath, StatusCodes]], + owned_rules: set[str], +) -> bool: + """Whether one group's owned rules are all verified against its current spec state. + + Delegates to :func:`_is_completion_history` over the group's own filtered history, + so a group with a distinct spec (distinct ``state_digest``) is evaluated in isolation + from interleaved runs of other groups. Overall verification completeness is the AND of + this over every group.""" + return _is_completion_history( + l=_history_for_group(l, group), + curr_digest=curr_digest, + expected_to_fail=expected_to_fail, + curr_status=curr_status, + all_rules=list(owned_rules), + ) + + +def group_pending_rules( + l: list[ProverHistoryItem], + *, + group: str | None, + curr_digest: str, + owned_rules: set[str], + expected_to_fail: set[str], + curr_status: list[tuple[RulePath, StatusCodes]] | None = None, +) -> set[str]: + """The owned rules a re-run of this group still needs to cover: those whose *latest* + verdict against the group's current spec state is not ``VERIFIED`` (and that are not + expected to fail), plus any never yet run. This is the incremental re-run engine — + verified rules are never re-submitted; a timed-out / spuriously-violated rule is, on + its own, within its group's setup. Empty means the group is fully covered and its run + can be skipped entirely.""" + latest: dict[str, StatusCodes] = {} + for results in _iterate_history(_history_for_group(l, group), curr_digest, list(curr_status or [])): + for (path, status) in results: # newest-first: first seen is the latest verdict + if path.rule in owned_rules and path.rule not in latest: + latest[path.rule] = status + return { + rule for rule in owned_rules + if rule not in expected_to_fail and latest.get(rule) != "VERIFIED" + } + + +@dataclass(frozen=True) +class GroupRun: + """One verification group's execution decision for a single verify pass.""" + group: VerificationGroup + #: The group's current spec-state digest (keys its completion history). + digest: str + #: Owned rules to (re-)submit this pass. Empty means the group is already + #: fully covered at this digest, so its prover run is skipped entirely. + pending: frozenset[str] + + +def plan_group_execution( + groups: Sequence[VerificationGroup], + *, + history: list[ProverHistoryItem], + all_rules: list[str], + agent_rules: list[str] | None, + agent_exclude: list[str] | None, + expected_to_fail: set[str], + digest_of: Callable[[VerificationGroup], str], +) -> list[GroupRun]: + """Decide, per group, which owned rules this verify pass (re-)submits. + + Each group's pending set is its owned rules not yet verified at its current + digest (:func:`group_pending_rules`) intersected with any explicit agent rule + selection. An empty pending set marks a group already fully covered — the + executor skips its run. Pure and total (never runs the prover), so the + parallel executor and this decision can be tested apart.""" + if agent_rules is not None: + selection = set(agent_rules) + elif agent_exclude is not None: + selection = set(all_rules) - set(agent_exclude) + else: + selection = set(all_rules) + plan: list[GroupRun] = [] + for group in groups: + digest = digest_of(group) + pending = group_pending_rules( + history, + group=group.name, + curr_digest=digest, + owned_rules=set(group.owned_rules), + expected_to_fail=expected_to_fail, + ) + plan.append(GroupRun(group=group, digest=digest, pending=frozenset(pending & selection))) + return plan + + def _merge_prover_history(left: list[ProverHistoryItem], right: list[ProverHistoryItem]) -> list[ProverHistoryItem]: to_ret = left.copy() to_ret.extend(right) @@ -269,6 +394,11 @@ class ProverStateExtra(TypedDict): # only ever replaced wholesale (commit_edit / revert_to_edit). vfs: NotRequired[dict[str, str]] + # The agent's declared verification-group partition (see agent_groups), set via the + # group-declaration tool; absent => the single-spec/single-run default. Replaced wholesale + # on redeclaration. + verification_groups: NotRequired[list[VerificationGroupSpec]] + type ProverEvents = CEXAnalysisStart | CloudPollingEvent | ProverOutputEvent | RuleAnalysisResult | ProverRun | ProverLink | ProverResult # ``verify_spec`` only runs in the source pipeline, whose state always seeds @@ -563,6 +693,162 @@ async def verify_spec( prover_msg = f"{component} iteration number {iteration}" + def _finalize_run( + *, + status_map: Mapping[RulePath, StatusCodes], + prover_update: list[ProverHistoryItem], + all_verified: bool, + content: str, + link: str | None, + ) -> Command: + """Post-run bookkeeping shared by the grouped and single-run paths: nag on rules stuck on + repeated identical failures, then build the tool_state_update. The caller supplies this + pass's own status map, history items, completion verdict, and result text/link.""" + stuck_rules = { + k: v for (k, v) in status_map.items() + if v in STUCK_STATUSES and k.rule not in state["rule_skips"] + } + known_tc_ids = { + l["id"] for msg in state["messages"] if isinstance(msg, AIMessage) + for l in msg.tool_calls if l["name"] == "verify_spec" + } + to_warn, seen_post_compaction_history = stuck_rule_warnings( + stuck_rules, state["prover_history"], known_tc_ids + ) + nag_channel: dict = {} + if len(to_warn) > 0: + prover_update.append(NagMarker(sort="nag", nagged_rules=list(to_warn))) + nag_channel["reminders_channel"] = [ + "The following rule(s) have had identical failures on the last 3 runs of the prover:", + *(f"- {it.pprint()}" for it in to_warn), + "You may need to significantly change your approach, or skip the property if this is a persistent issue (you may need to use rebuttals to communicate" + " these failures to the feedback judge).", + "If these are TIMEOUTs, re-running the same spec will not help. Consider " + "`declare_verification_groups` to split the properties into parallel runs, each keeping only " + "what it needs precise and summarizing the rest — a monolithic run pays the intersection of every " + "rule's precision needs.", + ] + if seen_post_compaction_history: + nag_channel["reminders_channel"].append( + "(NB: Some of these prover calls happened before your most recent task history summarization)" + ) + if all_verified: + nag_channel.setdefault("reminders_channel", []).append( + "You have successfully verified over your prior prover run(s) that all rules verify. This task is completed." + ) + return tool_state_update( + tool_call_id=tool_call_id, content=content, + prover_link=link, validations=stamper(state, state["version_history"]), + prover_history=prover_update, **nag_channel + ) + return tool_state_update( + tool_call_id=tool_call_id, content=content, prover_link=link, + prover_history=prover_update, **nag_channel + ) + + async def run_grouped( + run_root: str, all_rules: list[str], groups: list[VerificationGroup] + ) -> str | Command: + """Verify a multi-group partition: each group runs its pending owned rules + under its own spec/conf, concurrently, and the per-group verdicts recombine. + + Groups run in parallel via ``asyncio.gather`` — on cloud, ``sem`` is a no-op, + so the submissions are genuinely concurrent; each group writes its spec/conf + under a group-distinct name so the parallel same-stem writes don't race. + Completion is the AND of per-group completeness (a fully-covered group is + skipped and passes trivially); history entries are tagged with their group so + :func:`group_is_complete` evaluates each group over its own digest streak.""" + expected = set(state["rule_skips"].keys()) + + def digest_of(g: VerificationGroup) -> str: + return spec_digest( + g.spec_contents if g.spec_contents is not None else spec, + state["skipped"], state["version_history"], + ) + + # Drop phantom owned rules — declared but absent from the compiled spec; warn once. + groups, phantom = prune_phantom_owned_rules(groups, all_rules) + if phantom: + _logger.warning( + "verification groups: declared rule(s) absent from the compiled spec, ignored: %s", + sorted(phantom), + ) + + plan = plan_group_execution( + groups, history=state["prover_history"], all_rules=all_rules, + agent_rules=rules, agent_exclude=exclude_rules, + expected_to_fail=expected, digest_of=digest_of, + ) + + async def run_one(gr: GroupRun) -> tuple[GroupRun, ProverReport | str | None]: + if not gr.pending: + return (gr, None) # already fully covered — skip its run + g = gr.group + gspec = g.spec_contents if g.spec_contents is not None else spec + gstem = f"{spec_stem}__{g.name}" if spec_stem is not None else None + with setup_prover_config_in( + working_dir=run_root, + main_contract=main_contract, + spec_stem=gstem, + spec_contents=gspec, + conf_dir=conf_dir, + config={**conf, **g.conf_overlay}, + rule=sorted(gr.pending), + exclude_rule=None, + msg=f"{component} [{g.name}] iteration number {iteration}", + ) as (config_path, cfg): + async with sem: + res = await run_prover( + Path(run_root), [config_path], tool_call_id, prover_opts, + _SpecCallbacks(get_stream_writer(), tool_call_id, summary, cfg, + analysis_store=analysis_store), + DefaultCexHandler(llm, state, summarization_threshold=10), + ) + return (gr, res) + + runs = await asyncio.gather(*(run_one(gr) for gr in plan)) + + # A hard toolchain error (str) aborts the pass; otherwise sort into the + # groups that actually ran vs. those skipped as already-covered. + outcomes: list[tuple[GroupRun, ProverReport | None]] = [] + for (gr, res) in runs: + if isinstance(res, str): + return res + outcomes.append((gr, res)) + executed = [(gr, res) for (gr, res) in outcomes if res is not None] + merged = merge_group_results([(gr.group, res.raw_rule_status) for (gr, res) in executed]) + combined_str = "\n\n".join( + f"=== group {gr.group.name} ===\n{res.result_str}" for (gr, res) in executed + ) or "All verification groups already covered by prior runs; nothing to re-run." + link = next((res.link for (_gr, res) in executed if res.link is not None), None) + + prover_update: list[ProverHistoryItem] = [ + ProverRunLog( + tool_call_id=tool_call_id, + prover_results=[(k, v) for (k, v) in res.raw_rule_status.items()], + rules={"sort": "include", "selector": sorted(gr.pending)}, + spec_digest=string_hash(gr.group.spec_contents if gr.group.spec_contents is not None else spec), + sort="run", + declared_rules=all_rules, + state_digest=gr.digest, + group=gr.group.name, + ) + for (gr, res) in executed + ] + all_verified = all( + group_is_complete( + state["prover_history"], group=gr.group.name, curr_digest=gr.digest, + expected_to_fail=expected, + curr_status=[(k, v) for (k, v) in res.raw_rule_status.items()] if res is not None else [], + owned_rules=set(gr.group.owned_rules), + ) + for (gr, res) in outcomes + ) + return _finalize_run( + status_map=merged, prover_update=prover_update, + all_verified=all_verified, content=combined_str, link=link, + ) + async def run_in(run_root: str) -> str | Command: with setup_prover_config_in( working_dir=run_root, @@ -582,6 +868,19 @@ async def run_in(run_root: str) -> str | Command: ) except SpecCompilationError as exc: return f"The spec failed to compile:\n{exc.output}" + # Groups from the agent's declaration, else one shared-spec group. Only a genuine + # split — more than one group, or a group carrying its own spec/conf — takes + # run_grouped; a trivial single group falls through to the single-run path below. + declared = state.get("verification_groups") + if declared: + groups = groups_from_specs( + spec, list(declared), + cap=resolved_max_groups(), + ) + else: + groups = single_group(all_rules) + if len(groups) > 1 or groups[0].spec_contents is not None or groups[0].conf_overlay: + return await run_grouped(run_root, all_rules, groups) with setup_prover_config_in( working_dir=run_root, main_contract=main_contract, @@ -607,26 +906,10 @@ async def run_in(run_root: str) -> str | Command: if isinstance(result, str): return result - stuck_rules = { - k: v for (k,v) in result.raw_rule_status.items() if v in ("TIMEOUT", "ERROR", "SANITY_FAILED") and k.rule not in state["rule_skips"] - } - - known_tc_ids = { - l["id"] - for msg in state["messages"] if isinstance(msg, AIMessage) - for l in msg.tool_calls if l["name"] == "verify_spec" - } - - to_warn, seen_post_compaction_history = stuck_rule_warnings( - stuck_rules, state["prover_history"], known_tc_ids - ) - curr_state_digest = spec_digest( spec, state["skipped"], state["version_history"] ) - - prover_results : list[tuple[RulePath, StatusCodes]] = [(k, v) for (k,v) in result.raw_rule_status.items()] - + prover_results: list[tuple[RulePath, StatusCodes]] = [(k, v) for (k, v) in result.raw_rule_status.items()] all_verified = _is_completion_history( l=state["prover_history"], curr_digest=curr_state_digest, @@ -634,11 +917,10 @@ async def run_in(run_root: str) -> str | Command: curr_status=prover_results, all_rules=all_rules ) - - prover_update : list[ProverHistoryItem] = [ + prover_update: list[ProverHistoryItem] = [ ProverRunLog( tool_call_id=tool_call_id, - prover_results=[(k, v) for (k,v) in result.raw_rule_status.items()], + prover_results=prover_results, rules={"sort": "exclude", "selector": exclude_rules } if exclude_rules is not None else \ {"sort": "include", "selector": rules} if rules is not None else None, spec_digest=spec_hash, @@ -647,40 +929,9 @@ async def run_in(run_root: str) -> str | Command: state_digest=curr_state_digest ) ] - nag_channel = { - - } - if len(to_warn) > 0: - prover_update.append(NagMarker( - sort="nag", - nagged_rules=list(to_warn) - )) - nag_channel["reminders_channel"] = [ - "The following rule(s) have had identical failures on the last 3 runs of the prover:", - *(f"- {it.pprint()}" for it in to_warn), - "You may need to significantly change your approach, or skip the property if this is a persistent issue (you may need to use rebuttals to communicate" - " these failures to the feedback judge)." - ] - if seen_post_compaction_history: - nag_channel["reminders_channel"].append( - "(NB: Some of these prover calls happened before your most recent task history summarization)" - ) - if all_verified: - nag_channel.setdefault("reminders_channel", []).append( - "You have successfully verified over your prior prover run(s) that all rules verify. This task is completed." - ) - # Completing the coverage stamps, however the completing run was scoped: - # every declared rule was verified against exactly this authoring state - # (the state_digest match), so a piecemeal completion is as good as a - # full-run one. - return tool_state_update( - tool_call_id=tool_call_id, content=result.result_str, - prover_link=result.link, validations=stamper(state, state["version_history"]), - prover_history=prover_update, **nag_channel - ) - return tool_state_update( - tool_call_id=tool_call_id, content=result.result_str, prover_link=result.link, - prover_history=prover_update, **nag_channel + return _finalize_run( + status_map=result.raw_rule_status, prover_update=prover_update, + all_verified=all_verified, content=result.result_str, link=result.link, ) # The author's working copy decides where this run executes (in-situ for diff --git a/composer/spec/source/verification_groups.py b/composer/spec/source/verification_groups.py new file mode 100644 index 00000000..4dbd2de5 --- /dev/null +++ b/composer/spec/source/verification_groups.py @@ -0,0 +1,226 @@ +"""General-purpose partitioning of a spec's rules into independent Certora +verification runs ("verification groups"). + +A *verification group* is a subset of a spec's rules verified in its own prover +run, under its own spec / conf configuration. Groups exist so that different +rules can run under verification setups a single run cannot express. The +hard driver is CVL itself: every imported ``methods{}`` block is merged globally +within one spec, so giving different rules different summarization *requires* +splitting them into different spec files (hence different confs, hence different +runs). But splitting is deliberately not summarization-specific — a group may +equally carry its own ``loop_iter``, link/dispatch setup, ``global_timeout`` or +``prover_args``, or exist only to isolate one expensive rule. + +This module is policy-neutral. It owns: + * the group model (:class:`VerificationGroup`), + * the group-count cap and its env override, + * the cap-driven greedy merge (:func:`cap_groups`), and + * result aggregation across groups (:func:`merge_group_results`). + +It does NOT decide *why* rules are split — which rules share a group, and each +group's spec/conf configuration. That is a populating policy's job (e.g. the +summarization-footprint clustering), which constructs the groups this module +then bounds and whose results it recombines. With a single group covering every +rule, this machinery is a behavior-preserving pass-through of the current +one-spec/one-run model. +""" + +import logging +import os +from collections.abc import Callable, Mapping, Sequence +from dataclasses import dataclass, field, replace + +from composer.prover.ptypes import RulePath, StatusCodes + +_logger = logging.getLogger("composer.prover") + + +# A group count above this is merged down (see cap_groups). Each group is a +# separate prover run, so the cap bounds run fan-out (cost / parallelism) and the +# worst case of one-group-per-rule; at 1 the whole spec runs as a single group. +# Overridable per run via the env var, mirroring the AUTOPROVER_* prover-config knobs. +DEFAULT_MAX_VERIFICATION_GROUPS = 6 +MAX_VERIFICATION_GROUPS_ENV = "AUTOPROVER_MAX_VERIFICATION_GROUPS" + + +def resolved_max_groups() -> int: + """The verification-group cap: ``DEFAULT_MAX_VERIFICATION_GROUPS``, or the + integer value of ``$AUTOPROVER_MAX_VERIFICATION_GROUPS`` when set. Values + below 1, and non-integers, are ignored with a warning (a cap of 0 groups is + meaningless).""" + raw = os.environ.get(MAX_VERIFICATION_GROUPS_ENV) + if raw is None: + return DEFAULT_MAX_VERIFICATION_GROUPS + try: + value = int(raw) + except ValueError: + _logger.warning("Ignoring non-integer %s=%r", MAX_VERIFICATION_GROUPS_ENV, raw) + return DEFAULT_MAX_VERIFICATION_GROUPS + if value < 1: + _logger.warning("Ignoring %s=%r (must be >= 1)", MAX_VERIFICATION_GROUPS_ENV, raw) + return DEFAULT_MAX_VERIFICATION_GROUPS + return value + + +@dataclass(frozen=True) +class VerificationGroup: + """One independent verification run over a subset of a spec's rules. + + Groups partition the rule set: every rule is *owned* by exactly one group, + and that group's run is authoritative for its verdict (:func:`merge_group_results`). + """ + + #: Stable identifier, used in conf/spec names and logs. + name: str + #: Rules whose verdict is taken from this group's run. Partition-disjoint + #: across groups. + owned_rules: frozenset[str] + #: Per-group spec text. ``None`` means "use the shared spec unchanged" — the + #: single-group / behavior-preserving case. A populating policy sets this when + #: the group needs a distinct spec (e.g. a different ``methods{}`` block). + spec_contents: str | None = None + #: Per-group conf overlay merged onto the base config for this group's run + #: (e.g. ``{"loop_iter": 2}``). Empty means no overlay. + conf_overlay: Mapping[str, object] = field(default_factory=dict) + #: The summaries this group installs, APPENDED to the base spec's ``methods{}`` block + #: (:func:`append_summaries`): function -> the ``methods{}`` entry (opaque text — NONDET, + #: a monotone / injective ghost, a model, …). A function absent here is left as the base + #: spec has it — PRECISE only if the base spec itself leaves it unsummarized; a base-global + #: summary (e.g. a curated/oracle model) still applies. Groups add on top of the shared base, + #: they do not remove its summaries. + summaries: Mapping[str, str] = field(default_factory=dict) + + +def append_summaries(base_spec: str, summaries: Mapping[str, str]) -> str: + """The base spec plus a ``methods{}`` block of this group's ``summaries`` (each value a + full methods entry), in stable sorted-by-function order. Empty summaries return the base + spec unchanged. CVL merges ``methods`` blocks, so appending is sound; a ghost the base + spec defines but no installed summary uses is harmless.""" + if not summaries: + return base_spec + block = "methods {\n" + "\n".join(f" {summaries[f]}" for f in sorted(summaries)) + "\n}\n" + return base_spec.rstrip() + "\n\n// --- verification group: summaries installed here ---\n" + block + + +def merge_summaries(a: Mapping[str, str], b: Mapping[str, str]) -> dict[str, str]: + """The summaries two groups can BOTH keep when merged into one run: a function is kept only + where both groups summarize it identically; any disagreement (different text, or only one + group summarizes it) drops it to PRECISE. Order-free and always sound — dropping a summary + only adds precision — so it assumes no summary-strength ordering (summaries are not generally + comparable: e.g. 'monotone' and 'injective' are incomparable).""" + return {f: a[f] for f in a.keys() & b.keys() if a[f] == b[f]} + + +def single_group( + all_rules: Sequence[str], + *, + name: str = "all", + spec_contents: str | None = None, +) -> list[VerificationGroup]: + """The trivial partition: one group owning every rule, no per-group spec/conf. + + This is the behavior-preserving default — routing a run through + ``single_group`` reproduces the current one-spec/one-run model exactly.""" + return [VerificationGroup(name=name, owned_rules=frozenset(all_rules), spec_contents=spec_contents)] + + +def _default_merge_pair(a: VerificationGroup, b: VerificationGroup) -> VerificationGroup: + """Combine two groups when neither carries a distinct spec: union the owned rules, keep + the agreed summaries (:func:`merge_summaries`), keep the shared spec, and merge conf + overlays (``b`` wins on key conflicts). A policy that gives groups distinct + ``spec_contents`` must pass its own merge (it alone knows how to rebuild the merged spec + from the merged summaries); this default is correct for summary-free / conf-overlay-only + groups.""" + merged_overlay: dict[str, object] = {**a.conf_overlay, **b.conf_overlay} + return replace( + a, + name=f"{a.name}+{b.name}", + owned_rules=a.owned_rules | b.owned_rules, + summaries=merge_summaries(a.summaries, b.summaries), + conf_overlay=merged_overlay, + ) + + +def cap_groups( + groups: Sequence[VerificationGroup], + cap: int, + merge_pair: Callable[[VerificationGroup, VerificationGroup], VerificationGroup] = _default_merge_pair, +) -> list[VerificationGroup]: + """Merge ``groups`` down to at most ``cap`` groups, cheapest merges first. + + Each group is a prover run, so an unbounded partition (worst case: one group + per rule) must be bounded. When there are more groups than ``cap``, this + repeatedly merges the pair whose footprints are most similar — the pair whose + merged footprint is smallest — via ``merge_pair``. Merging keeps the *union* + of both groups' footprints precise, so it only ever removes summarization + (adds precision): sound, but slower. Merging the most-similar pair first sheds + the least precision per step. At ``cap == 1`` everything collapses into one + group (today's monolith). A partition already within ``cap`` is returned as-is + (a fresh list). + """ + if cap < 1: + raise ValueError(f"cap must be >= 1, got {cap}") + remaining = list(groups) + if len(remaining) <= cap: + return remaining + + def merge_cost(a: VerificationGroup, b: VerificationGroup) -> int: + # Summaries this merge would drop to precise (kept only where both agree): fewer = + # less precision lost. Ties broken by combined rule count (prefer merging the smaller + # groups, so no single run grows unnecessarily large). + lost = len(a.summaries.keys() | b.summaries.keys()) - len(merge_summaries(a.summaries, b.summaries)) + return lost * 100_000 + len(a.owned_rules) + len(b.owned_rules) + + while len(remaining) > cap: + best: tuple[int, int, int] | None = None # (cost, i, j) + for i in range(len(remaining)): + for j in range(i + 1, len(remaining)): + cost = merge_cost(remaining[i], remaining[j]) + if best is None or cost < best[0]: + best = (cost, i, j) + assert best is not None # len(remaining) > cap >= 1 => at least 2 groups + _cost, i, j = best + merged = merge_pair(remaining[i], remaining[j]) + # Remove the higher index first so the lower stays valid. + remaining.pop(j) + remaining.pop(i) + remaining.append(merged) + return remaining + + +def prune_phantom_owned_rules( + groups: Sequence[VerificationGroup], all_rules: Sequence[str] +) -> tuple[list[VerificationGroup], set[str]]: + """Remap each group's owned rules to those the compiled spec actually declares, and return the + dropped "phantom" rules — names owned by a group but absent from ``all_rules`` (an agent typo, or a + ``property_rules`` entry naming a non-existent rule). + + Left in, a phantom owned rule is never submitted (the submit set is filtered to ``all_rules``) yet + forever counts as pending, so its group would never complete — a silent perpetual re-run. Returns the + groups unchanged and an empty set when every owned rule is real, so the caller warns only on a genuine + mistake.""" + actual = frozenset(all_rules) + phantom = {r for g in groups for r in g.owned_rules} - actual + if not phantom: + return list(groups), set() + return [replace(g, owned_rules=g.owned_rules & actual) for g in groups], phantom + + +def merge_group_results( + per_group: Sequence[tuple[VerificationGroup, Mapping[RulePath, StatusCodes]]], +) -> dict[RulePath, StatusCodes]: + """Recombine per-group prover verdicts into one verdict map. + + For each group, keep only the statuses of the rules that group *owns*. A run's status map + carries rules the group does not own — the prover's always-run built-in checks (e.g. + ``envfreeFuncsStaticCheck``) and parametric instantiations, plus (where the group's rule is + shared with another) a verdict computed under the wrong group's precision — all of which are + dropped. The union over the owned sets is the authoritative status of every rule. With one + group owning all rules, this returns that group's map unchanged. + """ + combined: dict[RulePath, StatusCodes] = {} + for group, statuses in per_group: + for path, status in statuses.items(): + if path.rule in group.owned_rules: + combined[path] = status + return combined diff --git a/composer/templates/property_generation_prompt.j2 b/composer/templates/property_generation_prompt.j2 index d17757cf..d62ec1a9 100644 --- a/composer/templates/property_generation_prompt.j2 +++ b/composer/templates/property_generation_prompt.j2 @@ -165,6 +165,44 @@ You may also make use of the following CVL resources, available as CVL files. {% endfor %} {% endif %} +## Verify by splitting into groups — split PREEMPTIVELY when a single run would time out + +A spec has ONE global `methods{}` block, so a monolithic `verify_spec` verifies every rule under the +*intersection* of what all rules need precise: one property that needs a function exact forces EVERY rule +to pay that cost, and re-running the same spec never escapes it. So if you anticipate a timeout, or you +see properties with CONFLICTING precision needs — one property needs a function exact while another can +summarize it — do NOT run the monolith and wait for it to time out. **Split into verification groups +FIRST.** + +Use the `declare_verification_groups` tool: partition the properties into groups, each verified in its +own PARALLEL prover run under its OWN summarization. A function one group must keep exact can be +summarized in another. How: + +1. In your base spec (the VFS), write the rule/invariant bodies and any ghosts / CVL functions your + summaries need — but do NOT summarize those functions inline in the base spec. +2. Call `declare_verification_groups` with one group per cluster of properties that share a precision + need. Each group names its `property_rules` (the properties it verifies + the rule names that verify + them) and its `summaries` (for each function THIS group summarizes, the `methods{}` entry to use here, + e.g. `function MathUtils.uncheckedExp(uint256 a, uint256 b) internal returns (uint256) => expGhost(a, b);`). + A function a group omits is verified as the base spec has it there — precise unless the base spec + (including its imports) already summarizes it. Every non-skipped property must appear in exactly + one group. +3. Your group `summaries` are APPENDED to the base spec's `methods{}` block (which includes autosetup's + imported summaries). To CHANGE how an already-base-summarized function behaves in a group, your entry + must be MORE SPECIFIC than the base's: an exact `Contract.f(...)` overrides a wildcard `_.f(...)` + (most-specific wins). If the base already summarizes it EXACTLY for that contract, you CANNOT override + or remove it — a second exact entry is a duplicate and fails the typecheck — and there is currently no + way to make that function precise in one group. When you hit that wall, summarize a CALLER of the + function instead: the caller is usually not base-summarized, so your summary applies cleanly at its + boundary. + +Then `verify_spec` runs the groups in parallel and only (re-)runs rules not yet verified. In each group, +summarize everything its properties do NOT depend on and keep precise only what they do. The SAME +function may need different summaries in different groups — one property may allow `foo` monotone while +another needs it injective — so pick, per group, the weakest summary sound for that group's rules. Start +with a small number of groups aligned to the distinct precision needs you see, and refine as results +come in. + Use the available memory tools to track your progress through this algorithm, any important lessons about CVL you may have learned, or any other significant, relevant information to this task. In particular, be sure to update your memory before calling the prover to summarize diff --git a/tests/test_agent_groups.py b/tests/test_agent_groups.py new file mode 100644 index 00000000..62f44053 --- /dev/null +++ b/tests/test_agent_groups.py @@ -0,0 +1,178 @@ +"""Unit tests for the agent-declared, property-level group planner (transparent policy).""" + +import pytest + +from composer.spec.cvl_generation import PropertyRuleMapping +from composer.spec.source.agent_groups import ( + VerificationGroupSpec, + groups_from_specs, + over_cap_message, + render_group_plan_for_judge, + validate_declared_coverage, +) + +BASE = "// rules\ninvariant a() true;\nghost g(uint) returns uint;\n" +SORT_SUMMARY = "function KVL.sortByKey(uint256[] l) internal returns (uint256[]) => g(0);" +EXP_SUMMARY = "function MathUtils.uncheckedExp(uint256 a, uint256 b) internal returns (uint256) => g(a);" +# property title -> rule names +PROP_RULES = { + "P-bitmap": ["r_borrow", "r_collat"], + "P-accounting": ["r_supply", "r_premium"], + "P-misc": ["r_misc"], +} + + +def _spec(name, prop_rules, summaries=None): + return VerificationGroupSpec( + name=name, + property_rules=[ + PropertyRuleMapping(property_title=p, rules=rs) for p, rs in prop_rules + ], + summaries=summaries or {}, + ) + + +def _grp(name, props, summaries=None, rules=None): + # A group over property titles, sourcing each property's rules from `rules` (PROP_RULES + # by default); an unknown property maps to no rules. + rules = rules if rules is not None else PROP_RULES + return _spec(name, [(p, list(rules.get(p, []))) for p in props], summaries) + + +# --- coverage validation ---------------------------------------------------- + + +def test_coverage_ok(): + decls = [_grp("g1", ["P-bitmap"]), _grp("g2", ["P-accounting", "P-misc"])] + assert validate_declared_coverage( + decls, all_properties=set(PROP_RULES), skipped=set() + ) is None + + +def test_coverage_missing_property(): + decls = [_grp("g1", ["P-bitmap"])] + err = validate_declared_coverage(decls, all_properties=set(PROP_RULES), skipped=set()) + assert err is not None and "no group" in err + + +def test_coverage_duplicate_property(): + decls = [_grp("g1", ["P-bitmap"]), _grp("g2", ["P-bitmap", "P-accounting", "P-misc"])] + err = validate_declared_coverage(decls, all_properties=set(PROP_RULES), skipped=set()) + assert err is not None and "more than one group" in err + + +def test_coverage_skipped_property_not_required_nor_assignable(): + # P-misc is skipped -> need not be covered, but must not be assigned either. + ok = [_grp("g1", ["P-bitmap"]), _grp("g2", ["P-accounting"])] + assert validate_declared_coverage( + ok, all_properties=set(PROP_RULES), skipped={"P-misc"} + ) is None + bad = [_grp("g1", ["P-bitmap"]), _grp("g2", ["P-accounting", "P-misc"])] + err = validate_declared_coverage(bad, all_properties=set(PROP_RULES), skipped={"P-misc"}) + assert err is not None and "skipped" in err + + +def test_coverage_unknown_property(): + decls = [_grp("g1", ["P-bitmap"]), _grp("g2", ["P-accounting", "P-misc", "P-ghost"])] + err = validate_declared_coverage(decls, all_properties=set(PROP_RULES), skipped=set()) + assert err is not None and "unknown" in err + + +# --- build ------------------------------------------------------------------ + + +def test_build_expands_properties_to_rules_and_summaries(): + decls = [ + _grp("bitmap", ["P-bitmap"], summaries={"uncheckedExp": EXP_SUMMARY}), + _grp("rest", ["P-accounting", "P-misc"], summaries={"sortByKey": SORT_SUMMARY}), + ] + groups = groups_from_specs(BASE, decls, cap=4) + by = {g.name: g for g in groups} + # bitmap group installs the uncheckedExp summary and keeps sortByKey precise (not installed). + assert by["bitmap"].owned_rules == {"r_borrow", "r_collat"} + assert "KVL.sortByKey" not in by["bitmap"].spec_contents + assert "MathUtils.uncheckedExp" in by["bitmap"].spec_contents + # rest group is the mirror image and owns the other properties' rules. + assert by["rest"].owned_rules == {"r_supply", "r_premium", "r_misc"} + assert "MathUtils.uncheckedExp" not in by["rest"].spec_contents + assert "KVL.sortByKey" in by["rest"].spec_contents + + +def test_build_leaves_conf_overlay_at_substrate_default(): + # The agent tool does not expose conf_overlay; built groups carry the substrate default ({}). + decls = [_grp("g1", ["P-bitmap"]), _grp("g2", ["P-accounting", "P-misc"])] + groups = groups_from_specs(BASE, decls, cap=4) + assert all(g.conf_overlay == {} for g in groups) + + +def test_build_rule_partition_first_declaration_wins(): + # A rule shared by two properties placed in different groups is owned by the first. + shared = {"P-x": ["r1", "r2"], "P-y": ["r2", "r3"]} # r2 shared + decls = [_grp("gx", ["P-x"], rules=shared), _grp("gy", ["P-y"], rules=shared)] + groups = groups_from_specs(BASE, decls, cap=4) + by = {g.name: g for g in groups} + assert by["gx"].owned_rules == {"r1", "r2"} + assert by["gy"].owned_rules == {"r3"} # r2 already claimed by gx + # partition: every rule owned exactly once + owned = [r for g in groups for r in g.owned_rules] + assert sorted(owned) == ["r1", "r2", "r3"] + + +def test_build_asserts_declaration_within_cap(): + # groups_from_specs does NOT merge — the tool rejects an over-cap declaration at declare time + # (over_cap_message), so a >cap declaration reaching the builder is a programming error. + decls = [_grp(f"g{i}", [p]) for i, p in enumerate(["P-bitmap", "P-accounting", "P-misc"])] + with pytest.raises(AssertionError): + groups_from_specs(BASE, decls, cap=2) + + +# --- judge-facing group plan rendering -------------------------------------- + + +def test_over_cap_message_none_within_cap(): + specs = [_spec("a", [("P1", ["r1"])]), _spec("b", [("P2", ["r2"])])] + assert over_cap_message(specs, cap=4) is None + assert over_cap_message(specs, cap=2) is None + + +def test_over_cap_message_rejects_and_suggests_a_merge(): + # 3 groups, cap 2 -> reject. g1 and g2 agree on `foo` (cheapest merge, loses nothing); g3 is disjoint, + # so the suggested merge is g1+g2, which the message names so the agent can adopt or improve it. + specs = [ + _spec("g1", [("P1", ["r1"])], summaries={"foo": "S"}), + _spec("g2", [("P2", ["r2"])], summaries={"foo": "S"}), + _spec("g3", [("P3", ["r3"])], summaries={"bar": "T"}), + ] + msg = over_cap_message(specs, cap=2) + assert msg is not None + assert "3 verification groups" in msg and "at most 2" in msg + assert "AUTOPROVER_MAX_VERIFICATION_GROUPS" in msg + assert "g1+g2" in msg and "g3" in msg + + +def test_render_plan_none_when_no_groups(): + assert render_group_plan_for_judge([]) is None + + +def test_render_plan_shows_per_group_installed_summaries(): + specs = [ + _spec("bitmap", [("P-bitmap", ["r_borrow", "r_collat"])], summaries={"uncheckedExp": EXP_SUMMARY}), + _spec( + "acct", [("P-accounting", ["r_supply"])], + summaries={"sortByKey": SORT_SUMMARY, "uncheckedExp": EXP_SUMMARY}, + ), + ] + out = render_group_plan_for_judge(specs) + assert out is not None + # The bitmap group lists exactly the one summary it installs; sortByKey is precise there. + bitmap = out.split('Group "bitmap"')[1].split('Group "acct"')[0] + assert "installs summaries:" in bitmap + assert "uncheckedExp:" in bitmap + assert "sortByKey" not in bitmap + # The acct group installs both summaries. + acct = out.split('Group "acct"')[1] + assert "uncheckedExp:" in acct and "sortByKey:" in acct + # No conf is rendered — the judge is given the spec, not the .conf (and groups carry no conf). + assert "conf" not in out.lower() + # It is clearly marked informational so the judge does not treat it as spec text. + assert "informational" in out.lower() diff --git a/tests/test_group_completion.py b/tests/test_group_completion.py new file mode 100644 index 00000000..a3b12a43 --- /dev/null +++ b/tests/test_group_completion.py @@ -0,0 +1,194 @@ +"""Unit tests for per-group completion / pending-rule selection (Layer 2 engine). + +Pure logic over synthetic prover history — no prover. Focuses on the incremental +re-run behavior and the interleaving hazard: because groups have distinct spec +digests and ``_iterate_history`` stops at the first foreign digest, a group's +completion must be computed over its own filtered history, not the flat list. +""" + +from composer.prover.ptypes import RulePath +from composer.spec.source.prover import ( + ProverRunLog, + group_is_complete, + group_pending_rules, +) + + +def _run(group, digest, results, *, tcid="t", declared=None): + """A ProverRunLog with the given (rule, status) results at a digest+group.""" + paths = [(RulePath(rule=r), s) for r, s in results] + return ProverRunLog( + tool_call_id=tcid, + prover_results=paths, + spec_digest=digest, + rules=None, + sort="run", + declared_rules=declared if declared is not None else [r for r, _ in results], + state_digest=digest, + group=group, + ) + + +# --- pending-rule selection (incremental engine) ---------------------------- + + +def test_pending_excludes_already_verified(): + hist = [_run("A", "dA", [("r1", "VERIFIED"), ("r2", "VIOLATED")])] + pending = group_pending_rules( + hist, group="A", curr_digest="dA", owned_rules={"r1", "r2"}, expected_to_fail=set() + ) + assert pending == {"r2"} # r1 done, r2 still needs work + + +def test_pending_includes_never_run(): + pending = group_pending_rules( + [], group="A", curr_digest="dA", owned_rules={"r1", "r2"}, expected_to_fail=set() + ) + assert pending == {"r1", "r2"} + + +def test_pending_uses_latest_verdict_not_ever_verified(): + # r1 verified in an older run, then TIMEOUT in a newer run at the same digest: + # it is still pending (latest verdict wins), so it must be re-run. + hist = [ + _run("A", "dA", [("r1", "VERIFIED")], tcid="old"), + _run("A", "dA", [("r1", "TIMEOUT")], tcid="new"), + ] + pending = group_pending_rules( + hist, group="A", curr_digest="dA", owned_rules={"r1"}, expected_to_fail=set() + ) + assert pending == {"r1"} + + +def test_pending_forgives_expected_to_fail(): + hist = [_run("A", "dA", [("r1", "VIOLATED")])] + pending = group_pending_rules( + hist, group="A", curr_digest="dA", owned_rules={"r1"}, expected_to_fail={"r1"} + ) + assert pending == set() + + +def test_pending_ignores_other_group_and_stale_digest(): + hist = [ + _run("B", "dB", [("r1", "VERIFIED")]), # wrong group + _run("A", "OLD", [("r1", "VERIFIED")]), # right group, stale spec digest + ] + pending = group_pending_rules( + hist, group="A", curr_digest="dA", owned_rules={"r1"}, expected_to_fail=set() + ) + assert pending == {"r1"} # neither counts + + +# --- the interleaving hazard ------------------------------------------------ + + +def test_interleaved_foreign_group_does_not_truncate_completion(): + # Group A verified all its rules at dA (run 1). Then group B ran at dB (run 2, + # a different spec/digest). A flat _iterate_history for dA would STOP at run 2 + # (foreign digest) and lose run 1 -> falsely incomplete. Filtering to group A's + # own history first must keep A complete. + hist = [ + _run("A", "dA", [("r1", "VERIFIED"), ("r2", "VERIFIED")]), + _run("B", "dB", [("r3", "VIOLATED")]), + ] + assert group_is_complete( + hist, group="A", curr_digest="dA", expected_to_fail=set(), + curr_status=[], owned_rules={"r1", "r2"}, + ) + # ...and pending for A is empty despite the interleaved B run. + assert group_pending_rules( + hist, group="A", curr_digest="dA", owned_rules={"r1", "r2"}, expected_to_fail=set() + ) == set() + + +def test_group_complete_requires_all_owned_verified(): + hist = [_run("A", "dA", [("r1", "VERIFIED")])] + assert not group_is_complete( + hist, group="A", curr_digest="dA", expected_to_fail=set(), + curr_status=[], owned_rules={"r1", "r2"}, # r2 never verified + ) + + +def test_group_complete_counts_curr_status(): + # r2 verified in the just-finished run (curr_status), not yet in history. + hist = [_run("A", "dA", [("r1", "VERIFIED")])] + assert group_is_complete( + hist, group="A", curr_digest="dA", expected_to_fail=set(), + curr_status=[(RulePath(rule="r2"), "VERIFIED")], owned_rules={"r1", "r2"}, + ) + + +def test_untagged_history_is_default_group_none(): + # Backward compat: a legacy run with no `group` key belongs to group None. + legacy = ProverRunLog( + tool_call_id="t", prover_results=[(RulePath(rule="r1"), "VERIFIED")], + spec_digest="dA", rules=None, sort="run", declared_rules=["r1"], state_digest="dA", + ) + assert group_is_complete( + [legacy], group=None, curr_digest="dA", expected_to_fail=set(), + curr_status=[], owned_rules={"r1"}, + ) + + +# --- plan_group_execution (parallel-run decision) --------------------------- + +from composer.spec.source.prover import GroupRun, plan_group_execution # noqa: E402 +from composer.spec.source.verification_groups import VerificationGroup # noqa: E402 + + +def _grp(name, rules, spec=None): + return VerificationGroup(name=name, owned_rules=frozenset(rules), spec_contents=spec) + + +def _digest_by_name(g): # each group a distinct digest keyed by its name + return f"d_{g.name}" + + +def test_plan_skips_fully_covered_group(): + # Group A already verified both owned rules at its digest; group B has none run. + hist = [_run("A", "d_A", [("r1", "VERIFIED"), ("r2", "VERIFIED")])] + plan = plan_group_execution( + [_grp("A", ["r1", "r2"]), _grp("B", ["r3", "r4"])], + history=hist, all_rules=["r1", "r2", "r3", "r4"], + agent_rules=None, agent_exclude=None, expected_to_fail=set(), + digest_of=_digest_by_name, + ) + by = {gr.group.name: gr for gr in plan} + assert by["A"].pending == frozenset() # skipped + assert by["B"].pending == frozenset({"r3", "r4"}) + + +def test_plan_reruns_only_pending_within_group(): + hist = [_run("A", "d_A", [("r1", "VERIFIED"), ("r2", "TIMEOUT")])] + plan = plan_group_execution( + [_grp("A", ["r1", "r2"])], history=hist, all_rules=["r1", "r2"], + agent_rules=None, agent_exclude=None, expected_to_fail=set(), digest_of=_digest_by_name, + ) + assert plan[0].pending == frozenset({"r2"}) # r1 stays verified, only r2 re-runs + + +def test_plan_intersects_agent_rule_selection(): + plan = plan_group_execution( + [_grp("A", ["r1", "r2"]), _grp("B", ["r3"])], + history=[], all_rules=["r1", "r2", "r3"], + agent_rules=["r2"], agent_exclude=None, expected_to_fail=set(), digest_of=_digest_by_name, + ) + by = {gr.group.name: gr for gr in plan} + assert by["A"].pending == frozenset({"r2"}) # only the agent-selected rule + assert by["B"].pending == frozenset() # r3 not selected -> skipped + + +def test_plan_honors_exclude_selection(): + plan = plan_group_execution( + [_grp("A", ["r1", "r2"])], history=[], all_rules=["r1", "r2"], + agent_rules=None, agent_exclude=["r1"], expected_to_fail=set(), digest_of=_digest_by_name, + ) + assert plan[0].pending == frozenset({"r2"}) + + +def test_plan_records_per_group_digest(): + plan = plan_group_execution( + [_grp("A", ["r1"]), _grp("B", ["r2"])], history=[], all_rules=["r1", "r2"], + agent_rules=None, agent_exclude=None, expected_to_fail=set(), digest_of=_digest_by_name, + ) + assert {gr.group.name: gr.digest for gr in plan} == {"A": "d_A", "B": "d_B"} diff --git a/tests/test_verification_groups.py b/tests/test_verification_groups.py new file mode 100644 index 00000000..41c4279b --- /dev/null +++ b/tests/test_verification_groups.py @@ -0,0 +1,147 @@ +"""Unit tests for the general verification-group substrate (Layer 1). + +Pure-logic tests: the group cap, the cap-driven greedy merge, and result +recombination. No prover or LLM involved. +""" + +import pytest + +from composer.prover.ptypes import RulePath +from composer.spec.source.verification_groups import ( + DEFAULT_MAX_VERIFICATION_GROUPS, + MAX_VERIFICATION_GROUPS_ENV, + VerificationGroup, + cap_groups, + merge_group_results, + prune_phantom_owned_rules, + resolved_max_groups, + single_group, +) + + +def _g(name, rules, summaries=None): + return VerificationGroup(name=name, owned_rules=frozenset(rules), summaries=summaries or {}) + + +# --- single_group / behavior-preserving default ----------------------------- + + +def test_single_group_owns_all_rules(): + groups = single_group(["a", "b", "c"]) + assert len(groups) == 1 + assert groups[0].owned_rules == {"a", "b", "c"} + assert groups[0].spec_contents is None + assert groups[0].conf_overlay == {} + + +# --- resolved_max_groups ---------------------------------------------------- + + +def test_max_groups_default_when_unset(monkeypatch): + monkeypatch.delenv(MAX_VERIFICATION_GROUPS_ENV, raising=False) + assert resolved_max_groups() == DEFAULT_MAX_VERIFICATION_GROUPS + + +def test_max_groups_env_override(monkeypatch): + monkeypatch.setenv(MAX_VERIFICATION_GROUPS_ENV, "2") + assert resolved_max_groups() == 2 + + +@pytest.mark.parametrize("bad", ["0", "-3", "notanint", ""]) +def test_max_groups_invalid_falls_back(monkeypatch, bad): + monkeypatch.setenv(MAX_VERIFICATION_GROUPS_ENV, bad) + assert resolved_max_groups() == DEFAULT_MAX_VERIFICATION_GROUPS + + +# --- cap_groups ------------------------------------------------------------- + + +def test_cap_noop_when_within_cap(): + groups = [_g("x", ["a"]), _g("y", ["b"])] + capped = cap_groups(groups, cap=4) + assert capped == groups + + +def test_cap_to_one_collapses_everything(): + groups = [_g("x", ["a"], {"f1": "S1"}), _g("y", ["b"], {"f2": "S2"}), _g("z", ["c"], {"f3": "S3"})] + capped = cap_groups(groups, cap=1) + assert len(capped) == 1 + # The single surviving group owns every rule; the disjoint summaries all disagree, + # so every function drops to precise — i.e. the monolithic run. + assert capped[0].owned_rules == {"a", "b", "c"} + assert capped[0].summaries == {} + + +def test_cap_merges_most_agreeing_summaries_first(): + # x and y summarize f1 IDENTICALLY; z summarizes a disjoint f9. Capping 3->2 must merge + # x+y (cheapest: they agree on f1, losing nothing) and leave z alone. + x = _g("x", ["a"], {"f1": "S1"}) + y = _g("y", ["b"], {"f1": "S1"}) + z = _g("z", ["c"], {"f9": "S9"}) + capped = cap_groups([x, y, z], cap=2) + assert len(capped) == 2 + by_rules = {frozenset(g.owned_rules): g for g in capped} + assert frozenset({"a", "b"}) in by_rules # x+y merged + assert frozenset({"c"}) in by_rules # z untouched + assert by_rules[frozenset({"a", "b"})].summaries == {"f1": "S1"} + + +def test_merge_drops_disagreed_summaries_to_precise(): + # Both summarize f1 but INCOMPARABLY differently -> the merged group keeps neither + # (drops f1 to precise); no summary-strength ordering is assumed. + x = _g("x", ["a"], {"f1": "monotone"}) + y = _g("y", ["b"], {"f1": "injective"}) + capped = cap_groups([x, y], cap=1) + assert capped[0].summaries == {} + + +def test_cap_below_one_rejected(): + with pytest.raises(ValueError): + cap_groups([_g("x", ["a"])], cap=0) + + +def test_cap_partition_disjoint_after_merges(): + groups = [_g(str(i), [f"r{i}"], {f"f{i % 2}": f"S{i % 2}"}) for i in range(6)] + capped = cap_groups(groups, cap=2) + assert len(capped) == 2 + owned = [r for g in capped for r in g.owned_rules] + assert sorted(owned) == sorted(f"r{i}" for i in range(6)) # every rule kept exactly once + + +# --- merge_group_results ---------------------------------------------------- + + +def test_merge_results_keeps_only_owned_rules(): + ga = _g("A", ["r1", "r2"]) + gb = _g("B", ["r3"]) + # Group A's run also instantiated r3 (referenced), but under A's precision it + # must be ignored; r3's authoritative verdict comes from B. + res_a = {RulePath(rule="r1"): "VERIFIED", RulePath(rule="r2"): "VIOLATED", RulePath(rule="r3"): "TIMEOUT"} + res_b = {RulePath(rule="r3"): "VERIFIED"} + combined = merge_group_results([(ga, res_a), (gb, res_b)]) + assert combined[RulePath(rule="r1")] == "VERIFIED" + assert combined[RulePath(rule="r2")] == "VIOLATED" + assert combined[RulePath(rule="r3")] == "VERIFIED" # from B, not A's TIMEOUT + + +def test_merge_results_single_group_is_passthrough(): + g = _g("all", ["r1", "r2"]) + res = {RulePath(rule="r1"): "VERIFIED", RulePath(rule="r2"): "VIOLATED"} + assert merge_group_results([(g, res)]) == res + + +def test_prune_phantom_owned_rules_drops_undeclared_and_reports_them(): + # A rule owned by a group but absent from the compiled spec (agent typo) would otherwise be pending + # forever (never submitted, never verified) and wedge the group in a perpetual re-run. Prune it + + # report it so the caller can warn. + g1 = VerificationGroup(name="a", owned_rules=frozenset({"r1", "ghostRule"})) + g2 = VerificationGroup(name="b", owned_rules=frozenset({"r2"})) + pruned, phantom = prune_phantom_owned_rules([g1, g2], ["r1", "r2"]) + assert phantom == {"ghostRule"} + assert {g.name: set(g.owned_rules) for g in pruned} == {"a": {"r1"}, "b": {"r2"}} + + +def test_prune_phantom_owned_rules_noop_when_all_real(): + groups = [VerificationGroup(name="a", owned_rules=frozenset({"r1"}))] + pruned, phantom = prune_phantom_owned_rules(groups, ["r1", "r2"]) + assert phantom == set() and pruned == groups # unchanged (value-equal frozen dataclasses)