Add conditional tool dependencies (#35)

* Add tool prerequisite foundation: ToolDef.prerequisites, StepTracker arg tracking, StepEnforcer prereq checking

Adds the data model and enforcement logic for conditional tool dependencies
(e.g. "must call read_file before edit_file"). Backwards-compatible — all
new fields are optional with safe defaults.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Wire prerequisite enforcement into runner with PrerequisiteError and PREREQUISITE_NUDGE

Runner builds prereq dict from ToolDefs, checks before execution, blocks
batch on violation with nudge, raises PrerequisiteError on exhaustion.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Add PREREQUISITE_NUDGE to compaction and validate prereq tool names in Workflow

TieredCompact drops prerequisite nudges in all three phases (same priority
as step nudges). Workflow.__post_init__ validates prerequisite tool names
exist in the tools dict.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* Add compaction, workflow validation, and tests for tool prerequisites

TieredCompact drops PREREQUISITE_NUDGE in all phases. Workflow validates
prereq tool names exist. 26 new tests covering name-only, arg-matched,
mixed prereqs, batch blocking, exhaustion, runner integration, and nudge
templates. 703 total tests passing.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: antoinezambelli

src/forge/__init__.py

@@ -46,6 +46,7 @@
     ForgeError,
     HardwareDetectionError,
     MaxIterationsError,
+    PrerequisiteError,
     StepEnforcementError,
     StreamError,
     ThinkingNotSupportedError,
@@ -124,6 +125,7 @@
     "ForgeError",
     "HardwareDetectionError",
     "MaxIterationsError",
+    "PrerequisiteError",
     "StepEnforcementError",
     "StreamError",
     "ThinkingNotSupportedError",

src/forge/context/strategies.py

@@ -210,6 +210,7 @@ def _phase1(
             if 2 <= i < eligible_end:
                 if msg.metadata.type in (
                     MessageType.STEP_NUDGE,
+                    MessageType.PREREQUISITE_NUDGE,
                     MessageType.RETRY_NUDGE,
                 ):
                     continue
@@ -237,6 +238,7 @@ def _phase2(
             if 2 <= i < eligible_end:
                 if msg.metadata.type in (
                     MessageType.STEP_NUDGE,
+                    MessageType.PREREQUISITE_NUDGE,
                     MessageType.RETRY_NUDGE,
                     MessageType.TOOL_RESULT,
                 ):
@@ -253,6 +255,7 @@ def _phase3(
             if 2 <= i < eligible_end:
                 if msg.metadata.type in (
                     MessageType.STEP_NUDGE,
+                    MessageType.PREREQUISITE_NUDGE,
                     MessageType.RETRY_NUDGE,
                     MessageType.TOOL_RESULT,
                     MessageType.REASONING,

src/forge/core/inference.py

@@ -25,6 +25,7 @@
     "retry": MessageType.RETRY_NUDGE,
     "unknown_tool": MessageType.RETRY_NUDGE,
     "step": MessageType.STEP_NUDGE,
+    "prerequisite": MessageType.PREREQUISITE_NUDGE,
 }
 
 

src/forge/core/messages.py

@@ -27,6 +27,7 @@ class MessageType(str, Enum):
     REASONING = "reasoning"
     TEXT_RESPONSE = "text_response"
     STEP_NUDGE = "step_nudge"
+    PREREQUISITE_NUDGE = "prerequisite_nudge"
     RETRY_NUDGE = "retry_nudge"
     CONTEXT_WARNING = "context_warning"
     SUMMARY = "summary"

src/forge/core/runner.py

@@ -12,7 +12,7 @@
 from forge.core.inference import _NUDGE_KIND_TO_TYPE, _build_tool_call_infos, run_inference
 from forge.core.messages import Message, MessageMeta, MessageRole, MessageType, ToolCallInfo
 from forge.core.workflow import ToolCall, TextResponse, Workflow, ToolSpec
-from forge.errors import MaxIterationsError, StepEnforcementError, ToolCallError, ToolExecutionError, ToolResolutionError
+from forge.errors import MaxIterationsError, PrerequisiteError, StepEnforcementError, ToolCallError, ToolExecutionError, ToolResolutionError
 from forge.guardrails import ErrorTracker, ResponseValidator, StepEnforcer
 
 
@@ -121,9 +121,15 @@ def _emit(msg: Message) -> None:
         # Step 2 — Initialize guardrail middleware
         tool_names = list(workflow.tools.keys())
         validator = ResponseValidator(tool_names, rescue_enabled=self.rescue_enabled)
+        tool_prerequisites = {
+            name: td.prerequisites
+            for name, td in workflow.tools.items()
+            if td.prerequisites
+        }
         step_enforcer = StepEnforcer(
             required_steps=workflow.required_steps,
             terminal_tool=workflow.terminal_tool,
+            tool_prerequisites=tool_prerequisites,
         )
         error_tracker = ErrorTracker(
             max_retries=self.max_retries_per_step,
@@ -206,6 +212,46 @@ def _emit(msg: Message) -> None:
                 ))
                 continue
 
+            # 3b.2 — Check prerequisites
+            prereq_check = step_enforcer.check_prerequisites(tool_calls)
+
+            if prereq_check.needs_nudge:
+                if step_enforcer.prereq_exhausted:
+                    # Find the first violating tool for the error
+                    for tc in tool_calls:
+                        prereqs = tool_prerequisites.get(tc.tool)
+                        if prereqs:
+                            result = step_enforcer._tracker.check_prerequisites(
+                                tc.tool, tc.args, prereqs,
+                            )
+                            if not result.satisfied:
+                                raise PrerequisiteError(
+                                    tool_name=tc.tool,
+                                    violations=step_enforcer.prereq_violations,
+                                    missing_prereqs=result.missing,
+                                )
+                if tool_calls[0].reasoning:
+                    _emit(Message(
+                        MessageRole.ASSISTANT,
+                        tool_calls[0].reasoning,
+                        MessageMeta(MessageType.REASONING, step_index=iteration),
+                    ))
+                tc_infos, tool_call_counter = _build_tool_call_infos(tool_calls, tool_call_counter)
+                _emit(Message(
+                    MessageRole.ASSISTANT,
+                    "",
+                    MessageMeta(MessageType.TOOL_CALL, step_index=iteration),
+                    tool_calls=tc_infos,
+                ))
+                nudge = prereq_check.nudge
+                nudge_type = _NUDGE_KIND_TO_TYPE[nudge.kind]
+                _emit(Message(
+                    MessageRole.USER,
+                    nudge.content,
+                    MessageMeta(nudge_type, step_index=iteration),
+                ))
+                continue
+
             # 3c — Execute all tool calls in the batch
             tc_infos, tool_call_counter = _build_tool_call_infos(tool_calls, tool_call_counter)
             call_ids = [tc.call_id for tc in tc_infos]
@@ -262,7 +308,7 @@ def _emit(msg: Message) -> None:
                     continue
 
                 # Success
-                step_enforcer.record(tc.tool)
+                step_enforcer.record(tc.tool, tc.args)
                 result_str = result_val if isinstance(result_val, str) else json.dumps(result_val)
                 _emit(Message(
                     MessageRole.TOOL,
@@ -287,6 +333,7 @@ def _emit(msg: Message) -> None:
             else:
                 error_tracker.reset_errors()
                 step_enforcer.reset_premature()
+                step_enforcer.reset_prereq_violations()
 
             # 3e — If terminal tool was in the batch and succeeded, return
             if terminal_result is not None and not isinstance(terminal_result, Exception):

src/forge/core/steps.py

@@ -1,24 +1,46 @@
-"""Required-step tracking for workflow enforcement."""
+"""Required-step tracking and prerequisite enforcement."""
 
 from __future__ import annotations
 
 from dataclasses import dataclass, field
+from typing import Any
+
+
+@dataclass
+class PrerequisiteCheck:
+    """Result of checking prerequisites for a tool call.
+
+    If ``satisfied`` is False, ``missing`` lists the prerequisite tool names
+    that have not been called (or not called with matching args).
+    """
+
+    satisfied: bool
+    missing: list[str]
 
 
 @dataclass
 class StepTracker:
-    """Tracks which required steps have been completed.
+    """Tracks which required steps have been completed and which tools
+    have been executed (with args) for prerequisite enforcement.
 
     Lives on the WorkflowRunner, outside the message history.
     Compaction cannot invalidate step completion. See P0-1.
     """
 
     required_steps: list[str]
     completed_steps: dict[str, None] = field(default_factory=dict)
+    executed_tools: dict[str, list[dict[str, Any]]] = field(default_factory=dict)
 
-    def record(self, tool_name: str) -> None:
-        """Record a tool call as completed."""
+    def record(self, tool_name: str, args: dict[str, Any] | None = None) -> None:
+        """Record a successful tool execution.
+
+        Args:
+            tool_name: The tool that was executed.
+            args: The arguments the tool was called with. Stored for
+                arg-matched prerequisite checking.
+        """
         self.completed_steps[tool_name] = None
+        self.executed_tools.setdefault(tool_name, []).append(args or {})
 
     def is_satisfied(self) -> bool:
         """True if all required steps have been called."""
@@ -28,6 +50,45 @@ def pending(self) -> list[str]:
         """Return required steps not yet completed, preserving original order."""
         return [s for s in self.required_steps if s not in self.completed_steps]
 
+    def check_prerequisites(
+        self,
+        tool_name: str,
+        args: dict[str, Any],
+        prerequisites: list[str | dict[str, str]],
+    ) -> PrerequisiteCheck:
+        """Check whether prerequisites are satisfied for a tool call.
+
+        Args:
+            tool_name: The tool about to be called (for error context).
+            args: The arguments the tool is being called with.
+            prerequisites: The prerequisite definitions from ToolDef.
+
+        Returns:
+            PrerequisiteCheck with satisfied=True if all prereqs are met,
+            or satisfied=False with the list of unsatisfied prereq tool names.
+        """
+        missing: list[str] = []
+        for prereq in prerequisites:
+            if isinstance(prereq, str):
+                # Name-only: any prior call to this tool satisfies it
+                if prereq not in self.executed_tools:
+                    missing.append(prereq)
+            elif isinstance(prereq, dict):
+                # Arg-matched: a prior call with the same arg value satisfies it
+                prereq_tool = prereq["tool"]
+                match_arg = prereq["match_arg"]
+                required_value = args.get(match_arg)
+                if prereq_tool not in self.executed_tools:
+                    missing.append(prereq_tool)
+                    continue
+                if not any(
+                    call.get(match_arg) == required_value
+                    for call in self.executed_tools[prereq_tool]
+                ):
+                    missing.append(prereq_tool)
+
+        return PrerequisiteCheck(satisfied=len(missing) == 0, missing=missing)
+
     def summary_hint(self) -> str:
         """Human-readable hint for injection into compacted summaries."""
         if not self.completed_steps:

src/forge/core/workflow.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 from collections.abc import Callable
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import Any, Literal, Optional
 
 from pydantic import BaseModel, ConfigDict, Field, create_model
@@ -145,10 +145,17 @@ class ToolDef:
     Downstream projects define tools as ToolDefs. The Workflow holds these
     in a dict keyed by name, deriving the spec list (for the LLM) and
     callable lookup (for execution) internally.
+
+    Prerequisites express conditional dependencies: "if you call this tool,
+    you must have called tool X first." Entries can be:
+    - str: name-only ("read_file" — any prior call to read_file satisfies it)
+    - dict: arg-matched ({"tool": "read_file", "match_arg": "path"} — a prior
+      call to read_file with the same ``path`` value satisfies it)
     """
 
     spec: ToolSpec
     callable: Callable[..., Any]
+    prerequisites: list[str | dict[str, str]] = field(default_factory=list)
 
     @property
     def name(self) -> str:
@@ -214,6 +221,14 @@ def __post_init__(self) -> None:
             raise ValueError(
                 f"Terminal tool '{self.terminal_tool}' cannot also be a required step"
             )
+        for key, tool_def in self.tools.items():
+            for prereq in tool_def.prerequisites:
+                prereq_name = prereq if isinstance(prereq, str) else prereq["tool"]
+                if prereq_name not in tool_names:
+                    raise ValueError(
+                        f"Prerequisite '{prereq_name}' for tool '{key}' "
+                        f"not in tools: {tool_names}"
+                    )
 
     def build_system_prompt(self, **kwargs: str) -> str:
         """Render the system prompt with user-provided values."""

src/forge/errors.py

@@ -88,6 +88,24 @@ def __init__(
         self.pending_steps = pending_steps
 
 
+class PrerequisiteError(ForgeError):
+    """Model repeatedly called a tool without satisfying its prerequisites."""
+
+    def __init__(
+        self,
+        tool_name: str,
+        violations: int,
+        missing_prereqs: list[str],
+    ):
+        super().__init__(
+            f"Tool '{tool_name}' called {violations} times "
+            f"without satisfying prerequisites: {missing_prereqs}"
+        )
+        self.tool_name = tool_name
+        self.violations = violations
+        self.missing_prereqs = missing_prereqs
+
+
 class ContextBudgetExceeded(ForgeError):
     """Context exceeded budget even after compaction. Unrecoverable."""