diff --git a/certora_autosetup/cache/content_cache.py b/certora_autosetup/cache/content_cache.py index 691749ff..7499e842 100644 --- a/certora_autosetup/cache/content_cache.py +++ b/certora_autosetup/cache/content_cache.py @@ -15,6 +15,19 @@ from certora_autosetup.utils.constants import DIR_CERTORA_INTERNAL, DIR_CONTENT_CACHE +def hash_text(text: str) -> str: + """SHA-256 of a string's UTF-8 bytes (first 16 hex chars) — the in-memory analog of + :meth:`ContentCache._hash_file`, for content that isn't on disk (e.g. an in-state spec buffer).""" + return hashlib.sha256(text.encode()).hexdigest()[:16] + + +def hash_content_parts(parts: list[str]) -> str: + """Combine already-hashed content parts (e.g. ``"name:"`` and ``"extra:"`` strings) + into one 32-char cache key. The final step shared by :meth:`ContentCache.compute_cache_key` and + any in-memory content digest.""" + return hashlib.sha256("\n".join(parts).encode()).hexdigest()[:32] + + class ContentCache: """Content-hash-based cache for arbitrary data keyed by file contents. @@ -72,8 +85,7 @@ def compute_cache_key( for part in extra_key_parts: parts.append(f"extra:{part}") - combined = "\n".join(parts) - return hashlib.sha256(combined.encode()).hexdigest()[:32] + return hash_content_parts(parts) def get(self, cache_key: str) -> dict[str, Any] | None: """Retrieve cached data for the given key. diff --git a/composer/authoring/tools.py b/composer/authoring/tools.py index 571770df..f395dddb 100644 --- a/composer/authoring/tools.py +++ b/composer/authoring/tools.py @@ -259,7 +259,7 @@ def _verify_attempts(state: GatedGiveUpState) -> int: for msg in state.get("messages", []) if isinstance(msg, AIMessage) for call in msg.tool_calls - if call["name"] == "verify_spec" + if call["name"] == "submit_buffer" ) diff --git a/composer/prover/callbacks.py b/composer/prover/callbacks.py index 7c66af8b..a1de0786 100644 --- a/composer/prover/callbacks.py +++ b/composer/prover/callbacks.py @@ -1,5 +1,5 @@ """Stream-event prover callbacks shared by the codegen prover tool and the -source-pipeline ``verify_spec`` tool. +source-pipeline buffer prover jobs. ``ProverEventCallbacks`` translates the ``ProverCallbacks`` lifecycle into the custom stream events the UI renders, keyed by tool_call_id. Both prover entry diff --git a/composer/scripts/budget_math.py b/composer/scripts/budget_math.py index 962cde00..9571bd71 100644 --- a/composer/scripts/budget_math.py +++ b/composer/scripts/budget_math.py @@ -329,7 +329,7 @@ def headroom_note(cap: float) -> str: "groups/`prover_links` exclude curtailed", "- rendered HTML (`autoprove-report-render`) shows the budget appendix", "- thread trail (`ap-trail export` + this script): the `` wrap-up " - "appears in the author transcript; no `verify_spec`/`feedback_tool` calls after it", + "appears in the author transcript; no `submit_buffer`/`feedback_tool` calls after it", "- `components_to_prover_runs.json` lacks curtailed entries", "", "## Caveats", diff --git a/composer/spec/cvl_generation.py b/composer/spec/cvl_generation.py index 6941cd56..fe5995f5 100644 --- a/composer/spec/cvl_generation.py +++ b/composer/spec/cvl_generation.py @@ -57,6 +57,14 @@ class Rebuttal(RebuttalBase): etc. Do NOT file rebuttals for feedback you merely disagree with; address those by revising the spec. """ + buffer: str = Field( + description=( + "The name of the review unit whose feedback this rebuts — a buffer name, or " + '"skips_review" for the shared skip review. Each unit is reviewed by its own judge, so a ' + "rebuttal reaches only that unit's judge; file it under the unit the prior-round feedback " + "was about." + ) + ) evidence_type: Literal[ "typecheck_failure", "counterexample", @@ -96,6 +104,10 @@ class GeneratedCVL(BaseModel): # The last prover-run link (URL or local results dir), persisted for the report and so a # cache hit retains it. None when the prover never produced a link. final_link: str | None = Field(default=None) + # Every prover-run link whose results compose the buffers at their final digests, deduped (empty + # when no run-target buffer has a completed run at its final digest). With rule-striping a buffer's + # rules are run across several jobs, so this holds all of them, not just the last ``final_link``. + run_links: list[str] = Field(default_factory=list) # The author's working copy at completion: the edited source files the proof # actually ran against (empty when no edits were applied — always the case # outside the editing-enabled source pipeline), and the provenance of each @@ -288,16 +300,24 @@ async def _get_feedback( def _version_history(self) -> Sequence[str]: return () +def cvl_guidance_tools() -> list[BaseTool]: + """The dependency-free CVL *guidance* tools — no spec-writing tools. Used by the buffer-authoring + agent, which writes CVL through the buffer tools (put_buffer / edit_buffer) rather than put_cvl.""" + return [ + ERC20TokenGuidance.as_tool("erc20_guidance"), + UnresolvedCallGuidance.as_tool("unresolved_call_guidance"), + ] + + def static_tools() -> list[BaseTool]: - """The dependency-free CVL authoring tools. The property-management suite - (feedback / skip tools) is NOT here — it carries runtime deps; see + """The dependency-free CVL authoring tools — the single-``curr_spec`` writing tools plus guidance. + The property-management suite (feedback / skip tools) is NOT here — it carries runtime deps; see :func:`skip_tools` and :class:`FeedbackToolBase`.""" return [ put_cvl, put_cvl_raw, get_cvl(CVLGenerationState), edit_cvl(CVLGenerationState), - ERC20TokenGuidance.as_tool("erc20_guidance"), - UnresolvedCallGuidance.as_tool("unresolved_call_guidance"), + *cvl_guidance_tools(), ] diff --git a/composer/spec/source/artifacts.py b/composer/spec/source/artifacts.py index 76760a49..2b7189f1 100644 --- a/composer/spec/source/artifacts.py +++ b/composer/spec/source/artifacts.py @@ -76,7 +76,7 @@ def _write_conf( self, spec: ComponentSpec, base_config: dict | None, spec_path: Path, ) -> None: """The prover conf for the run: the generation's final ``state["config"]`` plus - the fixed run overlay (shared with the live ``verify_spec`` run). No-op if no + the fixed run overlay (shared with the live prover run). No-op if no base config.""" if base_config is None: _log.warning("no base config for %s; skipping conf dump", spec.stem) diff --git a/composer/spec/source/author.py b/composer/spec/source/author.py index b88d0c38..95ecc4ee 100644 --- a/composer/spec/source/author.py +++ b/composer/spec/source/author.py @@ -5,7 +5,7 @@ import json import pathlib -from dataclasses import dataclass +from dataclasses import dataclass, field from langchain_core.tools import BaseTool from pydantic import Field, BaseModel, Discriminator @@ -14,30 +14,39 @@ WithAsyncImplementation, WithImplementation, WithInjectedId, WithInjectedState, WithAsyncDependencies ) -from graphcore.graph import tool_state_update, RawPromptInput, CacheMarker, SummaryConfig +from graphcore.graph import tool_state_update, tool_return, RawPromptInput, CacheMarker, SummaryConfig from graphcore.tools.vfs import VFSAccessor, VFSState from composer.authoring.judge import PropertyFeedbackProtocol -from composer.authoring.state import SkippedProperty, check_completion, spec_digest +from composer.authoring.state import SkippedProperty from composer.authoring.tools import gated_give_up_tool, give_up_tool from composer.spec.guidance import StructuralInvariantGuidance from composer.spec.cvl_generation import ( - static_tools, property_tools, skip_tools, CVLGenerationExtra, FEEDBACK_VALIDATION_KEY, - validate_property_rules, CVL_JUDGE_KEY, run_cvl_generator, + cvl_guidance_tools, skip_tools, CVLGenerationExtra, FEEDBACK_VALIDATION_KEY, + CVL_JUDGE_KEY, run_cvl_generator, GeneratedCVL, PropertyRuleMapping, AppliedEdit, FeedbackToolBase, ) from composer.prover.core import run_prover, CexHandler, ProverCallbacks, ProverReport 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 -from composer.spec.types import PropertyFormulation, PropertyTitle +from composer.spec.source.spec_buffers import ( + SpecBuffersExtra, buffer_review_text, buffer_state_digest, check_buffer_completion, + combined_buffers_view, max_spec_buffers, requireinvariant_citations, run_targets, + skips_review_digest, SKIPS_VALIDATION_KEY, validate_coverage, validate_declared_rules_mapped, + validate_disjoint_rules, validate_requireinvariant_proved, +) +from composer.spec.source.buffer_tools import ( + put_buffer, get_buffer, edit_buffer, list_buffers, delete_buffer, +) +from composer.spec.context import WorkflowContext, CVLGeneration, CacheKey, CVLJudge, SourceCode +from composer.spec.types import PropertyFormulation, PropertyTitle, RuleName from composer.pipeline.core import GaveUp, ToolBinder, InjectingToolExtension, Curtailed from composer.pipeline.plugin_api import ProvidedTools from composer.spec.source.plugin import CertoraProverTools, CVLAuthorState from composer.spec.system_model import ContractComponentInstance, SolidityIdentifier, component_context from composer.spec.source.prover import ( OVERLAY_OWNED_KEYS, ProverStateExtra, DELETE_SKIP, VALIDATION_KEY as PROVER_VALIDATION_KEY, - covering_run_links, declared_rules_at, materializing_project, + materializing_project, completing_run_links, declared_rules_at, ) from langgraph.graph import MessagesState from pathlib import Path @@ -53,7 +62,7 @@ from langgraph.types import Command from graphcore.graph import Builder from composer.spec.feedback import ( - property_feedback_judge, source_feedback_judge, FeedbackTemplate, Properties, + source_feedback_judge, FeedbackTemplate, Properties, SourceSnapshot, ContextualFeedbackToolImpl, ) from composer.ui.tool_display import tool_display @@ -87,7 +96,7 @@ class SourceAuthorExtra(TypedDict): # ``vfs`` comes from ProverStateExtra (NotRequired, no merge op — replaced # wholesale by commit_edit / revert_to_edit); the generation input always # seeds it explicitly. -class SourceCVLGenerationExtra(CVLGenerationExtra, ProverStateExtra, SourceAuthorExtra, VersionedHistory): +class SourceCVLGenerationExtra(CVLGenerationExtra, ProverStateExtra, SourceAuthorExtra, VersionedHistory, SpecBuffersExtra): pass class SourceCVLGenerationInput(SourceCVLGenerationExtra, FlowInput): @@ -151,36 +160,58 @@ class PublishResultTool( Call to signal your completed cvl generation. """ commentary: str = Field(description="Commentary on your generated spec") - property_rules: list[PropertyRuleMapping] = Field( - description="The property->rules mapping. For every property you did NOT skip " - "(referenced by its unique snake_case title from the batch listing), list the " - "name(s) of the rule(s)/invariant(s) in your spec that verify it. Every non-skipped " - "property must appear with at least one rule." - ) @override async def run(self) -> Command | str: - st = self.state - if (err := check_completion(st, st["version_history"])) is not None: + # Completion requires every run-target buffer verified AND reviewed at its current digest + # (per-buffer stamps), with a clean property/rule partition across buffers. Each run-target + # declares its own property->rules on ``put_buffer``, so the published mapping is derived + # from the buffers. When every property is skipped there are no run-target buffers: the + # stamp check is then vacuous and ``validate_coverage`` alone decides whether publishing is + # allowed. + buffers = self.state.get("buffers") or {} + skipped_pairs = [(str(s.property_title), str(s.reason)) for s in self.state["skipped"]] + if (err := check_buffer_completion( + buffers, self.state["validations"], self.state["required_validations"], + skipped=skipped_pairs, version_history=self.state["version_history"], + )) is not None: return err - spec = st["curr_spec"] - assert spec is not None, "check_completion admits no spec-less state" - # What the typechecker actually found in the published spec, so the mapping is checked - # in both directions: a supporting invariant the author proved but never tied back to a - # property would otherwise be dropped from the report as an orphan. - declared = declared_rules_at( - st["prover_history"], spec_digest(spec, st["skipped"], st["version_history"]) - ) with self.tool_deps() as titles: - if (err := validate_property_rules( - self.property_rules, st["skipped"], titles, declared - )) is not None: - return err + skip_titles = {str(s.property_title) for s in self.state["skipped"]} + if (err := validate_coverage(buffers, all_properties=set(titles), skipped=skip_titles)) is not None: + return f"Completion REJECTED: {err}" + if (err := validate_disjoint_rules(buffers)) is not None: + return f"Completion REJECTED: {err}" + # Reverse of validate_coverage: every rule/invariant the typechecker declared in a buffer must be + # named in that buffer's property_rules, so the mapping accounts for everything proved (nothing + # proved is left attributed to no property). Read each buffer's declared set off its completing + # prover run (None when no run covered it — a lifted publish gate). + declared_by_buffer = { + b.name: declared_rules_at( + self.state["prover_history"], + buffer_state_digest(buffers, b.name, version_history=self.state["version_history"]), + ) + for b in run_targets(buffers) + } + if (err := validate_declared_rules_mapped(buffers, declared_by_buffer)) is not None: + return f"Completion REJECTED: {err}" + # Every invariant a buffer cites with requireInvariant must be declared (hence proved) in that + # same buffer: an imported invariant is not re-verified in the importing run, so citing one that + # lives only in another buffer — e.g. an unproven shared buffer — is an unproven assumption. + cited_by_buffer = {b.name: requireinvariant_citations(b.cvl) for b in run_targets(buffers)} + if (err := validate_requireinvariant_proved( + buffers, cited_by_buffer, declared_by_buffer + )) is not None: + return f"Completion REJECTED: {err}" + pr = [ + PropertyRuleMapping(property_title=PropertyTitle(p), rules=[RuleName(rn) for rn in rs]) + for b in run_targets(buffers) for p, rs in b.property_rules.items() + ] return tool_state_update( self.tool_call_id, "Accepted", result=self.commentary, - property_rules=self.property_rules, + property_rules=pr, failed=False, ) @@ -638,33 +669,123 @@ def judge_tools(self) -> tuple[BaseTool, ...]: ) +@dataclass +class _PerBufferJudge[J]: + """Lazily builds and caches one persistent judge per buffer, keyed by buffer name — each bound to + that buffer's claimed properties, on its own child context so its review memory stays scoped to the + buffer. Rebuilt only when the buffer's claimed properties change; because the child namespace is + derived from the name, the rebuilt judge keeps its memory. ``properties`` is the batch's full set, + for resolving each buffer's subset.""" + build: Callable[[str, list[PropertyFormulation]], J] + properties: list[PropertyFormulation] + _cache: dict[str, tuple[tuple[str, ...], J]] = field(default_factory=dict) + + def for_buffer(self, name: str, claimed: list[PropertyFormulation]) -> J: + sig = tuple(sorted(str(p.title) for p in claimed)) + cached = self._cache.get(name) + if cached is None or cached[0] != sig: + self._cache[name] = (sig, self.build(name, claimed)) + return self._cache[name][1] + + @tool_display("Getting feedback", "Feedback") class EditorAwareFeedbackTool( FeedbackToolBase[SourceCVLGenerationState], - WithAsyncDependencies[Command, ContextualFeedbackToolImpl[SourceSnapshot]], + WithAsyncDependencies[Command, _PerBufferJudge[ContextualFeedbackToolImpl[SourceSnapshot]]], ): + # Reviews each run-target buffer independently — against the properties that buffer claims — with its + # own persistent per-buffer judge (reached via _review), stamping feedback:; the skip set is + # reviewed once as a standalone unit (see run()). With no run-target buffers there is nothing to review. __doc__ = FeedbackToolBase.__doc__ @override - async def _get_feedback( - self, spec: str, skipped: list[SkippedProperty] + async def run(self) -> Command: + buffers = self.state.get("buffers") or {} + targets = run_targets(buffers) + if not targets: + return tool_return(self.tool_call_id, "No run-target buffers to review yet.") + + # Review each run-target buffer whose feedback stamp is missing or stale (its text, an import, or + # its claimed properties changed) in isolation, scored against the properties it claims, and stamp + # feedback: per approved buffer — so an approved, unchanged buffer is never re-reviewed and + # the hard buffer is reviewed alone. The claimed properties are part of the feedback digest + # (include_claim), so re-assigning a property re-triggers review even with unchanged CVL. Skips are + # NOT reviewed here; they are reviewed once, below, since a skip belongs to no single buffer. + skipped = self.state["skipped"] + skipped_pairs = [(str(s.property_title), str(s.reason)) for s in skipped] + vh = self._version_history() + validations = self.state["validations"] + all_props = self._all_properties() + + def digest(name: str) -> str: + return buffer_state_digest(buffers, name, version_history=vh, include_claim=True) + + new_stamps: dict[str, str] = {} + blocks: list[str] = [] + + for b in [b for b in targets if validations.get(f"feedback:{b.name}") != digest(b.name)]: + claimed = [p for p in all_props if str(p.title) in b.property_rules] + verdict = await self._review(b.name, buffer_review_text(buffers, b.name), [], claimed) + blocks.append(f"=== buffer {b.name} ===\nGood? {verdict.good}\nFeedback {verdict.feedback}") + if verdict.good: + new_stamps[f"feedback:{b.name}"] = digest(b.name) + + # Skip quality is a whole-spec concern, so review the skip set ONCE (a skip is owned by no + # buffer; scrutinizing it in every buffer's judge would make any skip change re-review every + # buffer). The judge sees the whole spec (a skip's justification can rest on the code) and only + # the skips, with no per-buffer claim to cover. + skips_digest = skips_review_digest(buffers, skipped=skipped_pairs, version_history=vh) + if skipped and validations.get(SKIPS_VALIDATION_KEY) != skips_digest: + verdict = await self._review(SKIPS_VALIDATION_KEY, combined_buffers_view(buffers), skipped, []) + blocks.append(f"=== skipped properties ===\nGood? {verdict.good}\nFeedback {verdict.feedback}") + if verdict.good: + new_stamps[SKIPS_VALIDATION_KEY] = skips_digest + + if not blocks: + return tool_return( + self.tool_call_id, "All buffers already reviewed and approved at their current state." + ) + return tool_state_update(self.tool_call_id, "\n\n".join(blocks), validations=new_stamps) + + async def _review( + self, name: str, spec: str, skipped: list[SkippedProperty], + properties: list[PropertyFormulation], ) -> PropertyFeedbackProtocol: - with self.tool_deps() as judge: + # Review one unit — a buffer's text, or the whole spec for the skip review — with that unit's + # cached judge, scored against `properties` (the unit's claimed subset; empty for the skip review). + with self.tool_deps() as judges: assert "vfs" in self.state snap = SourceSnapshot( vfs=self.state["vfs"], version_history=self.state["version_history"], ) - return await judge(snap, spec, skipped, self.rebuttals, self.tool_call_id) + # Each unit's judge sees only the rebuttals filed against its own feedback. + rebuttals = [r for r in self.rebuttals if r.buffer == name] + return await judges.for_buffer(name, properties)( + snap, spec, skipped, rebuttals, self.tool_call_id + ) + + def _all_properties(self) -> list[PropertyFormulation]: + # The batch's full property set, for resolving a unit's claimed subset. + with self.tool_deps() as judges: + return judges.properties @override def _version_history(self) -> Sequence[str]: return self.state["version_history"] + @override + async def _get_feedback( + self, spec: str, skipped: list[SkippedProperty] + ) -> PropertyFeedbackProtocol: + # run() reviews per unit through _review; this single-spec entry is unreachable (curr_spec is + # always None in buffer mode, and run() handles the no-buffers case directly). Present only to + # satisfy the abstract base. + raise AssertionError("buffer feedback uses per-unit _review, not _get_feedback") + _PropertyGenTemplate = TypedTemplate[PropertyGenParams]("property_generation_prompt.j2") -_PROPERTY_GEN_SYS_PROMPT = "property_generation_system_prompt.j2" #: The prover's tool extension: contributions come from plugins deriving #: ``CertoraProverTools``, dispatched via their ``certora_prover_tools`` hook. @@ -674,7 +795,8 @@ def _version_history(self) -> Sequence[str]: @dataclass class ProverTool: - lg_tool: BaseTool + #: The async multi-buffer tools (submit_buffer / collect_results) the agent verifies with. + buffer_tools: list[BaseTool] options: ProverOptions @dataclass @@ -793,16 +915,18 @@ async def batch_cvl_generation( }) sys_prompt : list[RawPromptInput | type[CacheMarker]] = [ - lambda load: load(_PROPERTY_GEN_SYS_PROMPT) + lambda load: load("property_generation_system_prompt.j2"), + f"\nCreate at most {max_spec_buffers()} run-target buffers; fold further properties into " + f"existing ones. A single run-target buffer is the one-spec case.", ] added_tools : list[BaseTool] = [] task_host = TaskHost() kit = editing_tools.editing - # The same run-root strategy verify_spec uses (see ProjectDirectory): an - # empty working copy is read in-situ, a non-empty one against a temporary - # materialization whose lifetime is the contributed tool's invocation. - project_directory = materializing_project(source.project_root, kit.live.mat) + # Run-root strategy (see ProjectDirectory): an empty working copy is read + # in-situ, a non-empty one against a temporary materialization whose lifetime + # is the contributed tool's invocation. + project_directory = materializing_project(kit.live.mat) @asynccontextmanager async def yield_state( @@ -827,7 +951,7 @@ async def propose( async with project_directory(st.get("vfs") or {}) as run_root: yield CVLAuthorState( working_dir=pathlib.Path(run_root), - curr_spec=st["curr_spec"], + buffers=st.get("buffers") or {}, prover_runner=WrappedProverRunner( st["config"], prover_tool.options, @@ -864,11 +988,17 @@ async def propose( "source_editing": True, }) protected = focus.protected if focus is not None else () - judge_impl = source_feedback_judge( - judge_ctx, _LiveJudgeHost(env, editing), judge_prompt, props + # One persistent judge per buffer, bound to that buffer's claimed properties, on its own child + # context (memory scoped to the buffer, and kept across a rebuild when the claim changes). + judge_host = _LiveJudgeHost(env, editing) + source_judges = _PerBufferJudge( + build=lambda name, claimed: source_feedback_judge( + judge_ctx.child(CacheKey[CVLJudge, CVLJudge](name)), judge_host, judge_prompt, claimed + ), + properties=props, ) feedback_suite = [ - EditorAwareFeedbackTool.bind(judge_impl).as_tool("feedback_tool"), + EditorAwareFeedbackTool.bind(source_judges).as_tool("feedback_tool"), *skip_tools(titles, protected=protected), ] @@ -886,8 +1016,18 @@ async def propose( ).with_tools( generate_edit_management_tools(ctx, env, editing.store, editing.live) ) + # Multi-buffer authoring is the only mode: the agent writes CVL through the buffer tools and + # verifies each run-target buffer with the async submit_buffer / collect_results pair. Guidance-only + # CVL tools are bound (no put_cvl/edit_cvl). + buffer_authoring: list[BaseTool] = [ + put_buffer(SourceCVLGenerationState), get_buffer(SourceCVLGenerationState), + edit_buffer(SourceCVLGenerationState), list_buffers(SourceCVLGenerationState), + delete_buffer(SourceCVLGenerationState), + ] task_graph = b.with_tools( - static_tools() + cvl_guidance_tools() + ).with_tools( + buffer_authoring ).with_tools( # Prover-only: the natspec author shares ``static_tools()`` but has no # prover and so no counterexample to remediate. @@ -895,7 +1035,7 @@ async def propose( ).with_tools( feedback_suite ).with_tools( - [prover_tool.lg_tool, + [*prover_tool.buffer_tools, ExpectRulePassage.as_tool("expect_rule_passage"), ExpectRuleFailure.as_tool("expect_rule_failure"), give_up_tool(name="give_up", description=_GIVE_UP_DESCRIPTION, label="CVL generation") @@ -969,6 +1109,7 @@ async def propose( vfs=restored_vfs, version_history=restored_history, spec_stem=spec_stem, + buffers={}, plugin_tools=[t.name for inj in tools for t in inj.tools], ) ) @@ -994,8 +1135,10 @@ async def propose( # unformalizable" judgment — it's the budget talking. Keep the agent's account. return Curtailed(None, detail=res_state["result"]) return GaveUp(reason=res_state["result"]) - d = res_state["curr_spec"] - assert d is not None + # The published artifact is the buffers combined into one document (empty when every property + # was skipped, i.e. there are no buffers to combine). + _buffers = res_state.get("buffers") or {} + d = combined_buffers_view(_buffers) applied_edits: list[AppliedEdit] = [] for edit_id in res_state["version_history"]: rec = await editing.store.read(edit_id) @@ -1009,6 +1152,12 @@ async def propose( # hit (which skips the prover) can still reconstruct certora/confs and retain the link. assert "vfs" in res_state + # Every run link that composes the buffers at their final digests, so verdicts spread across + # striped/per-buffer runs are all reachable from the result. + run_links = completing_run_links( + res_state["prover_history"], res_state.get("buffers") or {}, + version_history=res_state["version_history"], + ) generated = GeneratedCVL( commentary=res_state["result"], cvl=d, @@ -1016,12 +1165,9 @@ async def propose( property_rules=res_state["property_rules"], config=res_state["config"], final_link=res_state.get("prover_link"), + run_links=run_links, vfs=res_state["vfs"], applied_edits=applied_edits, - covering_links=covering_run_links( - res_state["prover_history"], - spec_digest(d, res_state["skipped"], res_state["version_history"]), - ), ) if res_state["budget_curtailed"]: # Published under lifted gates: hand it back as an explicitly unreliable partial. diff --git a/composer/spec/source/buffer_tools.py b/composer/spec/source/buffer_tools.py new file mode 100644 index 00000000..8895309f --- /dev/null +++ b/composer/spec/source/buffer_tools.py @@ -0,0 +1,191 @@ +"""Agent-facing tools for authoring several named CVL spec buffers — the multi-buffer generalization +of the single ``curr_spec`` tools in :mod:`composer.authoring.buffer`. + +Each tool reads and writes ``state["buffers"][name]`` (a :class:`NamedBuffer`) instead of the single +``curr_spec`` string, reusing the shared CVL validator (:func:`cvl_syntax_error`) and the +surgical-edit primitive (:func:`replace_unique`). Writes go through the buffers-map reducer, so an +edit to one buffer leaves the others untouched. Each tool is a pydantic model generic over the +concrete graph state; the factory subscribes it to that state and mints the tool. +""" + +from langchain_core.tools import BaseTool +from langgraph.types import Command +from pydantic import Field +from typing_extensions import TypedDict + +from graphcore.graph import tool_state_update +from graphcore.tools.schemas import WithImplementation, WithInjectedId, WithInjectedState + +from composer.core.edit import EditErr, EditOk, replace_unique +from composer.cvl.tools import cvl_syntax_error +from composer.spec.source.spec_buffers import ( + NamedBuffer, buffer_imports, duplicated_declarations, max_spec_buffers, +) +from composer.ui.tool_display import ToolDisplay, suppress_ack, tool_display_of + + +class WithBuffers(TypedDict): + buffers: dict[str, NamedBuffer] + + +def _dup_note(buffers: dict[str, NamedBuffer], name: str) -> str: + """A warning listing declarations in buffer ``name`` that also appear verbatim in another run-target + buffer, or "" if there are none. Restricted to ``name``: dups among other buffers were flagged when + those buffers were written.""" + dups = {d: [n for n in ns if n != name] + for d, ns in duplicated_declarations(buffers).items() if name in ns} + if not dups: + return "" + lines = "\n".join(f" {d} (also in {ns})" for d, ns in dups.items()) + return ("\n\nNOTE: these declarations are duplicated in other run-target buffers — consider moving " + "each to a shared buffer the duplicating buffers import, before proving:\n" + lines) + + +_put_display = ToolDisplay("Writing spec buffer", suppress_ack("Buffer write result")) +_get_display = ToolDisplay("Reading spec buffer", None) +_edit_display = ToolDisplay("Editing spec buffer", suppress_ack("Buffer edit result")) +_list_display = ToolDisplay("Listing spec buffers", None) +_delete_display = ToolDisplay("Deleting spec buffer", suppress_ack("Buffer delete result")) + + +class PutBuffer[S: WithBuffers](WithImplementation[str | Command], WithInjectedState[S], WithInjectedId): + """Create or replace a whole CVL spec buffer, identified by `name`. A buffer is a self-contained + spec: its own rules, its `methods{}` block, and `import` statements pulling in shared buffers. The + text is run through the CVL parser; if it fails to parse the update is rejected and the buffer is + unchanged. + + Set `is_run_target` false for a shared buffer that only supplies imports (ghosts/invariants/models) + and runs no rules of its own. To depend on a shared buffer, just `import ".spec";` in this + buffer's CVL — the dependency is read from those import statements, so editing a shared buffer + correctly re-verifies exactly the buffers that import it. Re-putting an existing buffer keeps its + property->rule mapping. + + The number of run-target buffers is capped; creating one past the cap is refused — fold those + properties into an existing run-target buffer instead.""" + + name: str = Field(description="Unique buffer name (also its on-disk spec stem).") + cvl: str = Field(description="The buffer's full CVL text (rules, methods{}, imports).") + property_rules: dict[str, list[str]] = Field( + default_factory=dict, + description="The properties this buffer verifies and, for each (by its snake_case title), the " + "rule/invariant names in this buffer's CVL that verify it. Across all run-target buffers every " + "non-skipped property must appear in exactly one buffer. Omit for a shared buffer.", + ) + is_run_target: bool = Field( + default=True, description="False for a shared, imported-only buffer that runs no rules." + ) + + def run(self) -> str | Command: + if (err := cvl_syntax_error(self.cvl)) is not None: + return err + buffers_now = self.state.get("buffers") or {} + # Cap how far the agent partitions: a new run-target buffer beyond the cap is refused. + if self.is_run_target and self.name not in buffers_now: + cap = max_spec_buffers() + if sum(1 for b in buffers_now.values() if b.is_run_target) >= cap: + return ( + f"Refusing to create run-target buffer {self.name!r}: the run-target buffer cap " + f"({cap}) is already reached. Fold these properties into an existing run-target " + f"buffer instead." + ) + existing = buffers_now.get(self.name) + # Keep the prior property->rule mapping when the agent re-puts text without restating it. + prop_rules = self.property_rules or (dict(existing.property_rules) if existing else {}) + buf = NamedBuffer( + name=self.name, cvl=self.cvl, + is_run_target=self.is_run_target, property_rules=prop_rules, + ) + return tool_state_update( + tool_call_id=self.tool_call_id, + content="Accepted" + _dup_note({**buffers_now, self.name: buf}, self.name), + buffers={self.name: buf}, + ) + + +def put_buffer[S: WithBuffers](ty: type[S]) -> BaseTool: + return tool_display_of(_put_display)(PutBuffer[ty].as_tool("put_buffer")) + + +class EditBuffer[S: WithBuffers](WithImplementation[str | Command], WithInjectedState[S], WithInjectedId): + """Make a surgical edit to one spec buffer instead of re-emitting it. Provide `name`, an exact + `old_string` span copied from that buffer (must occur exactly once — include context to + disambiguate), and `new_string`. The edited buffer is re-parsed; if it fails to parse the edit is + rejected and the buffer is unchanged. Dramatically cheaper than `put_buffer` for a small change.""" + + name: str = Field(description="The buffer to edit.") + old_string: str = Field(description="Exact span to replace; must occur exactly once.") + new_string: str = Field(description="Replacement text.") + + def run(self) -> str | Command: + existing = (self.state.get("buffers") or {}).get(self.name) + if existing is None: + return f"No buffer named {self.name!r}. Create it with put_buffer first." + match replace_unique(existing.cvl, self.old_string, self.new_string): + case EditErr(message=msg): + return msg + case EditOk(text=new_text): + if (err := cvl_syntax_error(new_text)) is not None: + return err + buf = existing.model_copy(update={"cvl": new_text}) + buffers_now = self.state.get("buffers") or {} + return tool_state_update( + tool_call_id=self.tool_call_id, + content="Accepted" + _dup_note({**buffers_now, self.name: buf}, self.name), + buffers={self.name: buf}, + ) + + +def edit_buffer[S: WithBuffers](ty: type[S]) -> BaseTool: + return tool_display_of(_edit_display)(EditBuffer[ty].as_tool("edit_buffer")) + + +class GetBuffer[S: WithBuffers](WithImplementation[str], WithInjectedState[S]): + """Read one spec buffer's current CVL text.""" + + name: str = Field(description="The buffer to read.") + + def run(self) -> str: + buf = (self.state.get("buffers") or {}).get(self.name) + return buf.cvl if buf is not None else f"No buffer named {self.name!r}." + + +def get_buffer[S: WithBuffers](ty: type[S]) -> BaseTool: + return tool_display_of(_get_display)(GetBuffer[ty].as_tool("get_buffer")) + + +class ListBuffers[S: WithBuffers](WithImplementation[str], WithInjectedState[S]): + """List the spec buffers: name, kind, imports, and rule count.""" + + def run(self) -> str: + buffers: dict[str, NamedBuffer] = (self.state.get("buffers") or {}) + if not buffers: + return "No spec buffers yet." + lines = [] + for name in sorted(buffers): + b = buffers[name] + kind = "run-target" if b.is_run_target else "shared" + imps = buffer_imports(buffers, name) + imp = f", imports {sorted(imps)}" if imps else "" + lines.append(f"- {name} ({kind}, {len(b.owned_rules)} rules{imp})") + return "\n".join(lines) + + +def list_buffers[S: WithBuffers](ty: type[S]) -> BaseTool: + return tool_display_of(_list_display)(ListBuffers[ty].as_tool("list_buffers")) + + +class DeleteBuffer[S: WithBuffers](WithImplementation[str | Command], WithInjectedState[S], WithInjectedId): + """Delete a spec buffer (e.g. after merging its rules into another).""" + + name: str = Field(description="The buffer to delete.") + + def run(self) -> str | Command: + if self.name not in (self.state.get("buffers") or {}): + return f"No buffer named {self.name!r}." + return tool_state_update( + tool_call_id=self.tool_call_id, content="Deleted", buffers={self.name: None} + ) + + +def delete_buffer[S: WithBuffers](ty: type[S]) -> BaseTool: + return tool_display_of(_delete_display)(DeleteBuffer[ty].as_tool("delete_buffer")) diff --git a/composer/spec/source/pipeline.py b/composer/spec/source/pipeline.py index b52ac0c6..33c87c48 100644 --- a/composer/spec/source/pipeline.py +++ b/composer/spec/source/pipeline.py @@ -39,7 +39,7 @@ lift_harnessed, ) from composer.spec.source.summarizer import setup_summaries -from composer.spec.source.prover import get_prover_tool, materializing_project +from composer.spec.source.prover import get_prover_tool, materializing_project, ProverToolset from composer.spec.source.plugin import CertoraProverTools from composer.spec.source.author import ( batch_cvl_generation, EditingTools, FocusPolicy, SourceEditing, ProverTool, @@ -77,14 +77,14 @@ class _ProverPipelineDeps: analysis_store: CexAnalysisStore editing: SourceEditing - def to_prover_tool(self, tool: BaseTool) -> ProverTool: - return ProverTool(lg_tool=tool, options=self.prover_options) + def to_prover_tool(self, tools: ProverToolset) -> ProverTool: + return ProverTool(buffer_tools=tools.make_buffer_tools(), options=self.prover_options) @dataclass class ProverRunner(Formalizer[GeneratedCVL, ContractComponentInstance]): """Immutable formalizer: per-batch CVL generation against a fixed prover config + resource set.""" - _prover_tool: BaseTool + _prover_tool: ProverToolset _prover_config: dict _resources: list[CVLResource] _fetch: VerdictFetcher[GeneratedCVL] @@ -182,7 +182,7 @@ class ProverPrepared(PreparedSystem[GeneratedCVL, ContractComponentInstance, Con ``prepare_formalization``.""" _sys_desc: SystemDescriptionHarnessed _harnessed: HarnessedApplication - _prover_tool: BaseTool + _prover_tool: ProverToolset _analyzed: SourceApplication _deps: _ProverPipelineDeps @@ -259,7 +259,7 @@ async def prepare_system( # VFS (invariants, or an author that never edited) runs in-situ; a # non-empty one runs in a temp materialization of the working copy. prover_tool = get_prover_tool( - run.env.llm_heavy(), run.source.contract_name, materializing_project(run.source.project_root, self.editing.live.mat), + run.env.llm_heavy(), run.source.contract_name, materializing_project(self.editing.live.mat), prover_opts=self._prover_opts, analysis_store=self.analysis_store, ) return ProverPrepared( diff --git a/composer/spec/source/plugin.py b/composer/spec/source/plugin.py index 500e1e33..3e96a4c1 100644 --- a/composer/spec/source/plugin.py +++ b/composer/spec/source/plugin.py @@ -11,7 +11,7 @@ import pathlib from abc import abstractmethod from dataclasses import dataclass -from typing import AsyncContextManager, Callable, Sequence, Protocol +from typing import AsyncContextManager, Callable, Mapping, Sequence, Protocol from composer.pipeline.plugin_api import ( FormalizationTool, PipelinePlugin, PluginToolContext, ProvidedTools, @@ -20,6 +20,7 @@ from composer.spec.types import PropertyFormulation from composer.prover.core import ProverReport, ProverCallbacks, CexHandler from composer.io.task_host import TaskHost +from composer.spec.source.spec_buffers import NamedBuffer, buffer_review_text class ProverRunner(Protocol): """One ad-hoc prover run: stages the spec/conf into ``working_dir`` for the @@ -66,13 +67,22 @@ class CVLAuthorState: # The run root the prover would execute in: the project itself, or a temporary # materialization of the author's working copy (lifetime = the read). working_dir: pathlib.Path - curr_spec: str | None + # The spec under authoring, as its named buffers — each a self-contained CVL unit (its own rules, + # methods{}, and imports). Every rule lives in exactly one buffer; ``spec_for_rule`` returns the CVL + # for a given rule. + buffers: Mapping[str, NamedBuffer] prover_runner: ProverRunner host: TaskHost # Propose source edits for the author to apply; records carry the # proposing plugin's attribution. edit_store: EditProposer + def spec_for_rule(self, rule: str) -> str | None: + """The CVL the author has written for ``rule``: the buffer that declares it, together with its + transitive import closure, as one document — or None if no buffer declares ``rule``.""" + name = next((nm for nm, b in self.buffers.items() if rule in b.owned_rules), None) + return buffer_review_text(self.buffers, name) if name is not None else None + type ProverStateReader[T] = Callable[[T], AsyncContextManager[CVLAuthorState]] diff --git a/composer/spec/source/prover.py b/composer/spec/source/prover.py index 995166f2..b27bf395 100644 --- a/composer/spec/source/prover.py +++ b/composer/spec/source/prover.py @@ -1,17 +1,19 @@ """ -Spec-side prover tool: wraps composer/prover/core.py into a LangGraph tool. +Spec-side prover tools: wrap composer/prover/core.py into LangGraph tools. -Provides get_prover_tool() which creates a verify_spec tool that: -- Reads curr_spec from injected state -- Writes a temporary .spec file -- Runs the Certora prover via run_prover() -- Streams output/polling events via custom stream writer +Provides get_prover_tool(), whose submit_buffer / collect_results tools: +- Materialize each run-target buffer to a temporary .spec file +- Run the Certora prover via run_prover() as background per-buffer jobs +- Stream output/polling events via a custom stream writer +- Report results as jobs finish, without blocking on a whole batch """ import asyncio import json import logging +import os import time +from dataclasses import dataclass from contextlib import contextmanager, asynccontextmanager, ExitStack, nullcontext from pathlib import Path from typing import ( @@ -24,19 +26,19 @@ from composer.spec.source.live_explorer import VersionedHistory -from langchain_core.tools import InjectedToolCallId, tool, BaseTool +from langchain_core.tools import tool, BaseTool from langchain_core.messages import AIMessage -from langgraph.prebuilt import InjectedState from pydantic import BaseModel, Field, Discriminator from langgraph.config import get_stream_writer from langgraph.types import Command from composer.prover.ptypes import RuleResult, RulePath from graphcore.graph import LLM +from graphcore.tools.schemas import WithInjectedId, WithInjectedState from composer.prover.core import ( ProverOptions, SpecCompilationError, declared_rules_list, run_prover, - DefaultCexHandler + DefaultCexHandler, ProverReport ) from composer.prover.callbacks import ProverEventCallbacks from composer.prover.ptypes import StatusCodes @@ -54,6 +56,9 @@ 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.spec_buffers import ( + NamedBuffer, SpecBuffersExtra, buffer_state_digest, run_targets, +) _logger = logging.getLogger("composer.prover") @@ -62,7 +67,7 @@ OVERLAY_OWNED_KEYS: frozenset[str] = frozenset({ # forced by prover_config_overlay "verify", "parametric_contracts", "optimistic_loop", "rule_sanity", - # set per-run by verify_spec + # set per prover run "rule", "msg", }) """Config keys the run pipeline forces onto the base config after spreading it: a @@ -74,7 +79,7 @@ def prover_config_overlay(base_config: dict, *, main_contract: str, verify_target: str) -> dict: """The fixed prover settings the source pipeline layers on top of the base config. - Shared by the live ``verify_spec`` run and the persisted ``certora/confs`` dump so the + Shared by the live prover run and the persisted ``certora/confs`` dump so the two can't drift. ``verify_target`` is the ``:`` the run verifies. """ return { @@ -106,6 +111,28 @@ class RuleSelection(TypedDict): sort: Literal["exclude", "include"] selector: list[str] +def _selection_of(rule: list[str] | None, exclude_rules: list[str] | None) -> RuleSelection | None: + """The ``RuleSelection`` a submit_buffer call asks for, or None to run the whole buffer.""" + if rule is not None: + return RuleSelection(sort="include", selector=rule) + if exclude_rules is not None: + return RuleSelection(sort="exclude", selector=exclude_rules) + return None + +def _selection_key(sel: RuleSelection | None) -> str: + """A stable key distinguishing one buffer's rule selections, so striped runs (different subsets of + the same buffer at the same content) coexist as separate jobs instead of deduping each other. The + whole-buffer run keys to the empty string.""" + if sel is None: + return "" + return f"{sel['sort']}:{','.join(sorted(sel['selector']))}" + +def _apply_selection(config: dict, selection: RuleSelection | None) -> None: + """Write a rule subset onto a prover conf: ``rule`` for an include selection, ``exclude_rule`` for + an exclude one; a None selection leaves the conf running every rule.""" + if selection is not None: + config["rule" if selection["sort"] == "include" else "exclude_rule"] = list(selection["selector"]) + class ProverRunLog(TypedDict): tool_call_id: str prover_results: list[tuple[RulePath, StatusCodes]] @@ -114,10 +141,12 @@ class ProverRunLog(TypedDict): sort: Literal["run"] declared_rules: list[str] state_digest: str - #: The run's job link, so a verdict can be traced back to the run that produced it: a - #: scoped run's results are the only record of the rules it alone covered. ``NotRequired`` - #: because a thread checkpointed before this field existed replays without it. - link: NotRequired[str | None] + # The spec buffer this run belongs to; absent for a single-curr_spec run. Each buffer has its own + # spec/digest, so completion is evaluated per buffer over its own runs (see _history_for_buffer). + buffer: NotRequired[str] + # This run's prover-run link: the job URL (cloud) or local results dir the run wrote its output + # to; None for a run that produced no link. + link: str | None class NagMarker(TypedDict): nagged_rules: list[RulePath] @@ -323,6 +352,61 @@ def _is_completion_history( return True return False +def _history_for_buffer(l: list[ProverHistoryItem], buffer: str) -> list[ProverHistoryItem]: + """The prover history restricted to one buffer's runs (nag markers pass through). Each buffer has + its own spec, hence its own ``state_digest``; filtering first keeps :func:`_iterate_history`'s + digest streak from being truncated by an interleaved run of a different buffer.""" + return [it for it in l if it["sort"] != "run" or it.get("buffer") == buffer] + + +def buffer_is_complete( + l: list[ProverHistoryItem], + *, + buffer: str, + curr_digest: str, + expected_to_fail: set[str], + curr_status: list[tuple[RulePath, StatusCodes]], + all_rules: list[str], +) -> bool: + """Whether one buffer's rules are all verified against its current digest, evaluated over that + buffer's own runs. Overall completion is the AND of this across every run-target buffer.""" + return _is_completion_history( + l=_history_for_buffer(l, buffer), + curr_digest=curr_digest, + expected_to_fail=expected_to_fail, + curr_status=curr_status, + all_rules=all_rules, + ) + + +def completing_run_links( + prover_history: list[ProverHistoryItem], + buffers: Mapping[str, NamedBuffer], + *, + version_history: Sequence[str], +) -> list[str]: + """The prover-run links whose completed results compose the run-target buffers at their current + digests — the same runs :func:`buffer_is_complete` considers. Rule-striping runs one buffer's rules + across several jobs, so its verdicts are spread over all of these links rather than carried by the + last one alone.""" + links: list[str] = [] + seen: set[str] = set() + for b in run_targets(buffers): + digest = buffer_state_digest( + buffers, b.name, version_history=version_history, + ) + for elem in reversed(_history_for_buffer(prover_history, b.name)): + if elem["sort"] != "run": + continue + if elem["state_digest"] != digest: + break # older runs sit at a superseded digest — the same cutoff as _iterate_history + link = elem.get("link") + if link and link not in seen: + seen.add(link) + links.append(link) + return links + + def _merge_prover_history(left: list[ProverHistoryItem], right: list[ProverHistoryItem]) -> list[ProverHistoryItem]: to_ret = left.copy() to_ret.extend(right) @@ -344,7 +428,7 @@ class ProverStateExtra(TypedDict): #: injectors that set no plugin tools. plugin_tools: NotRequired[list[str]] - # The author's working copy of the source under verification; verify_spec runs + # The author's working copy of the source under verification; the prover runs # against its materialization when non-empty (see ProjectDirectory). Absent/empty # outside the editing-enabled pipeline. No merge op intentionally: the vfs is # only ever replaced wholesale (commit_edit / revert_to_edit). @@ -352,11 +436,11 @@ class ProverStateExtra(TypedDict): type ProverEvents = CEXAnalysisStart | CloudPollingEvent | ProverOutputEvent | RuleAnalysisResult | ProverRun | ProverLink | ProverResult -# ``verify_spec`` only runs in the source pipeline, whose state always seeds -# ``version_history`` — permanently empty for an author that never edited, in -# which case it contributes nothing to the digest. The prover's validation stamp is bound to it so a +# The source pipeline's state always seeds ``version_history`` — permanently +# empty for an author that never edited, in which case it contributes nothing to +# the digest. The prover's validation stamp is bound to it so a # post-run edit invalidates the stamp. -class StateWithSkips(CVLGenerationState, ProverStateExtra, VersionedHistory): +class StateWithSkips(CVLGenerationState, ProverStateExtra, VersionedHistory, SpecBuffersExtra): pass class _SpecCallbacks(ProverEventCallbacks): @@ -463,32 +547,6 @@ async def on_analysis_complete(self, rule: RuleResult, explanation: str) -> None await super().on_analysis_complete(rule, explanation) -class VerifySpecSchema(BaseModel): - """ - Run the Certora prover to verify the current spec against the source code. - - Returns verification results: - - VERIFIED: Rule holds for all inputs - - VIOLATED: Counterexample found (with CEX analysis) - - TIMEOUT: Verification did not complete in time - - Use these results to refine your spec. - """ - tool_call_id: Annotated[str, InjectedToolCallId] - - rules: list[str] | None = Field( - default=None, - description="Specific rules to verify. If None, verifies all rules. Mutually exclusive with the `exclude_rules` argument" - ) - - exclude_rules: list[str] | None = Field( - default=None, - description="Specific rules to SKIP verifying. If none validates all rules. Mutually exclusive with `rules` argument" - ) - - state: Annotated[StateWithSkips, InjectedState] - - @contextmanager def tmp_spec( *, @@ -529,19 +587,14 @@ async def provide(vfs: dict[str, str]) -> AsyncIterator[str]: return provide -def materializing_project( - project_root: str, accessor: VFSAccessor[VFSState] -) -> ProjectDirectory: - """The editing strategy: an empty VFS runs in-situ; a non-empty VFS is - materialized over the project into a temporary directory that lives for - the duration of the run. The copy (and the teardown) run in a worker - thread — materializing a whole project is blocking IO that would - otherwise stall every concurrently-streaming batch.""" +def materializing_project(accessor: VFSAccessor[VFSState]) -> ProjectDirectory: + """Materialize the project — the author's VFS overlay unioned over the base source tree — into a + fresh temporary directory that lives for the duration of the run, so every run is the sole tenant + of its own folder and concurrent runs never share on-disk scratch. The copy (and the teardown) run + in a worker thread — materializing a whole project is blocking IO that would otherwise stall every + concurrently-streaming batch.""" @asynccontextmanager async def provide(vfs: dict[str, str]) -> AsyncIterator[str]: - if not vfs: - yield project_root - return stack = ExitStack() tmp = await asyncio.to_thread( stack.enter_context, accessor.materialize({"vfs": vfs}) @@ -574,10 +627,7 @@ def setup_prover_config_in( config, main_contract=main_contract, verify_target=f"{main_contract}:{generated_path}" ) config.update(config_extra) - if rule is not None: - config["rule"] = rule - if exclude_rule is not None: - config["exclude_rule"] = exclude_rule + _apply_selection(config, _selection_of(rule, exclude_rule)) with temp_certora_file( root=working_dir, content=json.dumps(config, indent=2), @@ -588,193 +638,441 @@ def setup_prover_config_in( ) as conf_path: yield (conf_path, config) +def stuck_rule_nag( + status_pairs: list[tuple[RulePath, StatusCodes]], + prover_update: list[ProverHistoryItem], + state: StateWithSkips, +) -> list[str]: + """Warn when a rule has repeated the identical failure across recent runs: append a NagMarker to + ``prover_update`` and return the reminder lines (empty when nothing is stuck). Shared by the + single-spec and per-buffer verify paths.""" + stuck_rules = { + k: v for (k, v) in status_pairs + 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"] == "collect_results" + } + to_warn, seen_post_compaction_history = stuck_rule_warnings( + stuck_rules, state["prover_history"], known_tc_ids + ) + if not to_warn: + return [] + prover_update.append(NagMarker(sort="nag", nagged_rules=list(to_warn))) + return stuck_rule_reminder( + to_warn, + plugin_tools=state.get("plugin_tools") or (), + seen_post_compaction_history=seen_post_compaction_history, + ) + + +@contextmanager +def materialize_buffers( + working_dir: str, buffers: Mapping[str, NamedBuffer] +) -> Iterator[dict[str, str]]: + """Write every buffer as ``{name}.spec`` into the specs dir (all at once, so any buffer's + ``import ".spec"`` resolves to its sibling), and yield ``name -> on-disk spec path``; + every file is removed on exit. The run owns its materialized project folder, so the deterministic + filenames never collide with a concurrent job's.""" + with ExitStack() as stack: + yield { + name: stack.enter_context(tmp_spec(root=working_dir, content=buf.cvl, name=name)) + for name, buf in buffers.items() + } + + +@contextmanager +def buffer_conf( + *, + working_dir: str, + config: dict, + main_contract: str, + spec_path: str, + buffer_name: str, + conf_dir: Path, + msg: str, + selection: RuleSelection | None = None, +) -> Iterator[tuple[str, dict]]: + """Build a conf verifying an already-materialized buffer spec at ``spec_path`` (its imports resolve + to the sibling ``.spec`` files written by :func:`materialize_buffers`). ``selection`` restricts the + run to a subset of the buffer's rules. Yields (conf_path, config).""" + cfg = prover_config_overlay( + config, main_contract=main_contract, verify_target=f"{main_contract}:{spec_path}" + ) + cfg["msg"] = msg + _apply_selection(cfg, selection) + with temp_certora_file( + root=working_dir, + content=json.dumps(cfg, indent=2), + ext="conf", + name=f"verify_{buffer_name}", + prefix="verify", + dest_dir=conf_dir, + ) as conf_path: + yield (conf_path, cfg) + + +class _SubmitBufferArgs(WithInjectedState[StateWithSkips], WithInjectedId): + """ + Submit one run-target buffer for verification as an independent background prover job, and return + immediately — the job proves while you keep working. Submit each buffer as soon as it is ready; + buffers prove in parallel. Re-submitting a buffer relaunches it (superseding any in-flight job for + it), which is how you re-verify a buffer after editing it, or after editing a shared buffer it + imports. A buffer already verified at its current content, or already running, is not re-launched. + Retrieve outcomes with collect_results. + + By default the job runs every rule of the buffer. To keep one expensive rule from holding up the + cheap ones, submit a subset with `rule` (or run the rest with `exclude_rules`): the subsets share + the buffer's one compiled spec, prove as separate parallel jobs, and their results combine — the + buffer is verified once every rule has been covered by some run at the current content. Use this to + isolate a rule by *cost*; use separate buffers to isolate rules by *precision* (differing summary + needs). + """ + name: str = Field(description="The run-target buffer to submit for verification.") + rule: list[str] | None = Field( + default=None, + description="Run only these rules of the buffer (a subset of its own rules). Mutually exclusive " + "with `exclude_rules`; omit both to run the whole buffer.", + ) + exclude_rules: list[str] | None = Field( + default=None, + description="Run every rule of the buffer except these. Mutually exclusive with `rule`.", + ) + + +class _CollectResultsArgs(WithInjectedState[StateWithSkips], WithInjectedId): + """ + Retrieve the results of finished buffer jobs (submitted with submit_buffer). Returns each finished + buffer's prover outcome plus a status board: which buffers are complete, still running, or need + (re)submission. By default it does NOT block — it returns whatever has finished so far (possibly + nothing), so you can go author or submit other buffers instead of waiting. Pass wait=true ONLY when + you have no other work: every buffer submitted and running, with no finished result left to process; + it then sleeps until the next job finishes. + """ + wait: bool = Field( + default=False, + description="Block until the next job finishes. Set true ONLY when you have no other work: " + "every buffer is submitted and running and you have no finished result left to process. " + "Leave false to take whatever has finished so far without waiting.", + ) + + +@dataclass +class ProverToolset: + """The prover-side agent tools. ``make_buffer_tools`` mints a fresh ``[submit_buffer, + collect_results]`` pair that submit per-buffer jobs asynchronously and consume results as they + finish; each pair owns its in-flight job state (queue, job table, submit counts, reported + dupes).""" + + make_buffer_tools: Callable[[], list[BaseTool]] + + +@dataclass +class _BufJob: + """One in-flight (or just-finished) per-buffer prover job. At most one per buffer name at a time; + re-submitting a buffer supersedes (cancels) a stale predecessor. Lives in the prover tool's closure, + not in graph state — asyncio tasks span agent turns and are not serializable.""" + + name: str + #: The buffer's content digest at submit time; a completion is credited only at the current digest, + #: so a job whose digest is now stale (its buffer or a shared import changed) can't mark it done. + digest: str + task: asyncio.Task[None] + #: The rule subset this job runs, or None for the whole buffer. Jobs of one buffer are keyed by + #: ``(name, _selection_key(selection))``, so striped runs at the same content coexist. + selection: RuleSelection | None = None + + +@dataclass +class _BufDone: + """A finished buffer job's payload, delivered through the completion queue to ``collect_results``.""" + + name: str + digest: str + #: The prover report, or a compile/toolchain error message (str) that aborts only this buffer. + result: ProverReport | str + all_rules: list[str] + #: The rule subset this run covered, recorded onto the run's ``ProverRunLog.rules``. + selection: RuleSelection | None = None + + def get_prover_tool( llm: LLM, main_contract: str, project_directory: ProjectDirectory, prover_opts: ProverOptions, analysis_store: CexAnalysisStore | None = None, -) -> BaseTool: +) -> ProverToolset: sem = _prover_sem(prover_opts.cloud) stamper = make_validation_stamper(VALIDATION_KEY) - # Serialize verify calls targeting the same spec name: the spec/conf are written - # under a deterministic name and unlinked on exit, so two overlapping same-stem - # calls (e.g. parallel verify_spec for one component) would race. Distinct stems - # stay concurrent (notably on cloud, where ``sem`` is a no-op). - # Not pruned: bounded by this run's stems (per-component + invariants) and dies with - # the per-run tool; popping a held lock would let a later same-stem call mint a fresh, - # non-excluding one. - spec_locks: dict[str, asyncio.Lock] = {} - - @tool_display("Running prover", None) - @tool(args_schema=VerifySpecSchema) - async def verify_spec( - tool_call_id: Annotated[str, InjectedToolCallId], - state: Annotated[StateWithSkips, InjectedState], - rules: list[str] | None = None, - exclude_rules: list[str] | None = None - ) -> str | Command: - last_msg = state["messages"][-1] - if isinstance(last_msg, AIMessage) and any( - i["id"] != tool_call_id for i in last_msg.tool_calls - ): - return "Cannot call the verify_spec tool in parallel with other tool calls. verify_spec must be the only tool you call in a turn" - - if rules is not None and exclude_rules is not None: - return "Cannot invoke the prover with both `rules` and `exclude_rules` set to non-none" - - spec = state["curr_spec"] - if spec is None: - return "Specification not yet put on VFS" - - spec_hash = string_hash( - spec - ) - - if (last_run := last_prover_run(state["prover_history"])) is not None: - if any(i == "TIMEOUT" for (_,i) in last_run["prover_results"]) and last_run["spec_digest"] == spec_hash: - return "Refusing to re-run prover on identical spec with a known TIMEOUT result; timeouts are not transient " \ - "errors and will not go away by re-running the tool." - - conf = state["config"] - # With a seeded stem, name the spec/conf after it (so on-disk names match the - # dump) under a lock; else fall back to unique uid names (no lock needed). - spec_stem = state.get("spec_stem") - summary = get_run_summary() - component = (spec_stem or main_contract).removeprefix("autospec_") - iteration = len(state["prover_history"]) + 1 - - conf_dir = (CERTORA_DIR / "confs") if spec_stem is not None else CERTORA_DIR - lock = spec_locks.setdefault(spec_stem, asyncio.Lock()) if spec_stem is not None else nullcontext() - prover_msg = f"{component} iteration number {iteration}" - - summary = get_run_summary() - - component = (spec_stem or main_contract).removeprefix("autospec_") - iteration = len(state["prover_history"]) + 1 - prover_msg = f"{component} iteration number {iteration}" - - - async def run_in(run_root: str) -> str | Command: - with setup_prover_config_in( - working_dir=run_root, - main_contract=main_contract, - spec_stem=spec_stem, - spec_contents=spec, - conf_dir=conf_dir, - config=conf, - rule=None, - exclude_rule=None, - msg="" - ) as (config_path, _ignored): - try: - all_rules = await declared_rules_list( - folder=Path(run_root), - args=[config_path] - ) - except SpecCompilationError as exc: - return f"The spec failed to compile:\n{exc.output}" - with setup_prover_config_in( - working_dir=run_root, - main_contract=main_contract, - spec_stem=spec_stem, - spec_contents=spec, - conf_dir=conf_dir, - config=conf, - rule=rules, - exclude_rule=exclude_rules, - msg=prover_msg - ) as (config_path, config): - async with sem: - result = await run_prover( - Path(run_root), - [config_path], - tool_call_id, - prover_opts, - _SpecCallbacks(get_stream_writer(), tool_call_id, summary, config, - analysis_store=analysis_store), - DefaultCexHandler(llm, state, summarization_threshold=10) - ) - - 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 + def component_of(state: StateWithSkips) -> str: + """The label prefix for this generation's prover runs: its seeded spec stem, or the main + contract, with the ``autospec_`` prefix stripped.""" + return (state.get("spec_stem") or main_contract).removeprefix("autospec_") + + # ---- Multi-buffer async submit / collect ------------------------------------------------- + # The agent submits each run-target buffer as an independent background job and consumes results + # as they finish, so a fast group is reviewed while a slow group is still proving. A shared-buffer + # edit re-verifying every importer rides the content digest. + + def make_buffer_tools() -> list[BaseTool]: + """Mint a fresh ``[submit_buffer, collect_results]`` pair with its own in-flight job state: + a new queue, job table, submit counts, and reported-dupe set per call, so separate pairs + never drain or suppress each other's jobs. The prover semaphore and the other deps stay + shared from the enclosing scope.""" + # Multi-buffer async job state, held in the closure (not graph state) so it spans agent turns: + # submit_buffer launches a background task per buffer and returns immediately; collect_results + # drains finished jobs off the queue. At most one live job per buffer name — a re-submit supersedes + # a stale predecessor. See submit_buffer / collect_results below. + # Keyed by (buffer name, selection key): one buffer can have several concurrent jobs, one per + # rule subset it was striped into. A content edit supersedes every one of them (digest changes). + buffer_jobs: dict[tuple[str, str], _BufJob] = {} + done_queue: asyncio.Queue[_BufDone] = asyncio.Queue() + submit_counts: dict[str, int] = {} + + async def _run_buffer_job( + *, name: str, digest: str, label: str, buffers: Mapping[str, NamedBuffer], + vfs: dict[str, str], conf: dict, cex_state: StateWithSkips, tool_call_id: str, + writer: Callable[[ProverEvents], None], summary: RunSummary, + selection: RuleSelection | None = None, + ) -> None: + """Verify one buffer end-to-end against a frozen snapshot (taken at submit time) of the source + and all buffers, then push the outcome onto the completion queue. The job runs in its own + materialized project folder, so editing + re-submitting a shared buffer (or a concurrent + sibling job) can never mutate the files this job is reading. Cancellation (a supersede) propagates + as CancelledError and pushes nothing — the superseded result is simply dropped.""" + conf_dir = CERTORA_DIR / "confs" + try: + async with sem, project_directory(vfs) as run_root: + with materialize_buffers(run_root, buffers) as paths: + spec_path = paths[name] + with buffer_conf( + working_dir=run_root, config=conf, main_contract=main_contract, + spec_path=spec_path, buffer_name=name, conf_dir=conf_dir, msg="", + ) as (cpath, _cfg): + try: + all_rules = await declared_rules_list(folder=Path(run_root), args=[cpath]) + except SpecCompilationError as exc: + await done_queue.put(_BufDone( + name, digest, f"[buffer {name}] failed to compile:\n{exc.output}", [], + )) + return + with buffer_conf( + working_dir=run_root, config=conf, main_contract=main_contract, + spec_path=spec_path, buffer_name=name, conf_dir=conf_dir, msg=label, + selection=selection, + ) as (cpath, cfg): + res = await run_prover( + Path(run_root), [cpath], tool_call_id, prover_opts, + _SpecCallbacks(writer, tool_call_id, summary, cfg, analysis_store=analysis_store), + DefaultCexHandler(llm, cex_state, summarization_threshold=10), + ) + await done_queue.put(_BufDone(name, digest, res, all_rules, selection)) + except asyncio.CancelledError: + raise + except Exception as exc: # a job crash must not sink silently — surface it on the queue + _logger.exception("buffer job %s crashed", name) + await done_queue.put(_BufDone(name, digest, f"[buffer {name}] job error: {exc}", [], selection)) + + def _cur_digest(state: StateWithSkips, buffers: Mapping[str, NamedBuffer], name: str) -> str: + return buffer_state_digest( + buffers, name, version_history=state["version_history"], ) - curr_state_digest = spec_digest( - spec, state["skipped"], state["version_history"] + def _buffer_complete_at( + state: StateWithSkips, buffers: Mapping[str, NamedBuffer], name: str, digest: str, + *, extra_history: Sequence[ProverHistoryItem] = (), + ) -> bool: + return buffer_is_complete( + list(state["prover_history"]) + list(extra_history), buffer=name, curr_digest=digest, + expected_to_fail=set(state["rule_skips"].keys()), curr_status=[], + all_rules=list(buffers[name].owned_rules), ) - 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, - expected_to_fail=set(state["rule_skips"].keys()), - curr_status=prover_results, - all_rules=all_rules + @tool_display("Submitting buffer", None) + @tool(args_schema=_SubmitBufferArgs) + async def submit_buffer(**args) -> str | Command: + state: StateWithSkips = args["state"] + name: str = args["name"] + tool_call_id: str = args["tool_call_id"] + buffers = state.get("buffers") or {} + b = buffers.get(name) + if b is None: + return f"No buffer named {name!r}. Create it with put_buffer first." + if not b.is_run_target: + return f"Buffer {name!r} is a shared (imports-only) buffer; it runs no rules of its own." + + rule: list[str] | None = args.get("rule") + exclude_rules: list[str] | None = args.get("exclude_rules") + if rule is not None and exclude_rules is not None: + return "Pass at most one of `rule` / `exclude_rules`; omit both to run the whole buffer." + selection = _selection_of(rule, exclude_rules) + if selection is not None: + owned = b.owned_rules + unknown = [r for r in selection["selector"] if r not in owned] + if unknown: + return f"Buffer {name!r} declares no rule(s) {unknown}; its rules are {sorted(owned)}." + would_run = selection["selector"] if selection["sort"] == "include" \ + else [r for r in owned if r not in set(selection["selector"])] + if not would_run: + return f"That selection would run no rule of buffer {name!r}; its rules are {sorted(owned)}." + + digest = _cur_digest(state, buffers, name) + if _buffer_complete_at(state, buffers, name, digest): + return f"Buffer {name!r} is already verified at its current content; nothing to submit." + + sel_key = _selection_key(selection) + existing = buffer_jobs.get((name, sel_key)) + if existing is not None and existing.digest == digest: + # This exact subset at this exact content is already in flight, or has just finished with + # its result not yet collected. Either way, do not launch a duplicate — the answer is + # (coming) on the queue; the agent should collect it, not re-run identical work. + proving = "is still proving" if not existing.task.done() else "has already finished" + return ( + f"Buffer {name!r} was already submitted at its current content and {proving}; do not " + f"re-submit it. Call collect_results to take its result — if this is your only " + f"remaining buffer/task and you are just waiting on it, use collect_results(wait=true)." + ) + # A content edit supersedes every subset job of this buffer (all now at a stale digest); the + # sibling subsets at the *current* digest are the parallel stripes and stay running. + for (nm, sk), j in list(buffer_jobs.items()): + if nm == name and j.digest != digest and not j.task.done(): + # TODO: this cancels the local task only; the cloud prover job itself keeps running. + j.task.cancel() + buffer_jobs.pop((nm, sk), None) + + n = submit_counts.get(name, 0) + 1 + submit_counts[name] = n + component = component_of(state) + task = asyncio.create_task(_run_buffer_job( + name=name, digest=digest, selection=selection, + label=f"{component}/{name} submission {n}", + buffers=dict(buffers), vfs=dict(state.get("vfs") or {}), conf=state["config"], + cex_state=state, tool_call_id=tool_call_id, + # The stream writer is captured here and used by the detached task: its prover-progress + # events carry this submit call's tool_call_id, which has already returned, so background-job + # progress can render loosely in the UI (functional results are unaffected). + writer=get_stream_writer(), summary=get_run_summary(), + )) + buffer_jobs[(name, sel_key)] = _BufJob(name=name, digest=digest, task=task, selection=selection) + running = sorted({nm for (nm, _sk), j in buffer_jobs.items() if not j.task.done()}) + sel_desc = "" if selection is None else ( + f" (rules {selection['selector']})" if selection["sort"] == "include" + else f" (excluding {selection['selector']})") + return ( + f"Submitted buffer {name!r}{sel_desc} (submission {n}); it is now proving in the " + f"background. Running: {running}. Call collect_results to retrieve results as jobs finish." ) - prover_update : list[ProverHistoryItem] = [ - ProverRunLog( - tool_call_id=tool_call_id, - prover_results=[(k, v) for (k,v) in result.raw_rule_status.items()], - 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, - sort="run", - declared_rules=all_rules, - state_digest=curr_state_digest, - link=result.link - ) - ] - nag_channel = { + @tool_display("Collecting prover results", None) + @tool(args_schema=_CollectResultsArgs) + async def collect_results(**args) -> str | Command: + state: StateWithSkips = args["state"] + tool_call_id: str = args["tool_call_id"] + wait: bool = args["wait"] + buffers = state.get("buffers") or {} + targets = run_targets(buffers) + if not targets: + return "No run-target buffers to collect. Author buffers and submit_buffer them first." + + # Current digest per buffer is stable within this call (buffers/skips/edit-history are fixed); + # memoize it — each is a content hash over the import closure, read at several points below. + _digests: dict[str, str] = {} + def cur_digest(nm: str) -> str: + if nm not in _digests: + _digests[nm] = _cur_digest(state, buffers, nm) + return _digests[nm] + + drained: list[_BufDone] = [] + while not done_queue.empty(): + drained.append(done_queue.get_nowait()) + if not drained and wait and any(not j.task.done() for j in buffer_jobs.values()): + # Idle wait: nothing else to do, sleep until one job finishes. Unbounded, but every job is + # guaranteed to land on the queue — run_prover self-bounds its subprocess, and a crash is + # caught and enqueued as an error — so this can't hang on a wedged job. + drained.append(await done_queue.get()) + while not done_queue.empty(): + drained.append(done_queue.get_nowait()) + + # Cancel jobs left running against a now-stale digest: a shared buffer they import was edited, so + # their result would be discarded anyway — and on local runs a doomed job needlessly holds the + # single prover slot. The agent re-submits them (they show under needs-(re)submission below). + for key, j in list(buffer_jobs.items()): + if not j.task.done() and j.name in buffers and j.digest != cur_digest(j.name): + j.task.cancel() + buffer_jobs.pop(key, None) + + # Retire finished jobs from the registry. A result that lands between the drain and here stays on + # the queue, so its buffer is picked up on the next collect even though its job is already gone. + for key in [key for key, j in buffer_jobs.items() if j.task.done()]: + buffer_jobs.pop(key, None) + + prover_update: list[ProverHistoryItem] = [] + fresh: dict[str, list[tuple[RulePath, StatusCodes]]] = {} + link: str | None = None + parts: list[str] = [] + for d in drained: + if isinstance(d.result, str): # compile/toolchain error: surface it, record no run + parts.append(f"=== buffer {d.name} ===\n{d.result}") + continue + results: list[tuple[RulePath, StatusCodes]] = list(d.result.raw_rule_status.items()) + fresh.setdefault(d.name, []).extend(results) # striped subsets of one buffer accumulate + link = d.result.link or link + stale = d.name in buffers and d.digest != cur_digest(d.name) + note = (" (NOTE: the spec changed since this was submitted — this result is STALE; re-submit " + "this buffer.)") if stale else "" + parts.append(f"=== buffer {d.name} ==={note}\n{d.result.result_str}") + prover_update.append(ProverRunLog( + tool_call_id=tool_call_id, prover_results=results, rules=d.selection, + spec_digest=string_hash(buffers[d.name].cvl) if d.name in buffers else "", + sort="run", declared_rules=d.all_rules, state_digest=d.digest, buffer=d.name, + link=d.result.link, + )) + # Per-buffer completion is re-evaluated over history + this drain against the CURRENT digest, so + # a stale run (state_digest mismatch) never credits completion. Overall completion is the AND of + # these, checked at publish (check_buffer_completion). + prover_stamps: dict[str, str] = {} + for b in targets: + d = cur_digest(b.name) + if _buffer_complete_at(state, buffers, b.name, d, extra_history=prover_update): + prover_stamps[f"prover:{b.name}"] = d + + # Status board — the agent's work-list. `running` counts only a live job at the CURRENT digest; + # a job left running at a stale digest (its shared import changed) is doomed, so its buffer falls + # under needs-(re)submission until the agent relaunches it. + complete = {b.name for b in targets if f"prover:{b.name}" in prover_stamps} + running = { + j.name for j in buffer_jobs.values() + if not j.task.done() and j.name in buffers and j.digest == cur_digest(j.name) } - if len(to_warn) > 0: - prover_update.append(NagMarker( - sort="nag", - nagged_rules=list(to_warn) - )) - nag_channel["reminders_channel"] = stuck_rule_reminder( - to_warn, - plugin_tools=state.get("plugin_tools") or (), - seen_post_compaction_history=seen_post_compaction_history, - ) - if all_verified: + needs_submit = [b.name for b in targets if b.name not in complete and b.name not in running] + board = [ + "", + f"[buffers] complete: {sorted(complete)}", + f"[buffers] running: {sorted(running)}", + f"[buffers] needs (re)submission: {sorted(needs_submit)}", + ] + if not drained: + parts.append("No finished jobs yet." if running else "No finished jobs and nothing running.") + + nag_channel: dict = {} + all_status = [pair for results in fresh.values() for pair in results] + if reminders := stuck_rule_nag(all_status, prover_update, state): + nag_channel["reminders_channel"] = reminders + if not needs_submit and not running: 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 + "Every run-target buffer is verified at its current content. Once each also has " + "feedback, you can publish." ) + return tool_state_update( - tool_call_id=tool_call_id, content=result.result_str, prover_link=result.link, - prover_history=prover_update, **nag_channel + tool_call_id=tool_call_id, content="\n".join(parts + board), prover_link=link, + validations=prover_stamps, prover_history=prover_update, **nag_channel, ) - # The author's working copy decides where this run executes (in-situ for - # an empty VFS, a temp materialization otherwise); the same-stem lock - # guards the deterministic spec/conf names within it. - async with lock, project_directory(state.get("vfs") or {}) as run_root: - return await run_in(run_root) + return [submit_buffer, collect_results] - return verify_spec + return ProverToolset(make_buffer_tools=make_buffer_tools) diff --git a/composer/spec/source/report/collect.py b/composer/spec/source/report/collect.py index 1cb5ef0a..e727fa2b 100644 --- a/composer/spec/source/report/collect.py +++ b/composer/spec/source/report/collect.py @@ -81,14 +81,14 @@ class Verdict: #: a BAD, error text for an ERROR). Provenance/diagnostics only; ``None`` when the backend #: gives no detail (the prover/foundry fetchers don't). message: str | None = None - #: The run this verdict came from, when the backend runs the units of one component across - #: several runs and the component-level link would misattribute them. ``None`` leaves the - #: unit pointing at the component's run. + #: The run link this verdict came from — with striping a unit's verdicts are spread across + #: several runs, so the winning outcome's own link is kept (see `merge`). ``None`` when the + #: fetcher tracks no per-run link (foundry); the caller falls back to the component run link. link: str | None = None def merge(self, other: "Verdict | None") -> "Verdict": """Combine two results for one unit within a run: higher-priority outcome wins, - line/duration/unit_file/message/link kept from whichever side has them.""" + line/duration/unit_file/message/link kept from whichever side has them (link from the winner).""" if other is None: return self hi, lo = ( @@ -250,8 +250,7 @@ def _ref(unit_name: RuleName) -> RuleRef: if key not in rules_by_key: rules_by_key[key] = RuleVerdict( name=unit_name, spec_file=key[0], outcome=v.outcome, line=v.line, - duration_seconds=v.duration_seconds, prover_link=v.link or run_link, - message=v.message, + duration_seconds=v.duration_seconds, prover_link=v.link or run_link, message=v.message, ) # A referenced unit with no verdict still needs an (UNKNOWN) entry to render. diff --git a/composer/spec/source/report_prover.py b/composer/spec/source/report_prover.py index 9308008e..5a166bad 100644 --- a/composer/spec/source/report_prover.py +++ b/composer/spec/source/report_prover.py @@ -11,7 +11,7 @@ from prover_output_utility import ProverOutputAPI from prover_output_utility.models import CheckResult, NodeStatus -from composer.spec.cvl_generation import GeneratedCVL +from composer.spec.cvl_generation import GeneratedCVL, _output_link from composer.spec.source.report.collect import Formalized, Verdict, VerdictFetcher from composer.spec.source.report.schema import Outcome, RuleName @@ -30,7 +30,12 @@ def _fetch(api: ProverOutputAPI, link: str) -> dict[RuleName, Verdict]: - """rule_name -> rolled-up `Verdict` for one prover run. Best-effort: any POU failure -> {}.""" + """rule_name -> rolled-up `Verdict` for one prover run. Best-effort: any POU failure -> {}. + + ``run_links`` holds raw ``/jobStatus/`` job URLs; POU (and the report's own links) want the + ``/output/`` view, so normalize before the call and stamp the normalized link onto the verdict. + """ + link = _output_link(link) or link try: checks: list[CheckResult] = api.get_all_checks(link) except Exception: @@ -57,9 +62,9 @@ def _fetch_covering( """rule_name -> `Verdict` across the runs that account for one spec. ``newest_first`` must be ordered newest run first, which is the order - `composer.spec.source.prover.covering_run_links` returns: it walks the author's history - backwards. The whole result rests on that order, because the first verdict found for a rule - is the one kept. + `composer.spec.source.prover.completing_run_links` returns (``GeneratedCVL.run_links``): it + walks each buffer's history backwards. The whole result rests on that order, because the first + verdict found for a rule is the one kept. Newest run wins per rule, not the most terminal outcome: a rule that timed out in one run and verified in a later scoped re-run is verified. Within a single run the rollup stays @@ -76,13 +81,18 @@ def _fetch_covering( def make_prover_fetcher(api: ProverOutputAPI | None = None) -> VerdictFetcher[GeneratedCVL]: """A `VerdictFetcher` that pulls per-rule verdicts from ProverOutputUtility, reading every - run that accounts for the component's published spec. POU calls run off the event loop (one - blocking call per run). Only ever invoked for delivered results (collect skips gave-up / + prover run that composes the component's buffers (``GeneratedCVL.run_links``). With buffers + + rule-striping one component's rules are run across several jobs, so a fetch keyed on a single link + would report every rule whose verdict came from another run as UNKNOWN; ``_fetch_covering`` unions + them newest-run-first, so a rule re-proved in a later scoped run wins over its earlier verdict. + ``run_links`` (per-buffer, newest-first) is the buffer-aware source — master's flat + ``covering_links`` does not match the per-buffer run digests. POU calls run off the event loop + (one blocking call per run). Only ever invoked for delivered results (collect skips gave-up / curtailed inputs).""" api = api or ProverOutputAPI() async def fetch(formalized: Formalized[GeneratedCVL]) -> dict[RuleName, Verdict]: - newest_first = formalized.result.covering_output_links + newest_first = formalized.result.run_links if not newest_first: return {} return await asyncio.to_thread(_fetch_covering, api, newest_first) diff --git a/composer/spec/source/spec_buffers.py b/composer/spec/source/spec_buffers.py new file mode 100644 index 00000000..9c3c80b5 --- /dev/null +++ b/composer/spec/source/spec_buffers.py @@ -0,0 +1,456 @@ +"""Multi-buffer CVL specs: the agent authors several independent spec buffers that share +infrastructure through CVL ``import``. + +A named *set* of buffers. A *run-target* buffer is a self-contained spec — its own rules, its +``methods{}`` block, and ``import`` statements pulling in shared buffers — verified and reviewed on +its own. A *shared* buffer holds common ghosts, invariants, and models that run-target buffers +import; it runs no rules itself. Every non-skipped property is owned by exactly one run-target +buffer, and every rule lives in exactly one buffer. + +Each buffer has a content *digest* over its own text plus its transitive import closure. Editing a +shared buffer therefore changes the digest of every buffer that imports it, which is what lets the +pipeline skip re-verifying / re-reviewing an unchanged buffer while correctly invalidating its +importers. + +``NamedBuffer`` is the value object: one buffer's text plus its metadata. +""" + +import os +import posixpath +import re +from collections.abc import Callable, Mapping, Sequence +from pathlib import PurePosixPath +from typing import Annotated + +from pydantic import BaseModel, Field, model_validator +from typing_extensions import TypedDict + +from certora_autosetup.cache.content_cache import hash_content_parts, hash_text +from certora_autosetup.parsers.spec_imports import imports_in_cvl +from certora_autosetup.setup.summary_resolver import extract_cvl_ast +from composer.spec.cvl_generation import FEEDBACK_VALIDATION_KEY +from composer.spec.gen_types import SPECS_DIR + + +MAX_SPEC_BUFFERS_ENV = "AUTOPROVER_MAX_SPEC_BUFFERS" +DEFAULT_MAX_SPEC_BUFFERS = 6 + + +def max_spec_buffers() -> int: + """The most run-target buffers the agent may create. Multi-buffer authoring is the only mode; this + caps how far the agent partitions. A cap of 1 is effectively the single-spec case (one run-target + buffer). Overridable via ``AUTOPROVER_MAX_SPEC_BUFFERS``; a non-integer or <1 value falls back to + the default.""" + raw = os.environ.get(MAX_SPEC_BUFFERS_ENV) + if raw is None: + return DEFAULT_MAX_SPEC_BUFFERS + try: + n = int(raw.strip()) + except ValueError: + return DEFAULT_MAX_SPEC_BUFFERS + return n if n >= 1 else DEFAULT_MAX_SPEC_BUFFERS + + +class NamedBuffer(BaseModel): + """One named CVL spec buffer the agent authors. A frozen pydantic model, so the same type is both + what the buffer logic operates on and what is stored (serializably) in graph state.""" + + model_config = {"frozen": True} + + #: Stable identifier, and the key this buffer is stored under: ``buffers[nm].name == nm`` holds. + name: str + #: The buffer's own CVL text — its rules, its ``methods{}`` block, and its ``import`` statements. + cvl: str + #: Project-relative path of the ``.spec`` this buffer occupies — its identity for import + #: resolution: an ``import ""`` in another buffer that resolves to this path depends on + #: this buffer (see :func:`buffer_imports`). The agent never sets this — it is not a ``put_buffer`` + #: argument — and for now it is derived from ``name`` as ``SPECS_DIR/.spec`` (see the + #: validator). It is a field rather than a computed property only as the hook for later letting the + #: *system* place a buffer elsewhere — notably overlaying an existing autosetup ``.spec`` so the + #: agent can edit it — without reshaping the type; resolution already keys on it. Lifting that stays + #: system-driven: it does not expose ``path`` to the agent. + path: str = "" + #: For a run-target buffer, its property -> rule mapping: each property title it verifies -> the + #: rule/invariant names in ``cvl`` that verify it. Empty for a shared (imported-only) buffer. + property_rules: dict[str, list[str]] = Field(default_factory=dict) + #: False for a shared buffer that only supplies imports and runs no rules of its own. + is_run_target: bool = True + + @model_validator(mode="before") + @classmethod + def _derive_path_from_name(cls, data): + # The agent never supplies a path (it is not a put_buffer argument), so for now every buffer's + # location is just derived from its name: SPECS_DIR/.spec. To later let the SYSTEM place a + # buffer elsewhere (e.g. overlay an existing autosetup .spec), fill this only when a path is + # absent instead of deriving unconditionally — still not agent-controlled. + if isinstance(data, dict) and data.get("name"): + data = {**data, "path": (SPECS_DIR / f"{data['name']}.spec").as_posix()} + return data + + @property + def properties(self) -> frozenset[str]: + """The property titles this buffer covers (the keys of ``property_rules``).""" + return frozenset(self.property_rules) + + @property + def owned_rules(self) -> frozenset[str]: + """The rules this buffer verifies (the union of its ``property_rules`` values).""" + return frozenset(r for rs in self.property_rules.values() for r in rs) + + +def merge_buffers( + left: Mapping[str, NamedBuffer], right: Mapping[str, "NamedBuffer | None"] +) -> dict[str, NamedBuffer]: + """State reducer for the buffers map: right-wins per name; a ``None`` value removes that buffer + (so a tool can merge/drop buffers). Granularity is the whole buffer — a tool that edits one + buffer's text passes the updated :class:`NamedBuffer` under its name.""" + out = dict(left) + for name, val in right.items(): + if val is None: + out.pop(name, None) + else: + out[name] = val + return out + + +class SpecBuffersExtra(TypedDict): + """Graph-state slice holding the agent's spec buffers, keyed by name. Empty until the agent + creates buffers; a single run-target buffer is the simple one-spec case.""" + + buffers: Annotated[dict[str, NamedBuffer], merge_buffers] + + +def buffer_imports(buffers: Mapping[str, NamedBuffer], name: str) -> tuple[str, ...]: + """The names of the sibling buffers ``name`` imports: each ``import ""`` in its CVL, + resolved against where ``name`` is written (``buffers[name].path``), and looked up among the paths + the buffers occupy. A resolved path no buffer occupies — e.g. an autosetup summary under + ``specs/summaries/`` — is a resource, not a sibling, and is skipped. No string surgery: just path + resolution + a map lookup, mirroring how the prover resolves a spec's imports on disk.""" + by_path = {posixpath.normpath(b.path): nm for nm, b in buffers.items()} + here = PurePosixPath(buffers[name].path).parent + out: list[str] = [] + for target in imports_in_cvl(buffers[name].cvl): + if (nm := by_path.get(posixpath.normpath(str(here / target)))) is not None: + out.append(nm) + return tuple(out) + + +def import_closure(buffers: Mapping[str, NamedBuffer], name: str) -> list[NamedBuffer]: + """Buffer ``name`` plus every buffer reachable through its imports, transitively — deduped and + returned sorted by name. A shared (``is_run_target=false``) buffer is in a run-target's closure only + if that run-target transitively imports it, so a shared buffer belongs to the closure (and + invalidation set) of exactly the run-targets that use it — that is what lets a subset of groups share + a summary without invalidating the others. An import resolving to no known buffer is skipped (a + dangling/resource import is a coverage concern, not a hashing one), and cycles terminate safely.""" + seen: set[str] = set() + stack = [name] + while stack: + n = stack.pop() + if n in seen or n not in buffers: + continue + seen.add(n) + stack.extend(buffer_imports(buffers, n)) + return [buffers[n] for n in sorted(seen)] + + +def buffer_digest( + buffers: Mapping[str, NamedBuffer], name: str, *, extra_parts: Sequence[str] = () +) -> str: + """A content digest of buffer ``name`` and its transitive import closure, plus any ``extra_parts`` + (e.g. skipped-property or conf-flag markers). Editing the buffer OR any buffer it imports changes + the digest, so it keys the buffer's cached verify/review. Mirrors + :meth:`ContentCache.compute_cache_key` (content-keyed, order-independent).""" + # TODO: a pure-comment edit (e.g. reframing a property's justification docstring) changes this + # digest and forces the prover to re-verify identical logic — a wasted job. A comment-stripped + # normalization would avoid that, but note this digest also keys the *feedback* review, and the + # judge legitimately reads justification comments — so stripping comments here would wrongly skip + # re-review after a comment-only edit. So find a solution to avoid running + # prover just because of comment-only change. + parts = [f"{b.name}:{hash_text(b.cvl)}" for b in import_closure(buffers, name)] + parts += [f"extra:{p}" for p in extra_parts] + return hash_content_parts(parts) + + +def run_targets(buffers: Mapping[str, NamedBuffer]) -> list[NamedBuffer]: + """The run-target buffers (those that verify rules), sorted by name.""" + return [buffers[n] for n in sorted(buffers) if buffers[n].is_run_target] + + +#: Standalone (not per-buffer) validation key for the single skip review. +SKIPS_VALIDATION_KEY = "skips_review" + + +def buffer_state_digest( + buffers: Mapping[str, NamedBuffer], + name: str, + *, + version_history: Sequence[str], + include_claim: bool = False, +) -> str: + """The per-buffer analogue of ``spec_digest``: a buffer's content + import closure bound to the + applied-edit history. Every per-buffer stamp — feedback and prover — and the completion check key + off this, so editing the buffer, anything it imports, or the source invalidates that buffer's stamps. + + Skips are deliberately NOT keyed here. A skip declaration is owned by no buffer (``validate_coverage`` + keeps skipped and buffer-assigned disjoint), and skip quality is reviewed once against the whole spec + (:func:`skips_review_digest`), so a skip change must not re-verify every buffer. + + With ``include_claim`` the buffer's declared ``property_rules`` also key the digest, so re-assigning + a claim re-triggers review. The feedback stamp sets it (the judge reviews a buffer against the + properties it claims); the prover stamp leaves it False (a claim change does not affect what was + verified).""" + extra = [f"edit:{e}" for e in version_history] + if include_claim: + claim = ";".join( + f"{t}={','.join(rs)}" for t, rs in sorted(buffers[name].property_rules.items()) + ) + extra.append(f"claim:{claim}") + return buffer_digest(buffers, name, extra_parts=extra) + + +def skips_review_digest( + buffers: Mapping[str, NamedBuffer], + *, + skipped: Sequence[tuple[str, str]], + version_history: Sequence[str], +) -> str: + """Digest keying the single skip-review stamp: the skip declarations, every buffer's text (the judge + checks a skip's justification against the code), and the edit history. Independent of any one buffer, + so a skip change re-reviews the skips alone, not every buffer; editing a buffer re-reviews the skips + (a justification can rest on the code) but does not touch other buffers' stamps.""" + parts = [f"buf:{b.name}:{hash_text(b.cvl)}" for b in sorted(buffers.values(), key=lambda b: b.name)] + parts += [f"skip:{t}:{r}" for (t, r) in sorted(skipped)] + parts += [f"edit:{e}" for e in version_history] + return hash_content_parts(parts) + + +def check_buffer_completion( + buffers: Mapping[str, NamedBuffer], + validations: Mapping[str, str], + required_validations: Sequence[str], + *, + skipped: Sequence[tuple[str, str]], + version_history: Sequence[str], +) -> str | None: + """None if every run-target buffer carries each required validation (e.g. ``feedback``, ``prover``) + stamped at its current digest AND (when anything is skipped) the single skip review is current, else + a message naming every buffer/validation missing or stale. The buffers analogue of + ``check_completion``: a per-buffer stamp is keyed ``":"`` and goes stale when that + buffer (or anything it imports, or the edit history) changes; the skip review is a standalone stamp. + + With no run-target buffers this is vacuously satisfied (there is nothing to stamp) — the + all-properties-skipped case, whose validity is decided by ``validate_coverage`` instead.""" + stale: list[str] = [] + for b in run_targets(buffers): + for key in required_validations: + # The feedback stamp tracks a buffer's claimed properties (the judge reviews against them); + # the prover stamp does not. + d = buffer_state_digest( + buffers, b.name, version_history=version_history, + include_claim=(key == FEEDBACK_VALIDATION_KEY), + ) + if validations.get(f"{key}:{b.name}") != d: + stale.append(f"{b.name!r} {key}") + if skipped and validations.get(SKIPS_VALIDATION_KEY) != skips_review_digest( + buffers, skipped=skipped, version_history=version_history + ): + stale.append("skip review") + if stale: + return ( + "Completion REJECTED: re-verify/re-review each of these buffer validations before " + f"publishing: {', '.join(stale)}." + ) + return None + + +def _render_buffers(ordered: Sequence[NamedBuffer], label: Callable[[NamedBuffer], str]) -> str: + """The given buffers concatenated into one document, each under a ``// ===== buffer + (