Improves/Fixes judge prompt; testing for templates

.github/workflows/integration-tests.yml

@@ -51,15 +51,16 @@ jobs:
           sudo chmod +x /usr/local/bin/solc8.29
           solc8.29 --version
 
-      - name: async runners
-        run: |
-          uv pip install pytest-xdist
-
       - name: run autoprover integration tests
         run: |
           uv run pytest -n 3 -m 'expensive' tests
         env:
           CERTORAKEY: ${{ secrets.CERTORAKEY }}
+      - name: run template fuzz tests
+        run: |
+          uv run pytest -n 2 -m 'fuzz' tests
+        env:
+          HYPOTHESIS_PROFILE: extended
 
       # Fires ONLY if a step above failed. The whole point of the file.
       - name: Notify Slack on failure

.github/workflows/pytest.yml

@@ -27,4 +27,4 @@ jobs:
 
       - name: Run pytest
         run: |
-          uv run pytest -m 'not expensive' tests/
+          COMPOSER_STRICT_TEMPLATES=1 uv run pytest -m 'not expensive' tests/

composer/cli/natspec_startup.py

@@ -1,11 +1,11 @@
 import pathlib
-from typing import Any, TypedDict
+from typing import TypedDict
 
 from langchain_core.tools import BaseTool
 
 from graphcore.tools.vfs import DirBackend, FSBackend, Materializer, fs_tools_layered
 
