codegen integration tests

.github/workflows/integration-tests.yml

@@ -43,6 +43,13 @@ jobs:
           uv pip install solc-select
           uv run solc-select install 0.8.29
           uv run solc-select use 0.8.29
+          # certoraRun resolves Certora-convention names like `solc8.29` as
+          # literal executables; solc-select only ships the `solc` shim, so
+          # fetch the static binary under that exact name.
+          sudo curl -fsSL https://github.com/ethereum/solidity/releases/download/v0.8.29/solc-static-linux \
+            -o /usr/local/bin/solc8.29
+          sudo chmod +x /usr/local/bin/solc8.29
+          solc8.29 --version
 
       - name: async runners
         run: |

composer/bind.py

@@ -32,3 +32,11 @@
         os.environ.get("COMPOSER_RECORD_OUT"),
         no_thinking=bool(os.environ.get("COMPOSER_RECORD_NO_THINKING")),
     )
+
+if (_resp := os.environ.get("COMPOSER_RESPONSE_TAPE")):
+    # Scripted human replies for console HITL interrupts (commit approvals,
+    # requirement relaxations). Independent of — and replayed alongside —
+    # COMPOSER_TEST_TAPE; the tape module exposes ``install_response_tape``.
+    import importlib
+    _rmod = importlib.import_module(f"composer.testing.ui_harness_{_resp}")
+    _rmod.install_response_tape()

composer/io/graph_runner.py

