Fork per-class exploration anchor for dynamic access

kimeta · kimeta · commit 1d91c5231214 · 2026-07-03T12:46:18.000+02:00
The dynamic-access iterative strategy previously ran every attempt for a
class on one shared session, so each attempt re-sent the full static
context and dragged the prior attempts' turns forward.

Now each class is explored once in a read-only anchor turn (new
dynamic-access-explore prompt) that plans coverage without writing files.
Every attempt forks that anchor and sends only a slim generation prompt
(dynamic-access-iteration, rewritten) carrying the latest report delta.

Because a fork branches the conversation but not the working tree, the
slim prompt tells the fork to read the on-disk tests and extend them
additively so it does not clobber coverage authored by a sibling attempt
it has no memory of.

A discarded fork child would otherwise drop its tokens from run metrics.
The fork child is now a context manager: on block exit it folds its token
delta (current total minus the baseline captured at fork, before its first
turn) into the anchor's fork-usage accumulator, and the token properties
report own plus forked usage. Call sites just wrap the fork in `with`, so
no per-site accounting is needed.
diff --git a/forge/ai_workflows/agents/agent.py b/forge/ai_workflows/agents/agent.py
@@ -32,6 +32,18 @@ class MyAgent(Agent):
 
     _registry: dict[str, type["Agent"]] = {}
 
+    # Fork token accounting. A fork child inherits this agent's counters, spends
+    # its own tokens, and is then discarded. On the child's ``with``-block exit
+    # it folds its delta into the parent's accumulators below, so a discarded
+    # child's usage still surfaces in run metrics. Token-tracking agents add
+    # these accumulators into their token properties. Class-level defaults let
+    # this work without every backend initializing them in ``__init__``.
+    _forked_tokens_sent: int = 0
+    _forked_tokens_received: int = 0
+    _forked_tokens_cached: int = 0
+    _fork_parent: "Agent | None" = None
+    _fork_baseline: "tuple[int, int, int] | None" = None
+
     @classmethod
     def register(cls, agent_key: str):
         """Class decorator that registers an agent implementation under the given key."""
@@ -68,6 +80,58 @@ def cached_input_tokens_used(self) -> int | None:
         """Return cumulative cached input tokens when the backend reports them."""
         return None
 
