How Step Enforcement Prevents Premature Terminal Tool Calls in Forge

Step enforcement uses the StepEnforcer guardrail to track required workflow steps via an internal StepTracker, detects attempts to invoke terminal tools before prerequisites are satisfied, and blocks premature calls by returning escalating nudges until all steps are completed.

In the antoinezambelli/forge repository, step enforcement ensures AI models follow prescribed workflows by preventing early termination. The system guarantees that terminal tools—those that signal task completion—cannot execute until all required intermediate steps have been successfully recorded. This guardrail mechanism maintains workflow integrity across multi-turn interactions by intercepting invalid tool batches before they reach execution.

Required Step Tracking with StepTracker

When instantiating StepEnforcer, you provide a list of tool names that must execute before any terminal tool. The enforcer initializes self._tracker as a StepTracker instance defined in src/forge/core/steps.py to maintain this state.

The tracker records successful tool executions via record() and provides pending() to list incomplete steps and is_satisfied() to verify completion of the full prerequisite set (lines 45-52). This internal state persists throughout the session, creating an auditable trail of which workflow stages have finished.

Detecting Terminal Tools in ToolCall Batches

The check() method in src/forge/guardrails/step_enforcer.py (lines 73-75) scans every proposed batch of ToolCall objects for terminal tools. It compares each tool name against the terminal_tools frozenset provided during initialization.

This detection happens before any tool execution, allowing the system to intercept premature termination attempts. The method differentiates between standard intermediate tools and terminal tools that would end the workflow.

Blocking Premature Terminal Tool Calls

When check() identifies a terminal tool while self._tracker.is_satisfied() returns False, it triggers premature-call prevention. The enforcer increments self._premature_attempts and constructs a StepCheck object with needs_nudge=True (lines 75-88).

This Boolean flag signals to the workflow runner that the current batch must be aborted. The runner returns the generated nudge to the model instead of executing the tools, forcing the model to address pending steps before retrying the terminal call.

Escalating Nudges and Retry Logic

The system implements tiered feedback through step_nudge() in src/forge/prompts/nudges.py (lines 35-60). The escalation tier calculates as min(self._premature_attempts, 3), progressing from polite reminders to direct and aggressive demands across repeated attempts.

The max_premature_attempts configuration limits total retries, while each nudge specifically lists pending steps retrieved from self._tracker.pending(). This graduated response helps correct persistent model behavior without immediately failing the task.

Stateful Enforcement Across Workflow Turns

Unlike stateless validation, StepEnforcer persists throughout a session or task. The step history survives across multiple model turns because the same instance tracks recorded steps via record().

This prevents the model from circumventing requirements by starting fresh batches—only successful executions that invoke record() modify the satisfied state. The enforcer ensures that once steps are completed, subsequent check() calls return StepCheck with needs_nudge=False, allowing the terminal tool to proceed.

Complete Implementation Example

from forge.guardrails.step_enforcer import StepEnforcer
from forge.core.steps import StepTracker
from forge.core.workflow import ToolCall

# Define a workflow where "search" and "summarize" must run before "final_report"

enforcer = StepEnforcer(
    required_steps=["search", "summarize"],
    terminal_tools=frozenset({"final_report"}),
)

# 1️⃣ First model attempt: tries to call the terminal tool too early

batch = [ToolCall(tool="final_report", args={})]

check = enforcer.check(batch)
assert check.needs_nudge  # ✅ Premature call blocked

print(check.nudge.content)   # → polite step‑nudge listing "search, summarize"

# Record the first required step

enforcer.record("search", {"query": "AI safety"})

# Record the second required step

enforcer.record("summarize", {"doc_id": 42})

# 2️⃣ Now the terminal tool is allowed

batch = [ToolCall(tool="final_report", args={"doc_id": 42})]

check = enforcer.check(batch)
assert not check.needs_nudge   # ✅ All steps satisfied, terminal tool proceeds

Summary

• StepTracker maintains the state of required steps through record(), pending(), and is_satisfied() methods in src/forge/core/steps.py
• StepEnforcer.check() scans batches for terminal tools and validates against the terminal_tools frozenset
• Premature attempts increment an internal counter and return StepCheck with needs_nudge=True, blocking execution
• Escalation tiers (calculated as min(premature_attempts, 3)) drive increasingly urgent nudges via step_nudge() in src/forge/prompts/nudges.py
• Stateful persistence across turns prevents circumvention, requiring explicit record() calls to satisfy prerequisites

Frequently Asked Questions

What is the difference between StepEnforcer and StepTracker?

StepEnforcer is the high-level guardrail defined in src/forge/guardrails/step_enforcer.py that orchestrates validation logic and generates nudges. StepTracker is the internal state machine in src/forge/core/steps.py that records tool executions and determines whether prerequisites are satisfied. The enforcer uses the tracker to make blocking decisions but handles the communication protocol with the model.

Can a model bypass step enforcement by including required steps and terminal tools in the same batch?

No. The check() method evaluates the entire batch for the presence of terminal tools when is_satisfied() is False. If any terminal tool appears while steps remain pending, the enforcer blocks the entire batch with a nudge, regardless of what other tools are included. Only after explicit record() calls satisfy prerequisites will terminal tools pass validation.

How does the escalation system work for repeated premature attempts?

The enforcer tracks attempt counts in self._premature_attempts and calculates tiers as min(count, 3), mapping to increasingly direct prompt templates. The first failure generates a polite reminder, while subsequent attempts trigger direct and finally aggressive language through step_nudge() in src/forge/prompts/nudges.py. The system respects max_premature_attempts to prevent infinite loops.

Where are terminal tools defined in a Forge workflow?

Terminal tools are specified as a frozenset during StepEnforcer initialization, typically including completion-signaling tools like final_report. This configuration identifies which tools constitute task termination, allowing the enforcement system to distinguish between intermediate processing steps and workflow endpoints that require strict prerequisite validation.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}