@@ -85,6 +85,7 @@ async def run_graph[H, S: StateLike, I: StateLike, C: StateLike | None](
                 curr_input = graph_input
                 graph_input = None
                 interrupted = False
+                interrupt_data: H | None = None
                 async for (ty, payload) in graph.astream(
                     curr_input, config=curr_config, context=ctxt, stream_mode=["checkpoints", "updates", "custom"]
                 ):
@@ -105,35 +106,28 @@ async def run_graph[H, S: StateLike, I: StateLike, C: StateLike | None](
                             assert human_handler is not None
                             if "configurable" in curr_config and "checkpoint_id" in curr_config["configurable"]:
                                 del curr_config["configurable"]["checkpoint_id"]
-                            interrupt_data = cast(H, payload["__interrupt__"][0].value)
-                            curr_state = cast(S, (await graph.aget_state({"configurable": {"thread_id": tid}})).values)
-                            human_response = await human_handler(interrupt_data, curr_state)
-                            graph_input = Command(resume=human_response)
-                            interrupted = True
-                            break
-                        elif ty == "custom":
-                            event_sink(
-                                CustomUpdate(payload, thread_id=tid, checkpoint_id=curr_checkpoint) # pyright: ignore[reportPossiblyUnboundVariable]
-                            )
-                        else:
-                            assert ty == "updates"
-                            if "__interrupt__" in payload:
-                                assert human_handler is not None
-                                if "configurable" in curr_config and "checkpoint_id" in curr_config["configurable"]:
-                                    del curr_config["configurable"]["checkpoint_id"]
+                            # Record the interrupt but keep draining the stream:
+                            # the interrupt checkpoint may not be committed when
+                            # this update is yielded, and the resume below targets
+                            # the thread's latest checkpoint. Breaking out here
+                            # races that write — a resume against the stale
+                            # checkpoint replays the previous node (duplicated LLM
+                            # turns / re-fired interrupts / dropped state updates).
+                            if not interrupted:
                                 interrupt_data = cast(H, payload["__interrupt__"][0].value)
-                                curr_state = cast(S, (await graph.aget_state({"configurable": {"thread_id": tid}})).values)
-                                human_response = await human_handler(interrupt_data, curr_state)
-                                graph_input = Command(resume=human_response)
                                 interrupted = True
-                                break
-                            event_sink(
-                                StateUpdate(
-                                    payload, thread_id=tid
-                                )
+                            continue
+                        event_sink(
+                            StateUpdate(
+                                payload, thread_id=tid
                             )
-                    if interrupted:
-                        continue
+                        )
+                if interrupted:
+                    assert human_handler is not None
+                    curr_state = cast(S, (await graph.aget_state({"configurable": {"thread_id": tid}})).values)
+                    human_response = await human_handler(cast(H, interrupt_data), curr_state)
+                    graph_input = Command(resume=human_response)
+                    continue
 
                 result_state = (await graph.aget_state({"configurable": {"thread_id": tid}})).values
                 return cast(S, result_state)

composer/natreq/extractor.py

@@ -31,6 +31,11 @@
 from composer.io.context import with_handler, run_graph
 from composer.io.event_handler import NullEventHandler
 from composer.ui.tool_display import tool_display
+from composer.diagnostics.timing import set_current_task_id
+
+# Requirements extraction runs as its own ``run_graph`` (no ``run_task`` scope);
+# set a task_id so the harness tape can address its LLM calls.
+REQUIREMENTS_TASK_ID = "requirements"
 
 
 @dataclass
@@ -172,6 +177,7 @@ async def get_requirements(
         graph_input = ExtractionInput(input=input_text, memory=None, did_read=False)
 
         async with with_handler(io, NullEventHandler()):  # type: ignore[arg-type]
-            final_state = await run_graph(built, ExtractionContext(rag_db=db), graph_input, config, description="Requirements extraction")
+            with set_current_task_id(REQUIREMENTS_TASK_ID):
+                final_state = await run_graph(built, ExtractionContext(rag_db=db), graph_input, config, description="Requirements extraction")  
         assert "reqs" in final_state
         return ExtractionResult(reqs=final_state["reqs"], thread_id=thread_id)

composer/prover/agentic_analyzer.py

@@ -51,8 +51,10 @@
 from composer.prover.ptypes import RuleResult, RulePath, AnalyzedDiagnosis
 from composer.prover.report_store import ReportStore
 from composer.spec.graph_builder import bind_standard, run_to_completion
-from composer.spec.util import uniq_thread_id
+from composer.spec.util import uniq_thread_id, string_hash
 from composer.templates.loader import load_jinja_template
+from composer.diagnostics.timing import set_current_task_id
+from composer.prover.cex_task_ids import cex_rule_task_id, cex_aggregator_task_id
 from composer.tools.thinking import RoughDraftState, get_rough_draft_tools
 
 
@@ -479,8 +481,17 @@ async def _process_rule(
                 for i, entry in enumerate(scratchpad)
             ]
 
+        async def _process_rule_addressed(
+            rule_group: FailingRule,
+        ) -> list[_PerRuleRootCause]:
+            # Distinct task_id lane per concurrent per-rule analysis (keyed by the
+            # prover tool_call_id + rule name) so the harness tape can address the
+            # gathered sub-agents; in production this just scopes timing per rule.
+            with set_current_task_id(cex_rule_task_id(tool_call_id, rule_group.rule_name)):
+                return await _process_rule(rule_group)
+
         per_rule = await asyncio.gather(
-            *(_process_rule(rg) for rg in failing_rules)
+            *(_process_rule_addressed(rg) for rg in failing_rules)
         )
         # Flatten preserving rule order (gather preserves input order).
         all_causes: list[_PerRuleRootCause] = [
@@ -501,9 +512,10 @@ async def _process_rule(
         # Cross-rule aggregator. Sees the per-rule root causes (already
         # within-rule-clustered) and partitions them by cross-rule
         # equivalence.
-        agg = await self._run_aggregator(
-            all_causes, report_fs=report_fs, tool_call_id=tool_call_id,
-        )
+        with set_current_task_id(cex_aggregator_task_id(tool_call_id)):
+            agg = await self._run_aggregator(
+                all_causes, report_fs=report_fs, tool_call_id=tool_call_id,
+            )
 
         # Validator (see ``_aggregator_validator``) already enforced that
         # every cause_index is in range, no duplicates, every input
@@ -514,7 +526,10 @@ async def _process_rule(
 
         for partition in agg.partitions:
             attributed: list[RulePath] = []
-            report_key = uuid.uuid4().hex
+            # Content-addressed rather than a uuid: stable across identical
+            # re-runs and reconstructible by the harness tape (which scripts the
+            # diagnosis text), while staying opaque to the author.
+            report_key = string_hash(partition.diagnosis)
             for i in partition.cause_indices:
                 for attr in all_causes[i].attributed_rule_paths:
                     attributed.append(attr)

composer/prover/cex_task_ids.py

@@ -0,0 +1,23 @@
+"""Task-ids for the agentic CEX analyzer's sub-agents.
+
+The codegen pipeline runs through ``run_graph`` without a ``run_task`` scope, and
+the per-rule CEX analyzers fan out under ``asyncio.gather`` (see
+``AgenticCexHandler.analyze``). Concurrent sub-agents that share one task_id
+can't be routed deterministically by the harness tape, so each is scoped to its
+own task_id lane via ``set_current_task_id``.
+
+The keys fold in the prover ``tool_call_id`` so analyses from distinct prover
+runs land in distinct lanes (otherwise repeated runs of the same rule collide).
+Centralized here so the harness tape reconstructs the same lane keys — the tape
+controls the ``tool_call_id`` it scripts, so the derivation stays deterministic.
+"""
+
+
+def cex_rule_task_id(tool_call_id: str, rule_name: str) -> str:
+    """Lane for the per-rule CEX analyzer of ``rule_name`` under one prover run."""
+    return f"cex-{tool_call_id}-{rule_name}"
+
+
+def cex_aggregator_task_id(tool_call_id: str) -> str:
+    """Lane for the cross-rule CEX aggregator of one prover run."""
+    return f"cex-agg-{tool_call_id}"

composer/spec/cex_remediation.py

@@ -28,7 +28,6 @@
 from typing_extensions import TypedDict
 from dataclasses import dataclass
 import difflib
-import uuid
 
 from pydantic import BaseModel, Field
 
@@ -44,7 +43,7 @@
 from composer.core.state import AIComposerState
 from composer.input.files import Document
 from composer.spec.graph_builder import bind_standard, run_to_completion
-from composer.spec.util import uniq_thread_id
+from composer.spec.util import uniq_thread_id, string_hash
 from composer.tools.thinking import RoughDraftState, get_rough_draft_tools
 from composer.ui.tool_display import tool_display
 from composer.prover.report_store import ReportStore
@@ -430,7 +429,9 @@ async def run(self) -> str:
             # apply_remediation_proposal can fetch it without the
             # codegen author having to re-emit (or paraphrase) the spec
             # text. The codegen author only ever sees the diff + the key.
-            proposal_key = uuid.uuid4().hex
+            # Content-addressed (not a uuid): deterministic across identical
+            # re-runs and reconstructible by the harness tape, still opaque.
+            proposal_key = string_hash(result.proposed_cvl)
             await deps.proposals.record(proposal_key, result.proposed_cvl)
             return _render_remediation(
                 result,

composer/spec/code_explorer.py

@@ -67,9 +67,7 @@ def _code_explorer_graph(
         sys_prompt
     ).with_initial_prompt(
         "Answer the following question about the source code"
-    ).compile_async(
-        checkpointer=InMemorySaver()
-    )
+    ).compile_async()
 
 class _ExploreCodeCommon(BaseModel):
     """

composer/testing/harness_tape.py

@@ -1,4 +1,4 @@
-from typing import Any, Callable, Sequence, override, cast
+from typing import Any, Callable, Iterator, Sequence, override, cast
 import random
 import asyncio
 
@@ -118,6 +118,52 @@ async def ainvoke(
         return cast(AIMessage, lane[i])
 
 
+class _DummyUploader:
+    """A ``FileUploader`` stand-in that never touches a Files API: every input is
+    read from disk and returned as an in-memory text document. Installed under the
+    harness so a taped run does no real uploads — the codegen path uploads the spec
+    + interface via ``upload_text_file_if_needed``, which would otherwise hit the
+    live Files API."""
+
+    async def upload_text_file_if_needed(self, file_path: Any) -> Any:
+        return self._inline(file_path)
+
+    async def upload_file_if_needed(self, file_path: Any) -> Any:
+        return self._inline(file_path)
+
+    async def get_document(self, path: Any) -> Any:
+        import os
+        return self._inline(path) if os.path.isfile(str(path)) else None
+
+    @staticmethod
+    def _inline(path: Any) -> Any:
+        import os
+        from pathlib import Path
+        from composer.input.files import InMemoryTextFile
+        p = str(path)
+        return InMemoryTextFile(
+            basename=os.path.basename(p),
+            string_contents=Path(p).read_text(encoding="utf-8"),
+            provider="anthropic"
+        )
+
+
+# The currently-installed tape state for this process. The seam patches below are
+# installed once per process as stable dispatcher functions that read these slots
+# at call time; re-installing a tape (a later test in the same xdist worker) only
+# swaps the slot. Rebinding the module attributes per install instead would strand
+# any module that imported a patched name by value under the earlier test's tape.
+_active_fake: Any = None
+_active_responses: Iterator[str] | None = None
+_llm_seams_patched = False
+_prompt_seam_patched = False
+
+
+def _current_fake() -> Any:
+    assert _active_fake is not None, "no harness tape installed"
+    return _active_fake
+
+
 def install_fake_llm(fake: Any) -> None:
     """Route every LLM-construction path in the pipeline to ``fake``.
 
@@ -129,19 +175,25 @@ def install_fake_llm(fake: Any) -> None:
     and short-circuits ``get_provider_for`` before it tries to ``_lookup`` a fake
     model name.
 
-    Patches the ``registry`` / ``services`` source bindings, so it must run BEFORE
-    the entry path imports ``get_provider_for`` by name (``composer/bind.py`` is
-    that hook). Eager-import callers (the integration test) additionally rebind
-    their own ``get_provider_for`` reference to the patched one.
+    The first install in a process must run BEFORE the entry path imports
+    ``get_provider_for`` by name (``composer/bind.py`` is that hook). Eager-import
+    callers (the integration tests) additionally rebind their own
+    ``get_provider_for`` reference to the patched one. Later installs just swap
+    the active fake.
     """
+    global _active_fake, _llm_seams_patched
+    _active_fake = fake
+    if _llm_seams_patched:
+        return
+
     import composer.llm.registry as registry
     import composer.workflow.services as services
 
     class _FakeProvider:
         provider : ProviderKind = "anthropic"
 
         def builder_for(self, *, cache_level: Any = None, disable_thinking: bool = False) -> Any:
-            return fake
+            return _current_fake()
 
     fp = _FakeProvider()
 
@@ -153,5 +205,51 @@ def _fake_get_provider_for(
         return fp
 
     registry.get_provider_for = _fake_get_provider_for
-    services.create_llm = lambda args: fake
-    services.create_llm_base = lambda args: fake
+    registry.uploader_for = lambda _provider: _DummyUploader()
+    services.create_llm = lambda args: _current_fake()
+    services.create_llm_base = lambda args: _current_fake()
+    _llm_seams_patched = True
+
+
+def install_fake_responses(responses: list[str]) -> None:
+    """Replay scripted human replies for console HITL interrupts.
+
+    Patches ``composer.ui.prompt.prompt_input`` (and the binding
+    ``composer.ui.console`` imported it under) to return each ``responses`` entry
+    in call order, applying the call's own ``filter`` as a sanity check. Raises if
+    the tape is exhausted or a scripted reply fails the prompt's filter. Install
+    before the entry path imports ``prompt_input`` by name (``composer/bind.py`` is
+    that hook). Replayed alongside ``install_fake_llm``.
+
+    Covers only the console HITL path; the autoprove interactive-refinement
+    conversation uses a different input path and is not handled here.
+    """
+    global _active_responses, _prompt_seam_patched
+    _active_responses = iter(responses)
+    if _prompt_seam_patched:
+        return
+
+    import composer.ui.prompt as prompt_mod
+
+    def _fake_prompt_input(
+        prompt_str: str,
+        debug_thunk: Callable[[], None],
+        filter: Callable[[str], str | None] | None = None,
+    ) -> str:
+        assert _active_responses is not None, "no response tape installed"
+        try:
+            resp = next(_active_responses)
+        except StopIteration:
+            raise RuntimeError(
+                f"response tape exhausted — no scripted reply for HITL prompt: {prompt_str!r}"
+            )
+        if filter is not None and (rejection := filter(resp)) is not None:
+            raise RuntimeError(
+                f"scripted response {resp!r} rejected for prompt {prompt_str!r}: {rejection}"
+            )
+        return resp
+
+    prompt_mod.prompt_input = _fake_prompt_input
+    import composer.ui.console as console_mod
+    console_mod.prompt_input = _fake_prompt_input
+    _prompt_seam_patched = True