+    def add_forked_usage(
+            self,
+            tokens_sent: int,
+            tokens_received: int,
+            cached_input_tokens: int = 0,
+    ) -> None:
+        """Accumulate token usage handed back by a discarded fork child.
+
+        Token-tracking agents surface these accumulators through their token
+        properties; keeping them on the base means the bookkeeping works for any
+        backend without extra per-agent code.
+        """
+        self._forked_tokens_sent += max(int(tokens_sent), 0)
+        self._forked_tokens_received += max(int(tokens_received), 0)
+        self._forked_tokens_cached += max(int(cached_input_tokens), 0)
+
+    def _begin_fork_child(self, parent: "Agent") -> None:
+        """Stamp a freshly forked child with its parent and pre-turn baseline.
+
+        Call this after the child inherits the parent's counters but before its
+        first turn is counted, so the child's delta (``total - baseline``) covers
+        every turn it runs, including the initial generation prompt.
+        """
+        self._fork_parent = parent
+        self._fork_baseline = self._usage_tuple()
+
+    def _usage_tuple(self) -> tuple[int, int, int]:
+        """Current (sent, received, cached) totals, tolerant of ``None`` cached."""
+        return (
+            int(self.total_tokens_sent or 0),
+            int(self.total_tokens_received or 0),
+            int(self.cached_input_tokens_used or 0),
+        )
+
+    def __enter__(self) -> "Agent":
+        return self
+
+    def __exit__(self, exc_type, exc, traceback) -> bool:
+        # Block exit is the fork child's "kill" point: fold its delta into the
+        # parent so the discarded child's tokens survive in run metrics.
+        if self._fork_parent is not None:
+            sent, received, cached = self._usage_tuple()
+            base_sent, base_received, base_cached = self._fork_baseline or (0, 0, 0)
+            self._fork_parent.add_forked_usage(
+                sent - base_sent,
+                received - base_received,
+                cached - base_cached,
+            )
+            self._fork_parent = None
+            self._fork_baseline = None
+        return False
+
     def _create_session_log_path(
             self,
             agent_name: str,
diff --git a/forge/ai_workflows/agents/codex_agent.py b/forge/ai_workflows/agents/codex_agent.py
@@ -128,15 +128,15 @@ def __init__(
 
     @property
     def total_tokens_sent(self) -> int:
-        return self._total_tokens_sent
+        return self._total_tokens_sent + self._forked_tokens_sent
 
     @property
     def total_tokens_received(self) -> int:
-        return self._total_tokens_received
+        return self._total_tokens_received + self._forked_tokens_received
 
     @property
     def cached_input_tokens_used(self) -> int:
-        return self._cached_input_tokens_used
+        return self._cached_input_tokens_used + self._forked_tokens_cached
 
     @property
     def thread_id(self) -> str | None:
@@ -302,6 +302,7 @@ def fork(self, prompt: str) -> "CodexAgent":
         child._total_tokens_sent = self._total_tokens_sent
         child._cached_input_tokens_used = self._cached_input_tokens_used
         child._total_tokens_received = self._total_tokens_received
+        child._begin_fork_child(self)
         child.send_prompt(prompt)
         return child
 
@@ -315,6 +316,7 @@ def compact_fork(self, prompt: str) -> "CodexAgent":
         child._total_tokens_sent = self._total_tokens_sent
         child._cached_input_tokens_used = self._cached_input_tokens_used
         child._total_tokens_received = self._total_tokens_received
+        child._begin_fork_child(self)
         child.send_prompt(prompt)
         return child
 
diff --git a/forge/ai_workflows/agents/pi_agent.py b/forge/ai_workflows/agents/pi_agent.py
@@ -60,15 +60,15 @@ def __init__(
 
     @property
     def total_tokens_sent(self) -> int:
-        return self._total_tokens_sent
+        return self._total_tokens_sent + self._forked_tokens_sent
 
     @property
     def total_tokens_received(self) -> int:
-        return self._total_tokens_received
+        return self._total_tokens_received + self._forked_tokens_received
 
     @property
     def cached_input_tokens_used(self) -> int:
-        return self._cached_input_tokens_used
+        return self._cached_input_tokens_used + self._forked_tokens_cached
 
     def graphify(self, source_dirs: list[str]) -> str:
         """Send /skill:graphify to the Pi session to build a merged knowledge graph context."""
@@ -134,6 +134,7 @@ def fork(self, prompt: str) -> "PiAgent":
         child._prev_input = self._prev_input
         child._prev_output = self._prev_output
         child._prev_cache = self._prev_cache
+        child._begin_fork_child(self)
         child._apply_fork_token_delta(result.session_stats)
         child._print_session_log_once("Pi", child._session_log_path)
         child._write_turn_log(child._session_path, prompt, result)
diff --git a/forge/ai_workflows/core/dynamic_access_iterative_strategy.py b/forge/ai_workflows/core/dynamic_access_iterative_strategy.py
@@ -17,6 +17,7 @@
     DynamicAccessCoverageReport,
     compute_class_delta,
     format_call_sites,
+    format_full_report,
     load_dynamic_access_coverage_report,
 )
 from utility_scripts.dynamic_access_exhaust_report import DynamicAccessExhaustReport
@@ -34,6 +35,7 @@
 FALLBACK_STRATEGY_NAME = "basic_iterative_pi_gpt-5.5"
 DEFAULT_MAX_NATIVE_TEST_VERIFICATION_ITERATIONS = 40
 DEFAULT_NATIVE_TEST_VERIFICATION_BATCH_SIZE = 5
+DEFAULT_DYNAMIC_ACCESS_EXPLORE_PROMPT = "prompt_templates/dynamic_access/dynamic-access-explore.md"
 
 
 @WorkflowStrategy.register("dynamic_access_iterative")
@@ -55,6 +57,10 @@ class DynamicAccessIterativeStrategy(WorkflowStrategy):
 
     def __init__(self, strategy_obj: dict, **context):
         super().__init__(strategy_obj, **context)
+        # The per-class exploration anchor prompt is optional in the bundle: when
+        # a strategy does not declare it, fall back to the packaged default so
+        # existing configs keep working without listing the new prompt key.
+        self.prompts.setdefault("dynamic-access-explore", DEFAULT_DYNAMIC_ACCESS_EXPLORE_PROMPT)
         self.library = self.context["library"]
         self.reachability_repo_path = self.context["reachability_repo_path"]
         self.keep_tests_without_dynamic_access = bool(self.context.get("keep_tests_without_dynamic_access", False))
@@ -360,6 +366,12 @@ def flush_native_test_batch(reason: str, active_class_name: str | None = None) -
             class_failed = False
             class_committed = False
             gate_failed = False
+            # Build one exploration anchor per class, then fork it for each
+            # attempt: the shared exploration is reused, but each attempt starts
+            # from the same lean, pre-generation state instead of accumulating
+            # the previous attempts' turns (§WF-dynamic-access-iterative-strategy).
+            agent.clear_context()
+            self._explore_class(agent, class_name, active_class, current_report)
             while class_attempts < self.max_class_iterations:
                 self._print_dynamic_access_detail(
                     "attempt {attempt}/{max_attempts}".format(
@@ -375,53 +387,55 @@ def flush_native_test_batch(reason: str, active_class_name: str | None = None) -
                     dynamic_access_progress=self._format_progress(delta, class_attempts),
                     uncovered_dynamic_access_calls=format_call_sites(active_class.uncovered_call_sites),
                 )
-                self._print_dynamic_access_detail("agent: running dynamic-access prompt", indent_level=2)
-                agent.send_prompt(dynamic_prompt)
-                self._print_dynamic_access_detail("agent: complete", indent_level=2)
-                prompt_iterations += 1
-                save_phase_update(
-                    self.continuation_marker_path,
-                    lambda marker: marker.record_iteration(PHASE_EXPLORE, prompt_iterations),
-                )
-                class_attempts += 1
-
+                self._print_dynamic_access_detail("agent: forking exploration anchor for generation", indent_level=2)
                 reached_native_test = False
                 last_test_output = ""
                 last_failed_task = None
-                for test_iteration in range(self.max_class_test_iterations):
-                    self._print_dynamic_access_detail(
-                        "test {current}/{maximum}: running ./gradlew test -Pcoordinates={library}".format(
-                            current=test_iteration + 1,
-                            maximum=self.max_class_test_iterations,
-                            library=self.library,
-                        ),
-                        indent_level=2,
-                    )
-                    test_output = agent.run_test_command(f"./gradlew test -Pcoordinates={self.library}")
-                    failed_task = self._get_first_failed_task(test_output)
-                    last_test_output = test_output
-                    last_failed_task = failed_task
-                    self._print_dynamic_access_detail(
-                        "test: complete (failed task: {failed_task})".format(
-                            failed_task=failed_task or "none",
-                        ),
-                        indent_level=2,
-                    )
-                    if failed_task in {"nativeTest", None}:
-                        reached_native_test = True
-                        break
-                    self._print_dynamic_access_detail(
-                        "agent: test failed before nativeTest; sending failure output back to agent",
-                        indent_level=2,
-                    )
-                    agent.send_prompt(
-                        "When `./gradlew test -Pcoordinates={library}` is ran this is the error:\n{error_output}".format(
-                            library=self.library,
-                            error_output=test_output,
-                        )
-                    )
+                # Fork the anchor for this attempt. The child folds its own token
+                # usage into the anchor when the `with` block exits, so no per-site
+                # accounting is needed here (see Agent.fork / Agent.__exit__).
+                with agent.fork(dynamic_prompt) as attempt_agent:
                     self._print_dynamic_access_detail("agent: complete", indent_level=2)
                     prompt_iterations += 1
+                    save_phase_update(
+                        self.continuation_marker_path,
+                        lambda marker: marker.record_iteration(PHASE_EXPLORE, prompt_iterations),
+                    )
+                    class_attempts += 1
+
+                    for test_iteration in range(self.max_class_test_iterations):
+                        self._print_dynamic_access_detail(
+                            "test {current}/{maximum}: running ./gradlew test -Pcoordinates={library}".format(
+                                current=test_iteration + 1,
+                                maximum=self.max_class_test_iterations,
+                                library=self.library,
+                            ),
+                            indent_level=2,
+                        )
+                        test_output = attempt_agent.run_test_command(f"./gradlew test -Pcoordinates={self.library}")
+                        failed_task = self._get_first_failed_task(test_output)
+                        last_test_output = test_output
+                        last_failed_task = failed_task
+                        self._print_dynamic_access_detail(
+                            "test: complete (failed task: {failed_task})".format(
+                                failed_task=failed_task or "none",
+                            ),
+                            indent_level=2,
+                        )
+                        if failed_task in {"nativeTest", None}:
+                            reached_native_test = True
+                            break
+                        self._print_dynamic_access_detail(
+                            "agent: test failed before nativeTest; sending failure output back to agent",
+                            indent_level=2,
+                        )
+                        failure_message = (
+                            f"When `./gradlew test -Pcoordinates={self.library}` is ran "
+                            f"this is the error:\n{test_output}"
+                        )
+                        attempt_agent.send_prompt(failure_message)
+                        self._print_dynamic_access_detail("agent: complete", indent_level=2)
+                        prompt_iterations += 1
 
                 if not reached_native_test:
                     # Tests failed for this class, roll back to before this class
@@ -550,8 +564,8 @@ def flush_native_test_batch(reason: str, active_class_name: str | None = None) -
                     )
                 active_class = updated_class
 
-            # Clear context between classes to keep the agent window focused.
-            agent.clear_context()
+            # The exploration anchor is rebuilt from scratch for the next class
+            # (clear + explore at the top of the loop), so no clear is needed here.
             if gate_failed:
                 self._print_failure_analysis(
                     "native_test_verification_gate_failed",
@@ -666,6 +680,25 @@ def flush_native_test_batch(reason: str, active_class_name: str | None = None) -
                 )
                 return True, prompt_iterations
 
+    def _explore_class(self, agent, class_name: str, active_class, current_report) -> None:
+        """Prime a per-class exploration anchor that later attempts fork from.
+
+        This is the read-only exploration turn of
+        §WF-dynamic-access-iterative-strategy: the agent studies the active class
+        and the complete report and plans coverage, but writes nothing. Each
+        attempt then forks this anchor so the exploration is reused without
+        carrying prior attempts' turns forward.
+        """
+        explore_prompt = self._render_prompt(
+            "dynamic-access-explore",
+            active_class_name=class_name,
+            active_class_source_file=active_class.resolved_source_file or active_class.source_file or "N/A",
+            dynamic_access_report=format_full_report(current_report),
+        )
+        self._print_dynamic_access_detail("agent: exploring class (no test generation)", indent_level=2)
+        agent.send_prompt(explore_prompt)
+        self._print_dynamic_access_detail("agent: exploration complete", indent_level=2)
+
     def _mark_continuation_explore_completed(
             self,
             iteration: int,
diff --git a/forge/docs/workflows/dynamic-access.md b/forge/docs/workflows/dynamic-access.md
@@ -39,6 +39,18 @@ classes. Each selected class receives bounded prompt attempts, bounded test
 repair attempts, coverage-delta checks, per-class checkpointing, and native-test
 verification after each configured batch of coverage-gain commits.
 
+Within a class, the engine first primes a per-class exploration anchor: a single
+read-only turn (the `dynamic-access-explore` prompt) that studies the active
+class and the complete report and plans coverage without writing any files. Each
+attempt then **forks** that anchor and sends only a slim generation prompt (the
+`dynamic-access-iteration` prompt) carrying the latest report delta, so the
+shared exploration is reused once instead of re-accumulating across attempts.
+Because a fork branches the conversation but not the working tree, the slim
+prompt instructs the fork to read the tests already on disk and extend them
+additively so it does not clobber coverage authored by a sibling attempt it has
+no memory of. Token usage from each discarded fork child is folded back into the
+anchor so run metrics stay accurate.
+
 The iterative engine may return `RUN_STATUS_CHUNK_READY` when chunked issue
 orchestration gave it a class limit and the current chunk passed all local
 verification gates. Otherwise it returns success only when the dynamic-access
diff --git a/forge/prompt_templates/dynamic_access/dynamic-access-explore.md b/forge/prompt_templates/dynamic_access/dynamic-access-explore.md
@@ -0,0 +1,33 @@
+Role: You are an expert JVM test engineer specializing in high-coverage integration testing for external libraries.
+
+Task:
+Explore — do not write tests yet. This is a read-only planning turn for the active dynamic-access class in `{library}`. Subsequent turns will fork this session to generate the tests, so build a durable understanding here that those turns can rely on.
+
+Source Context:
+{source_context_overview}
+
+{resolved_edit_scope_context}
+
+Issue-Requested Metadata:
+{issue_requested_metadata_context}
+
+Library Preparation Preflight:
+{library_preparation_preflight_context}
+
+Active class (focus your exploration here):
+- Class: {active_class_name}
+- Source file: {active_class_source_file}
+
+Complete dynamic-access report for this library:
+{dynamic_access_report}
+
+What to produce:
+- Read the active class source and any collaborators needed to reach its uncovered dynamic-access call sites.
+- Identify the public API entry points that drive execution to each uncovered call site of the active class.
+- Note the setup, fixtures, and inputs a test would need, and any call sites reachable only through broken or version-specific paths that should be left alone.
+- Summarize a concrete coverage plan for the active class that a follow-up test-writing turn can execute.
+
+Rules:
+- Do NOT create, edit, or delete any files during this turn. Exploration only.
+- Focus on the active class; use the rest of the report only as context for shared setup or collaborators.
+- Prefer coverage through the library's normal public API. Do not plan to satisfy coverage by asserting a known broken, regressed, or version-specific failure path.
diff --git a/forge/prompt_templates/dynamic_access/dynamic-access-iteration.md b/forge/prompt_templates/dynamic_access/dynamic-access-iteration.md
diff --git a/forge/tests/test_agent_fork_usage.py b/forge/tests/test_agent_fork_usage.py
diff --git a/forge/tests/test_dynamic_access_iterative_strategy.py b/forge/tests/test_dynamic_access_iterative_strategy.py
