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/prover/cloud.py b/composer/prover/cloud.py index 33dba95b..1aa865f2 100644 --- a/composer/prover/cloud.py +++ b/composer/prover/cloud.py @@ -7,6 +7,7 @@ import asyncio import logging +import shutil import tempfile from contextlib import asynccontextmanager from dataclasses import dataclass @@ -159,6 +160,42 @@ def _results_api() -> ProverOutputAPI: return ProverOutputAPI(enable_cache=False) +#: A fetch failure here is a *transport* fault, not a bad job: by the time we download documents +#: the job has already polled ``SUCCEEDED``. Re-fetching the same completed job is cheap; re-running +#: the whole proof (what the caller does if this context raises) throws away a finished — often +#: hours-long — verification. So retry the fetch itself a few times with exponential backoff before +#: giving up. POU retries at the individual-request level; this covers a whole-fetch failure that +#: outlives those (e.g. a mid-transfer reset that exhausts the request-level retries on one file). +_FETCH_MAX_ATTEMPTS = 3 +_FETCH_BACKOFF_BASE_S = 2.0 + + +async def _fetch_results(job_id: str, dest: Path) -> None: + """Download a completed job's sources + tree view into ``dest``, retrying a transient failure. + + Each attempt starts from an empty ``dest`` so a partially-written archive from a failed attempt + cannot leave a truncated file behind. Never re-runs the job — only re-reads it. + """ + for attempt in range(1, _FETCH_MAX_ATTEMPTS + 1): + try: + await asyncio.to_thread( + _results_api().fetch_sources_and_treeview_files, job_id, dest + ) + return + except Exception as exc: + if attempt == _FETCH_MAX_ATTEMPTS: + raise + backoff = _FETCH_BACKOFF_BASE_S * 2 ** (attempt - 1) + logger.warning( + "Fetching results for completed job %s failed (attempt %d/%d): %s. " + "Re-fetching the finished job in %.0fs (not re-proving).", + job_id[:8], attempt, _FETCH_MAX_ATTEMPTS, exc, backoff, + ) + for child in dest.iterdir(): + shutil.rmtree(child) if child.is_dir() else child.unlink() + await asyncio.sleep(backoff) + + @asynccontextmanager async def cloud_results( run_result_link: str, @@ -202,7 +239,5 @@ async def on_status(status: str) -> None: # tens of gigabytes on real jobs and used to exhaust the disk; across the # jobs measured here these two subtrees are ~3% of the archive. POU writes # them in the same layout the archive had, so the parse is unchanged. - await asyncio.to_thread( - _results_api().fetch_sources_and_treeview_files, cloud_job.job_id, dest - ) + await _fetch_results(cloud_job.job_id, dest) yield (dest, runtime_ms) diff --git a/composer/scripts/budget_math.py b/composer/scripts/budget_math.py index 26a6bf1c..1287a802 100644 --- a/composer/scripts/budget_math.py +++ b/composer/scripts/budget_math.py @@ -347,7 +347,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 5dba554b..07c22c71 100644 --- a/composer/spec/cvl_generation.py +++ b/composer/spec/cvl_generation.py @@ -269,16 +269,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 6ce13ce2..8d81d476 100644 --- a/composer/spec/source/artifacts.py +++ b/composer/spec/source/artifacts.py @@ -100,7 +100,7 @@ def _write_conf( self, spec: SpecIdentity, 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 1af8fe72..4a31db05 100644 --- a/composer/spec/source/author.py +++ b/composer/spec/source/author.py @@ -5,7 +5,8 @@ import json import pathlib -from dataclasses import dataclass +from abc import abstractmethod +from dataclasses import dataclass, field from langchain_core.tools import BaseTool from pydantic import Field, BaseModel, Discriminator @@ -14,22 +15,30 @@ 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 +from composer.authoring.state import SkippedProperty from composer.authoring.tools import gated_give_up_tool, give_up_tool 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, - GeneratedCVL, PropertyRuleMapping, AppliedEdit, FeedbackToolBase, + cvl_guidance_tools, skip_tools, CVLGenerationExtra, FEEDBACK_VALIDATION_KEY, + CVL_JUDGE_KEY, run_cvl_generator, + GeneratedCVL, PropertyRuleMapping, AppliedEdit, FeedbackToolBase, FeedbackServices, ) 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, run_targets, validate_coverage, + validate_disjoint_rules, +) +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 @@ -86,7 +95,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): @@ -147,25 +156,37 @@ 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: - if (err := check_completion(self.state, self.state["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 with self.tool_deps() as titles: - if (err := validate_property_rules(self.property_rules, self.state["skipped"], titles)) 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}" + 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, ) @@ -632,28 +653,154 @@ 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] + + +class _BufferReviewFeedback(FeedbackToolBase[SourceCVLGenerationState]): + """Feedback base that reviews each run-target buffer independently — against the properties that + buffer claims — with its own persistent per-buffer judge, and stamps ``feedback:``. Both + source feedback tools — editor-aware (source editing) and property-only (structural invariants / + immutable source) — build on it. Subclasses supply ``_review`` (how the per-buffer judge is reached) + and ``_all_properties`` (the batch's property set). With no run-target buffers (nothing authored yet, + or every property skipped) there is nothing to review.""" + + @abstractmethod + async def _review( + self, name: str, spec: str, skipped: list[SkippedProperty], + properties: list[PropertyFormulation], + ) -> PropertyFeedbackProtocol: + """Review ``spec`` with the cached per-buffer judge for ``name``, scored against ``properties`` + (that buffer's claimed subset).""" + ... + + @abstractmethod + def _all_properties(self) -> list[PropertyFormulation]: + """The batch's full property set, for resolving a buffer's claimed subset.""" + ... + + @override + def _version_history(self) -> Sequence[str]: + return self.state["version_history"] + + @override + async def _get_feedback( + self, spec: str, skipped: list[SkippedProperty] + ) -> PropertyFeedbackProtocol: + # Buffer feedback reviews per buffer 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-buffer _review, not _get_feedback") + + @override + 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, a + # skip, 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. + 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, skipped=skipped_pairs, version_history=vh, include_claim=True + ) + + stale = [b for b in targets if validations.get(f"feedback:{b.name}") != digest(b.name)] + if not stale: + return tool_return( + self.tool_call_id, "All buffers already reviewed and approved at their current state." + ) + new_stamps: dict[str, str] = {} + blocks: list[str] = [] + for b in stale: + 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), skipped, 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) + return tool_state_update(self.tool_call_id, "\n\n".join(blocks), validations=new_stamps) + + @tool_display("Getting feedback", "Feedback") class EditorAwareFeedbackTool( - FeedbackToolBase[SourceCVLGenerationState], - WithAsyncDependencies[Command, ContextualFeedbackToolImpl[SourceSnapshot]], + _BufferReviewFeedback, + WithAsyncDependencies[Command, _PerBufferJudge[ContextualFeedbackToolImpl[SourceSnapshot]]], ): __doc__ = FeedbackToolBase.__doc__ @override - async def _get_feedback( - self, spec: str, skipped: list[SkippedProperty] + async def _review( + self, name: str, spec: str, skipped: list[SkippedProperty], + properties: list[PropertyFormulation], ) -> PropertyFeedbackProtocol: - with self.tool_deps() as judge: + 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) + return await judges.for_buffer(name, properties)( + snap, spec, skipped, self.rebuttals, self.tool_call_id + ) @override - def _version_history(self) -> Sequence[str]: - return self.state["version_history"] + def _all_properties(self) -> list[PropertyFormulation]: + with self.tool_deps() as judges: + return judges.properties + + +@tool_display("Getting feedback", "Feedback") +# Buffer-aware feedback for the non-editing author (structural invariants, immutable source): reviews +# each buffer via the property judge (no source snapshot) and stamps ``feedback:``. The +# agent-facing description is the shared FeedbackToolBase one. +class BufferPropertyFeedbackTool( + _BufferReviewFeedback, + WithAsyncDependencies[Command, _PerBufferJudge[FeedbackServices]], +): + __doc__ = FeedbackToolBase.__doc__ + + @override + async def _review( + self, name: str, spec: str, skipped: list[SkippedProperty], + properties: list[PropertyFormulation], + ) -> PropertyFeedbackProtocol: + with self.tool_deps() as judges: + return await judges.for_buffer(name, properties).feedback_thunk( + spec, skipped, self.rebuttals, self.tool_call_id + ) + + @override + def _all_properties(self) -> list[PropertyFormulation]: + with self.tool_deps() as judges: + return judges.properties _PropertyGenTemplate = TypedTemplate[PropertyGenParams]("property_generation_prompt.j2") @@ -663,6 +810,7 @@ class PropertyGenSystemParams(TypedDict): _PropertyGenSysTemplate = TypedTemplate[PropertyGenSystemParams]("property_generation_system_prompt.j2") + #: The prover's tool extension: contributions come from plugins deriving #: ``CertoraProverTools``, dispatched via their ``certora_prover_tools`` hook. _PROVER_TOOLS = InjectingToolExtension( @@ -671,7 +819,8 @@ class PropertyGenSystemParams(TypedDict): @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 @@ -788,17 +937,19 @@ async def batch_cvl_generation( }) sys_prompt : list[RawPromptInput | type[CacheMarker]] = [ - _PropertyGenSysTemplate.bind({"source_editing": editing is not None}).render_to + _PropertyGenSysTemplate.bind({"source_editing": editing is not None}).render_to, + 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] = [] if editing_tools is not None: 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( @@ -823,7 +974,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, @@ -863,16 +1014,28 @@ async def propose( }) protected = focus.protected if focus is not None else () if editing is None: - feedback_suite = property_tools( - property_feedback_judge(judge_ctx, env, judge_prompt, props), - protected=protected, + # 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). + property_judges = _PerBufferJudge( + build=lambda name, claimed: property_feedback_judge( + judge_ctx.child(CacheKey[CVLJudge, CVLJudge](name)), env, judge_prompt, claimed + ), + properties=props, ) + feedback_suite = [ + BufferPropertyFeedbackTool.bind(property_judges).as_tool("feedback_tool"), + *skip_tools(titles, protected=protected), + ] else: - judge_impl = source_feedback_judge( - judge_ctx, _LiveJudgeHost(env, editing), judge_prompt, props + 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), ] @@ -893,12 +1056,23 @@ async def propose( ) else: b = b.with_tools(env.source_tools) + # Multi-buffer authoring (opt-in): the agent partitions the spec into several named buffers, + # 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( 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") @@ -972,7 +1146,8 @@ async def propose( reminders_channel=[], vfs=restored_vfs, version_history=restored_history, - spec_stem=spec_stem + spec_stem=spec_stem, + buffers={}, ) ) except BudgetExceeded as e: @@ -998,8 +1173,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] = [] if editing is not None: for edit_id in res_state["version_history"]: diff --git a/composer/spec/source/buffer_tools.py b/composer/spec/source/buffer_tools.py new file mode 100644 index 00000000..25053d83 --- /dev/null +++ b/composer/spec/source/buffer_tools.py @@ -0,0 +1,188 @@ +"""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.""" + + 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 55a63897..9f0c5dfa 100644 --- a/composer/spec/source/pipeline.py +++ b/composer/spec/source/pipeline.py @@ -44,7 +44,7 @@ from composer.spec.source.summarizer import setup_summaries from composer.spec.source.struct_invariant import get_invariant_formulation from composer.spec.source.autosetup import SetupSuccess -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, @@ -83,8 +83,8 @@ 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) #: The invariant CVL's slot in the report: a real delivery (imported by every component spec) #: or the quarantined leftovers of a budget-curtailed generation (appendix only). @@ -96,7 +96,7 @@ class ProverRunner(Formalizer[GeneratedCVL, ContractComponentInstance]): """Immutable formalizer: per-batch CVL generation against a fixed prover config + resource set (already including ``invariants.spec`` when there are structural invariants), plus the in-memory invariant result for the report.""" - _prover_tool: BaseTool + _prover_tool: ProverToolset _prover_config: dict _resources: list[CVLResource] _invariant: tuple[list[PropertyFormulation], InvariantResult] | None @@ -210,7 +210,7 @@ class ProverPrepared(PreparedSystem[GeneratedCVL, ContractComponentInstance, Con prover-only pre-formalization fan-out in ``prepare_formalization``.""" _sys_desc: SystemDescriptionHarnessed _harnessed: HarnessedApplication - _prover_tool: BaseTool + _prover_tool: ProverToolset _analyzed: SourceApplication _deps: _ProverPipelineDeps @@ -371,7 +371,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 b54eac1d..603b1ce2 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 @@ -63,13 +64,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 4f26d2b6..a1af9def 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,6 +141,9 @@ class ProverRunLog(TypedDict): sort: Literal["run"] declared_rules: list[str] state_digest: str + # 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] class NagMarker(TypedDict): nagged_rules: list[RulePath] @@ -247,6 +277,33 @@ 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 _merge_prover_history(left: list[ProverHistoryItem], right: list[ProverHistoryItem]) -> list[ProverHistoryItem]: to_ret = left.copy() to_ret.extend(right) @@ -264,7 +321,7 @@ class ProverStateExtra(TypedDict): prover_history: Annotated[list[ProverHistoryItem], _merge_prover_history] reminders_channel: 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). @@ -272,12 +329,12 @@ class ProverStateExtra(TypedDict): type ProverEvents = CEXAnalysisStart | CloudPollingEvent | ProverOutputEvent | RuleAnalysisResult | ProverRun | ProverLink | ProverResult -# ``verify_spec`` only runs in the source pipeline, whose state always seeds +# The source pipeline's state always seeds # ``version_history`` — permanently empty in phases without the edit tools # (structural invariants, never-edited authors), 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): @@ -384,32 +441,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( *, @@ -450,19 +481,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}) @@ -495,10 +521,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), @@ -509,197 +532,447 @@ 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))) + reminders = [ + "The following rule(s) have had identical failures on the last 3 runs of the prover:", + *(f"- {it.pprint()}" for it in to_warn), + "You may need to significantly change your approach, or skip the property if this is a persistent issue (you may need to use rebuttals to communicate" + " these failures to the feedback judge).", + ] + if seen_post_compaction_history: + reminders.append( + "(NB: Some of these prover calls happened before your most recent task history summarization)" + ) + return reminders + + +@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: + skipped_pairs = [(str(s.property_title), str(s.reason)) for s in state["skipped"]] + return buffer_state_digest( + buffers, name, skipped=skipped_pairs, 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 - ) - ] - 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, + )) + # 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"] = [ - "The following rule(s) have had identical failures on the last 3 runs of the prover:", - *(f"- {it.pprint()}" for it in to_warn), - "You may need to significantly change your approach, or skip the property if this is a persistent issue (you may need to use rebuttals to communicate" - " these failures to the feedback judge)." - ] - if seen_post_compaction_history: - nag_channel["reminders_channel"].append( - "(NB: Some of these prover calls happened before your most recent task history summarization)" - ) - if all_verified: + 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/spec_buffers.py b/composer/spec/source/spec_buffers.py new file mode 100644 index 00000000..14b9d594 --- /dev/null +++ b/composer/spec/source/spec_buffers.py @@ -0,0 +1,341 @@ +"""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 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] + + +def buffer_state_digest( + buffers: Mapping[str, NamedBuffer], + name: str, + *, + skipped: Sequence[tuple[str, 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 + current authoring state (skip declarations as ``(title, reason)`` pairs, and 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, a skip, or the source invalidates that buffer's stamps. + + 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"skip:{t}:{r}" for (t, r) in sorted(skipped)), + *(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 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, 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 skips/edit history) changes. + + 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, skipped=skipped, 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 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 + (