-from composer.spec.gen_types import TypedTemplate
+from composer.spec.gen_types import PartialTemplate
 from composer.spec.system_model import Application, FromSourceApplication
 from composer.spec.natspec.models import (
     InterfaceResult,
@@ -19,9 +19,8 @@
 class PathChoosingConf(TypedDict):
     agent_chooses_path: bool
 
-_InterfaceTemplate = TypedTemplate[PathChoosingConf]("interface_generation_prompt.j2")
-_StubTemplate = TypedTemplate[PathChoosingConf]("stub_generation_prompt.j2")
-
+_InterfaceTemplate = PartialTemplate[PathChoosingConf, InterfaceGenCallParams]("interface_generation_prompt.j2")
+_StubTemplate = PartialTemplate[PathChoosingConf, StubGenCallParams]("stub_generation_prompt.j2")
 
 def build_mental_model(
     *,
@@ -34,10 +33,10 @@ def build_mental_model(
     agent_chooses_path = source_root is not None
     interface_prompt = _InterfaceTemplate.bind(
         {"agent_chooses_path": agent_chooses_path}
-    ).depends(InterfaceGenCallParams)
+    )
     stub_prompt = _StubTemplate.bind(
         {"agent_chooses_path": agent_chooses_path}
-    ).depends(StubGenCallParams)
+    )
 
     if source_root is not None:
         return MentalModel(

composer/foundry/author.py

@@ -47,8 +47,7 @@
 from composer.spec.gen_types import TypedTemplate
 from composer.spec.graph_builder import bind_standard, run_to_completion
 from composer.spec.types import PropertyFormulation
-from composer.spec.system_model import ContractComponentInstance
-from composer.spec.tool_env import BasicAgentTools
+from composer.spec.system_model import ContractComponentInstance, component_context
 from composer.spec.service_host import ServiceHost
 from composer.spec.util import uniq_thread_id
 from composer.tools.thinking import RoughDraftState, get_rough_draft_tools
@@ -611,14 +610,15 @@ def get_resume_prompt(self, state: FoundryGenerationState, summary: str) -> str:
 # Top-level batch entry
 # ---------------------------------------------------------------------------
 
-
+@component_context
 class FoundryPropertyGenParams(TypedDict):
     """Per-batch render variables for ``foundry_property_generation_prompt.j2``.
     Mirror of ``PropertyGenParams`` in the CVL author, minus ``resources``
     (no CVL-importable resource concept here)."""
     context: ContractComponentInstance | None
     properties: list[PropertyFormulation]
     contract_name: str
+    sort: Literal["existing"]
 
 
 _FoundryPropertyGenTemplate = TypedTemplate[FoundryPropertyGenParams](
@@ -668,6 +668,7 @@ async def batch_foundry_test_generation(
         "context": component,
         "properties": props,
         "contract_name": contract_name,
+        "sort": "existing"
     })
 
     titles = [p.title for p in props]

composer/meta/__init__.py

@@ -0,0 +1,3 @@
+"""
+Module for introspection of the composer module. Stay away
+"""

composer/meta/resolver.py

@@ -0,0 +1,13 @@
+from typing import Iterable, get_args
+import sys
+import importlib
+from .types import TemplateDecl
+
+def resolve_params(s: TemplateDecl) -> Iterable[type]:
+    importlib.import_module(s.module)
+    t = sys.modules[s.module].__dict__[s.qualname]
+
+    res = getattr(t, "__orig_class__")
+    for i in get_args(res):
+        assert isinstance(i, type), f"got {i} {type(i)}"
+        yield i

composer/meta/templates.py

@@ -0,0 +1,100 @@
+import ast
+import pathlib
+
+from composer.meta.types import TemplateSort, TemplateDecl
+
+PACKAGE = "composer"
+FULL_TYPE_NAME = "TypedTemplate"
+PARTIAL_NAME = "PartialTemplate"
+TEMPLATE_ARG = 0
+
+TYPE_NAMES = (FULL_TYPE_NAME, PARTIAL_NAME)
+
+def _names_type(node: ast.expr) -> TemplateSort | None:
+    """True if `node` textually refers to TYPE_NAME: bare, attribute, or subscripted."""
+    if isinstance(node, ast.Name):
+        if node.id in TYPE_NAMES:
+            return node.id
+    if isinstance(node, ast.Attribute):
+        if node.attr in TYPE_NAMES:
+            return node.attr
+    if isinstance(node, ast.Subscript):  # TYPE_NAME[...] generic annotation
+        return _names_type(node.value)
+    return None
+
+def _extract_template(value: ast.expr | None) -> str | None:
+    """Pull the template-name argument (positional or kwarg per TEMPLATE_ARG)."""
+    if not isinstance(value, ast.Call):
+        return None
+    arg = value.args[TEMPLATE_ARG] if len(value.args) > TEMPLATE_ARG else None
+    if arg is None:
+        return None
+    if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
+        return arg.value
+    raise ValueError(
+        f"template argument must be a string literal so the static "
+        f"scanner can see it (got {ast.dump(arg)[:60]})"
+    )
+
+def _match_statement(stmt: ast.stmt) -> tuple[str, TemplateSort, ast.expr | None] | None:
+    """Return (variable_name, value_expr_or_None) if `stmt` declares a TYPE_NAME."""
+    if isinstance(stmt, ast.AnnAssign):
+        if (ty_name := _names_type(stmt.annotation)) is not None and isinstance(stmt.target, ast.Name):
+            return stmt.target.id, ty_name, stmt.value
+        return None
+    if isinstance(stmt, ast.Assign):
+        if (isinstance(stmt.value, ast.Call) and (ty_sort := _names_type(stmt.value.func)) is not None
+                and len(stmt.targets) == 1 and isinstance(stmt.targets[0], ast.Name)):
+            return stmt.targets[0].id, ty_sort, stmt.value
+    return None
+
+def iter_declarations(package_dir: pathlib.Path, skips: tuple[pathlib.Path, ...]):
+    """Yield (module_name, var_name, template_or_None) for each top-level declaration.
+
+    Purely textual/AST -- aliased type names (``from registry import
+    TemplateDecl as TD``) or declarations built in loops will NOT be found.
+    Keep declaration style boring.
+    """
+    base = package_dir.parent
+    for py in sorted(package_dir.rglob("*.py")):
+        if any(py.is_relative_to(skip) for skip in skips):
+            continue
+        source = py.read_text(encoding="utf-8")
+        if FULL_TYPE_NAME not in source and PARTIAL_NAME not in source:  # cheap pre-filter before parsing
+            continue
+        tree = ast.parse(source, filename=str(py))
+        module_name = ".".join(py.relative_to(base).with_suffix("").parts)
+
+        if module_name.endswith(".__init__"):
+            module_name = module_name.removesuffix(".__init__")
+        for stmt in tree.body:  # top level ONLY -- no ast.walk on purpose
+            match = _match_statement(stmt)
+            if match is None:
+                continue
+            var_name, ty_sort, value = match
+            try:
+                template = _extract_template(value)
+            except ValueError as exc:
+                raise ValueError(f"{py}:{stmt.lineno}: {exc}") from None
+            yield module_name, var_name, template, ty_sort
+
+def build_manifest(root_package: pathlib.Path, skips: tuple[pathlib.Path, ...]) -> dict[str, TemplateDecl]:
+    """Scan and return {"module:VAR": {"module", "qualname", "template"?}}."""
+    manifest: dict[str, TemplateDecl] = {}
+    templates_seen: dict[str, str] = {}
+    for module_name, var_name, template, ty_sort in iter_declarations(root_package, skips):
+        key = f"{module_name}:{var_name}"
+        if key in manifest:
+            raise ValueError(f"{key} declared twice at module top level")
+        if template is None:
+            raise ValueError(f"No template name extracted for {key}")
+        if template in templates_seen:
+            raise ValueError(
+                f"template {template!r} declared by both "
+                f"{templates_seen[template]} and {key}"
+            )
+        templates_seen[template] = key
+        manifest[key] = TemplateDecl(
+            qualname=var_name, module=module_name, ty_sort=ty_sort, template_name=template
+        )
+    return manifest

composer/meta/types.py

@@ -0,0 +1,12 @@
+from typing import Literal
+from pydantic import BaseModel, TypeAdapter
+
+type TemplateSort = Literal["TypedTemplate", "PartialTemplate"]
+
+class TemplateDecl(BaseModel):
+    module: str
+    qualname: str
+    template_name: str
+    ty_sort: TemplateSort
+
+Manifest = TypeAdapter(dict[str, TemplateDecl])

composer/pipeline/core.py

@@ -55,7 +55,7 @@ class Formalizer[FormT: BackendResult](ABC):
     state — never set post-hoc. `FormT: ReportableResult` is what makes the report a core step."""
     formalized_type: type[FormT]
     backend_tag: ReportBackend
-    
+
     @abstractmethod
     async def formalize(
         self,

composer/scripts/template_manifest.py

@@ -0,0 +1,41 @@
+import importlib.util
+import pathlib
+import sys
+
+from composer.meta.types import TemplateDecl, Manifest
+from composer.meta.templates import build_manifest
+
+# --------------------------------------------------------------------------
+# STUBS -- fill these in for your repo
+# --------------------------------------------------------------------------
+PACKAGE = "composer"
+
+_THIS_DIR = pathlib.Path(__file__).parent
+
+def _package_root() -> pathlib.Path:
+    """Locate PACKAGE on disk without importing it.
+
+    find_spec on a *top-level* name only consults path finders; the module is
+    never executed. (Dotted names would import parents -- don't pass those.)
+    """
+    spec = importlib.util.find_spec(PACKAGE)
+    if spec is None or not spec.submodule_search_locations:
+        raise SystemExit(f"cannot locate package {PACKAGE!r} on sys.path")
+    to_ret = pathlib.Path(next(iter(spec.submodule_search_locations)))
+    assert pathlib.Path(__file__).is_relative_to(to_ret)
+    return to_ret
+
+def _repo_root() -> pathlib.Path:
+    return _package_root().parent
+
+def render_manifest(manifest: dict[str, TemplateDecl]) -> str:
+    """Canonical, diff-stable serialization (sorted keys, trailing newline)."""
+    return Manifest.dump_json(manifest, indent=2).decode("utf-8")
+
+def _main():
+    (_repo_root() / "template_manifest.json").write_text(
+        render_manifest(build_manifest(_package_root(), (_THIS_DIR,)))
+    )
+
+if __name__ == "__main__":
+    sys.exit(_main())

composer/spec/feedback.py

@@ -17,9 +17,9 @@
 from composer.spec.graph_builder import bind_standard, run_to_completion
 from composer.cvl.tools import get_cvl
 from composer.tools.thinking import RoughDraftState, get_rough_draft_tools
-from composer.spec.gen_types import TemplateInstantiation, InjectedTemplate, TypedTemplate
+from composer.spec.gen_types import TemplateInstantiation, TypedTemplate, ITypedTemplate, PartialTemplate
 from composer.spec.cvl_generation import FeedbackToolContext, Rebuttal, SkippedProperty
-from composer.spec.system_model import ContractComponentInstance
+from composer.spec.system_model import ContractComponentInstance, component_context
 from composer.spec.util import uniq_thread_id
 
 class PropertyFeedback(BaseModel):
@@ -32,6 +32,11 @@ class PropertyFeedback(BaseModel):
 class Properties(TypedDict):
     properties: list[PropertyFormulation]
 
+class FeedbackInputs(Properties):
+    rebuttals: Sequence[Rebuttal]
+    skipped: Sequence[SkippedProperty]
+
+@component_context
 class FeedbackInherentParams(TypedDict):
     context: ContractComponentInstance | None
     # Matches the tri-state on the env-level ``sort``:
@@ -42,7 +47,7 @@ class FeedbackInherentParams(TypedDict):
     #                    has real immutable source.
     sort: Sort
 
-FeedbackTemplate = TypedTemplate[FeedbackInherentParams]("property_judge_prompt.j2")
+FeedbackTemplate = PartialTemplate[FeedbackInherentParams, FeedbackInputs]("property_judge_prompt.j2")
 
 class JudgeSystemParams(TypedDict):
     sort: Sort
@@ -56,7 +61,7 @@ class JudgeSystemParams(TypedDict):
 def property_feedback_judge(
     ctx: WorkflowContext[CVLJudge],
     env: ServiceHost,
-    prompt: InjectedTemplate[Properties] | TemplateInstantiation,
+    prompt: ITypedTemplate[FeedbackInputs],
     props: list[PropertyFormulation],
     *,
     extra_inputs: list[str | dict] | Callable[[], list[str | dict]] | None = None,
@@ -88,24 +93,28 @@ def did_rough_draft_read(s: ST, _) -> str | None:
 
     mem = ctx.get_memory_tool()
 
-    final_prompt = prompt if isinstance(prompt, TemplateInstantiation) else prompt.inject({"properties": props})
-
-    workflow = bind_standard(
+    staged_workflow = bind_standard(
         builder, ST, validator=did_rough_draft_read
     ).with_input(
         SpecJudgeInput
-    ).inject(
-        lambda b: final_prompt.render_to(b.with_initial_prompt_template)
     ).inject(
         lambda g: system_prompt.render_to(g.with_sys_prompt_template)
-    ).with_tools([*rough_draft_tools, mem, get_cvl(ST), ]).compile_async()
+    ).with_tools([*rough_draft_tools, mem, get_cvl(ST), ])
 
     async def the_tool(
         cvl: str,
         skipped: Sequence[SkippedProperty],
         rebuttals: Sequence[Rebuttal],
         within_tool: str,
     ) -> PropertyFeedback:
+        workflow = staged_workflow.inject(
+            lambda b: prompt.bind({
+                "properties": props,
+                "rebuttals": rebuttals,
+                "skipped": skipped
+            }).render_to(b.with_initial_prompt_template)
+        ).compile_async()
+
         input_parts: list[str | dict] = []
         if extra_inputs:
             if isinstance(extra_inputs, list):
@@ -115,25 +124,6 @@ async def the_tool(
 
         input_parts.append("The proposed CVL file is")
         input_parts.append(cvl)
-        if skipped:
-            input_parts.append("The following properties were explicitly skipped by the author:")
-            for s in skipped:
-                input_parts.append(f"  Property {s.property_title}: {s.reason}")
-        if rebuttals:
-            input_parts.append(
-                "The author has filed the following rebuttals against feedback from "
-                "prior rounds. Evaluate each per the Step 1 rebuttal rule (and the "
-                "Criteria 7 exception for skip-related rebuttals). Empirical evidence "
-                "types (`typecheck_failure`, `counterexample`, `manual_citation`) "
-                "carry near-binding weight; `reasoned` rebuttals are a conversation, "
-                "not a veto."
-            )
-            for i, r in enumerate(rebuttals, 1):
-                input_parts.append(
-                    f"  Rebuttal {i} [{r.evidence_type}]\n"
-                    f"    Addressing: {r.prior_feedback_reference}\n"
-                    f"    Evidence: {r.evidence}"
-                )
         res = await run_to_completion(
             workflow,
             SpecJudgeInput(input=input_parts, curr_spec=cvl, memory=None, did_read=False),