feedforward-controls

name: feedforward-controls description: > Use this skill to implement proactive planning and anticipation mechanisms that steer AI agents BEFORE they act. Covers OODA loops, Plan-and-Execute patterns, task decomposition, pre-flight validation, intent classification, action planning, resource pre-allocation, constraint propagation, goal decomposition trees, and anticipatory error prevention. This skill enforces: structured observation-orientation-decision-action cycles, hierarchical goal decomposition, constraint satisfaction propagation, and pre-execution validation gates. Do NOT use for: post-execution correction, feedback loops, output verification, or retrospective analysis. version: "2.0.0" author: "j4flmao" license: "MIT" type: skill compatibility: claude-code: true cursor: true codex: true windsurf: true tags: [harness-engineering, feedforward-controls, agent-planning, ooda-loop, task-decomposition]

Feedforward Controls Skill

Purpose

Establishes a production-grade proactive control framework for AI agent execution. Feedforward controls operate on the principle that planning, validation, and constraint analysis BEFORE action produces superior outcomes compared to reactive correction. This system implements structured observation-orientation-decision-action cycles, hierarchical task decomposition, intent classification pipelines, pre-flight validation gates, and anticipatory error prevention mechanisms. The goal is to minimize wasted computation, prevent invalid actions, and ensure agents pursue well-structured plans aligned with user intent.

Core Principles

1. Plan Before Execute: Every agent action must be preceded by an explicit planning phase. No execution without a validated plan artifact.
2. Constraint Propagation First: Propagate constraints through the entire plan tree before committing to any action. Early constraint violations prevent cascading failures.
3. Intent Alignment Gate: Classify and confirm user intent before decomposing goals. Misclassified intent poisons the entire execution pipeline.
4. Hierarchical Decomposition: Break complex goals into atomic sub-tasks with clear preconditions, postconditions, and dependency edges. Never execute monolithic plans.
5. Anticipatory Validation: Simulate execution paths mentally before committing resources. Pre-flight checks catch errors at zero cost compared to runtime failures.

Agent Protocol

Triggers

Use this skill when processing:

• Complex multi-step tasks requiring structured planning.
• Tasks with resource constraints (token budgets, API rate limits, file system operations).
• Ambiguous user requests requiring intent classification.
• Operations with irreversible side effects (file writes, API calls, deployments).
• Goal hierarchies with interdependent sub-tasks.
• Scenarios where execution cost of errors is high.

Input Context Required

• User Request: The raw natural language instruction or goal statement.
• Available Resources: List of tools, APIs, file system access, and token budgets.
• Constraint Set ($\mathcal{C}$): Hard constraints (must-satisfy) and soft constraints (should-satisfy) on the plan.
• Domain Context: Relevant codebase structure, project conventions, and prior execution history.
• Risk Tolerance ($\rho$): A scalar [0,1] indicating acceptable failure probability.

Output Artifact

• Execution Plan: A validated, ordered sequence of atomic actions with dependency edges.
• Constraint Satisfaction Report: Verification that all hard constraints are met.
• Intent Classification Result: Structured classification of user intent with confidence scores.
• Pre-flight Validation Log: Results of all pre-execution checks.

Response Formats

For programmatic compilation, the output must be delivered in this format:

{
  "intent": {
    "primary_class": "code_modification",
    "confidence": 0.94,
    "sub_intents": ["refactor", "add_feature"]
  },
  "plan": {
    "phases": [
      {
        "id": "phase_1",
        "name": "Analysis",
        "tasks": [
          {"id": "t1", "action": "read_file", "target": "src/main.py", "preconditions": [], "postconditions": ["file_content_loaded"]}
        ]
      }
    ],
    "dependency_graph": {"t2": ["t1"], "t3": ["t1"]},
    "estimated_cost": {"tokens": 4200, "api_calls": 3}
  },
  "constraints_satisfied": true,
  "preflight_passed": true,
  "risk_assessment": 0.12
}

Decision Matrix for Feedforward Control

Incoming Request Analysis
├── Intent Clear?
│   ├── YES
│   │   └── Complexity Assessment
│   │       ├── Simple (1-2 steps)
│   │       │   → Direct Plan-and-Execute with pre-flight validation.
│   │       │
│   │       ├── Moderate (3-8 steps)
│   │       │   → OODA Loop with task decomposition tree.
│   │       │   → Propagate constraints before execution.
│   │       │
│   │       └── Complex (9+ steps or multi-domain)
│   │           → Full hierarchical goal decomposition.
│   │           → Constraint propagation + resource pre-allocation.
│   │           → Multi-phase Plan-and-Execute with checkpoints.
│   │
│   └── NO
│       ├── Ambiguous Intent
│       │   → Run intent classification pipeline.
│       │   → Request clarification if confidence < 0.75.
│       │
│       └── Contradictory Constraints
│           → Execute constraint satisfaction analysis.
│           → Report conflicts to user before proceeding.

Detailed Architectural Overview

Feedforward controls form the upstream planning layer that precedes all agent execution. Below is the comprehensive architecture mapping observation through planning to validated execution.

+-------------+       +------------------+       +-------------------+       +--------------------+       +--------------+
| User Request| ───►  | Intent Classifier| ───►  | Goal Decomposer   | ───►  | Constraint Engine  | ───►  | Plan Validator|
+-------------+       +------------------+       +-------------------+       +--------------------+       +--------------+
                                                                                                                  │
                                                                                                                  ▼
+--------------+                                                                                          +--------------+
| Agent Engine |  ◄──────────────────────────────────────────────────────────────────────────────────────  | Pre-flight   |
| (Execution)  |                                                                                          | Gate         |
+--------------+                                                                                          +--------------+

Feedforward Control Lifecycle

Below is the execution pipeline for proactive planning:

[User Request Received]
       │
       ├──► (A) OODA Observe ──► Gather context, read files, analyze codebase
       │
       ├──► (B) OODA Orient ──► Classify intent, assess complexity, identify constraints
       │
       ├──► (C) OODA Decide ──► Decompose goals, build task tree, propagate constraints
       │
       ├──► (D) Pre-flight Gate ──► Validate plan feasibility, check resource availability
       │
       └──► (E) OODA Act ──► Execute validated plan with monitoring hooks attached

Workflow Steps

Phase 1: Observation & Context Gathering

1. Parse User Request: Extract explicit instructions, implicit requirements, and contextual references from the raw input.
2. Scan Environment State: Read relevant files, check tool availability, and assess current system state.
3. Identify Domain Signals: Detect programming languages, frameworks, project conventions, and architectural patterns.
4. Catalog Available Resources: Enumerate tools, API endpoints, token budgets, and time constraints.

Phase 2: Orientation & Intent Classification

1. Classify Primary Intent: Map the user request to a canonical intent category (create, modify, debug, explain, deploy).
2. Extract Sub-Intents: Identify secondary objectives embedded within the primary request.
3. Assess Complexity Score: Calculate task complexity based on step count, domain breadth, and constraint density.
4. Evaluate Risk Profile: Score the potential impact of incorrect execution on the codebase or system.

Phase 3: Goal Decomposition & Planning

1. Build Goal Tree: Decompose the top-level goal into hierarchical sub-goals with AND/OR relationships.
2. Generate Task Sequence: Convert leaf-level goals into ordered, atomic action steps with clear preconditions.
3. Map Dependencies: Construct a directed acyclic graph (DAG) of task dependencies.
4. Allocate Resources: Pre-assign token budgets, tool selections, and execution time estimates per task.

Phase 4: Constraint Propagation

1. Extract Hard Constraints: Identify must-satisfy constraints from user instructions, project configs, and system limits.
2. Propagate Through Plan Tree: Use arc consistency algorithms to prune infeasible branches early.
3. Detect Constraint Conflicts: Identify contradictions between constraints and flag for resolution.
4. Compute Feasibility Score: Calculate the probability of successful plan completion given current constraints.

Phase 5: Pre-Flight Validation

1. Verify File Access: Confirm all target files exist and are writable before planning modifications.
2. Check Tool Availability: Ensure all required tools and APIs are accessible and within rate limits.
3. Validate Token Budget: Confirm the plan fits within the available context window and output token budget.
4. Simulate Execution Path: Mentally trace through the plan to identify logical errors or missing steps.

Phase 6: Plan Execution Handoff

1. Serialize Plan Artifact: Generate the structured plan document with all metadata.
2. Attach Monitoring Hooks: Wire up feedback loop triggers for post-execution verification.
3. Set Rollback Points: Define checkpoints where execution can be safely reversed if needed.
4. Initiate Execution: Hand the validated plan to the agent execution engine.

Extended Troubleshooting Guide

When implementing feedforward control configurations, you may encounter the following common failure modes:

Complete Execution Scenario

Let's inspect how the feedforward pipeline behaves under a multi-file refactoring request:

[User Request] ──► "Refactor the auth module to use JWT instead of session tokens"
                        │
[Observe] ──► Read auth/ directory ──► Identify 4 files: auth.py, middleware.py, config.py, tests.py
                                            │
[Orient] ──► Intent: code_modification (0.96) ──► Complexity: Moderate (6 steps)
                                                       │
[Decide] ──► Goal Tree:
             ├── Replace session token generation → auth.py
             ├── Update middleware validation → middleware.py
             ├── Add JWT config parameters → config.py
             └── Update test assertions → tests.py
                        │
[Pre-flight] ──► All files exist ✓ ──► No constraint conflicts ✓ ──► Budget OK ✓
                        │
[Act] ──► Execute tasks in dependency order ──► t1 → t2 → t3 → t4

Rules and Guidelines

• Rule 1: Never execute an action without a preceding plan artifact. Even single-step tasks require explicit pre-flight validation.
• Rule 2: Intent classification must achieve confidence ≥ 0.75 before proceeding to goal decomposition. Below this threshold, request user clarification.
• Rule 3: Constraint propagation must complete before any resource allocation. Do not pre-allocate resources to infeasible plan branches.
• Rule 4: Goal decomposition trees must be acyclic. Circular dependencies indicate a modeling error that must be resolved before execution.
• Rule 5: Pre-flight validation failures are hard stops. Do not bypass pre-flight gates under any circumstances.

Reference Guides

Below are links to the reference guides detailing the algorithms, patterns, and implementations used in this feedforward control framework:

• ooda-loop-patterns.md Provides OODA loop implementation patterns for AI agents, including cycle timing, observation strategies, orientation heuristics, decision frameworks, and action execution protocols.
• plan-execute-architectures.md Details Plan-and-Execute agent architectures, including LangChain-style planners, ReAct variants, and multi-phase execution engines with re-planning capabilities.
• task-decomposition-strategies.md Covers hierarchical task decomposition strategies, AND/OR goal trees, recursive decomposition algorithms, and atomic task specification formats.
• preflight-validation.md Defines pre-execution validation check suites, file system validators, token budget verifiers, API availability checkers, and constraint satisfaction verifiers.
• intent-classification.md Outlines intent classification pipelines for agent systems, including multi-label classifiers, confidence calibration, and disambiguation strategies.
• constraint-propagation.md Explains constraint propagation algorithms (AC-3, backtracking), constraint satisfaction problems in agent planning, and conflict resolution strategies.
• goal-decomposition-trees.md Covers hierarchical goal decomposition with AND/OR trees, goal refinement operators, leaf-node task generation, and dependency graph construction.
• anticipatory-error-prevention.md Explores proactive error prevention mechanisms, failure mode prediction, defensive planning patterns, and pre-emptive mitigation strategies.

Handoff

For projects requiring post-execution verification and correction, hand off to feedback-loops. For systems implementing core orchestrator loops, hand off to core-master-orchestrator. For context window optimization within plans, hand off to context-engineering.

Implementation Patterns

OODA Loop Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
import json

@dataclass
class OODAState:
    observation: Dict = field(default_factory=dict)
    orientation: Dict = field(default_factory=dict)
    decision: Dict = field(default_factory=dict)
    action: Dict = field(default_factory=dict)
    iteration: int = 0

class OODALoop:
    def __init__(self, observe_fn: Callable, orient_fn: Callable,
                 decide_fn: Callable, act_fn: Callable):
        self.observe_fn = observe_fn
        self.orient_fn = orient_fn
        self.decide_fn = decide_fn
        self.act_fn = act_fn
        self.state = OODAState()

    async def execute(self, goal: str, context: Dict) -> Dict:
        self.state.iteration += 1
        self.state.observation = await self.observe_fn(goal, context)
        self.state.orientation = await self.orient_fn(
            self.state.observation
        )
        self.state.decision = await self.decide_fn(
            self.state.orientation
        )
        self.state.action = await self.act_fn(
            self.state.decision
        )
        return {
            "iteration": self.state.iteration,
            "observation": self.state.observation,
            "orientation": self.state.orientation,
            "decision": self.state.decision,
            "action": self.state.action,
        }

class PlanExecutor:
    def __init__(self, max_retries: int = 2):
        self.max_retries = max_retries

    def decompose_goal(self, goal: str, depth: int = 0, max_depth: int = 4) -> Dict:
        if depth >= max_depth:
            return {"type": "atomic", "description": goal}
        return {
            "type": "composite",
            "description": goal,
            "children": [
                self.decompose_goal(sub_goal, depth + 1, max_depth)
                for sub_goal in self._split_goal(goal)
            ],
        }

    def _split_goal(self, goal: str) -> List[str]:
        if " and " in goal.lower():
            parts = goal.split(" and ")
        elif " then " in goal.lower():
            parts = goal.split(" then ")
        else:
            return [goal]
        return [p.strip() for p in parts if p.strip()]

    def build_dependency_graph(self, tasks: List[Dict]) -> Dict[str, List[str]]:
        graph = {}
        for i, task in enumerate(tasks):
            deps = []
            task_name = task.get("name", f"task_{i}")
            for j in range(i):
                prev_name = tasks[j].get("name", f"task_{j}")
                if self._depends_on(task, tasks[j]):
                    deps.append(prev_name)
            graph[task_name] = deps
        return graph

    def _depends_on(self, task: Dict, potential_dep: Dict) -> bool:
        task_targets = set(task.get("targets", []))
        dep_targets = set(potential_dep.get("outputs", []))
        return len(task_targets & dep_targets) > 0

    def validate_plan(self, plan: Dict) -> Dict:
        errors = []
        warnings = []
        dep_graph = plan.get("dependency_graph", {})
        all_tasks = set(dep_graph.keys())
        for task, deps in dep_graph.items():
            for dep in deps:
                if dep not in all_tasks:
                    errors.append(f"Task '{task}' depends on unknown task '{dep}'")
        visited = set()
        path = []

        def dfs(node):
            if node in path:
                cycle_start = path[path.index(node):]
                errors.append(f"Circular dependency detected: {' → '.join(cycle_start + [node])}")
                return
            if node in visited:
                return
            visited.add(node)
            path.append(node)
            for dep in dep_graph.get(node, []):
                dfs(dep)
            path.pop()

        for task in all_tasks:
            dfs(task)
        return {
            "valid": len(errors) == 0,
            "errors": errors,
            "warnings": warnings,
        }

class IntentClassifier:
    def __init__(self, confidence_threshold: float = 0.75):
        self.threshold = confidence_threshold
        self.categories = {
            "create": ["create", "generate", "write", "implement", "build", "add"],
            "modify": ["change", "update", "modify", "edit", "refactor", "fix"],
            "analyze": ["analyze", "review", "check", "inspect", "audit", "debug"],
            "explain": ["explain", "describe", "document", "summarize", "clarify"],
            "deploy": ["deploy", "release", "publish", "push", "ship", "launch"],
            "delete": ["delete", "remove", "destroy", "clean", "erase"],
        }

    def classify(self, user_request: str) -> Dict:
        request_lower = user_request.lower()
        scores = {}
        for category, keywords in self.categories.items():
            score = sum(1 for kw in keywords if kw in request_lower)
            scores[category] = score / max(len(request_lower.split()), 1)
        primary = max(scores, key=scores.get)
        return {
            "primary_class": primary,
            "confidence": scores[primary] * 5,
            "all_scores": scores,
            "needs_clarification": (scores[primary] * 5) < self.threshold,
        }

class ConstraintEngine:
    def __init__(self):
        self.hard_constraints: List[Dict] = []
        self.soft_constraints: List[Dict] = []

    def add_constraint(self, constraint: Dict, is_hard: bool = True):
        target = self.hard_constraints if is_hard else self.soft_constraints
        target.append(constraint)

    def propagate(self, plan_tree: Dict) -> Dict:
        violations = []
        self._check_node(plan_tree, violations, [])
        return {
            "all_satisfied": len(violations) == 0,
            "hard_violations": [v for v in violations if v["is_hard"]],
            "soft_violations": [v for v in violations if not v["is_hard"]],
        }

    def _check_node(self, node: Dict, violations: List, path: List[str]):
        current_path = path + [node.get("description", "unknown")]
        for constraint in self.hard_constraints:
            if not self._satisfies(node, constraint):
                violations.append({
                    "path": current_path,
                    "constraint": constraint["name"],
                    "is_hard": True,
                })
        for child in node.get("children", []):
            self._check_node(child, violations, current_path)

    def _satisfies(self, node: Dict, constraint: Dict) -> bool:
        node_props = set(node.get("properties", {}).keys())
        required = set(constraint.get("requires", []))
        return required.issubset(node_props)

class PreflightValidator:
    def __init__(self):
        self.checks: List[Callable] = []

    def add_check(self, check_fn: Callable, name: str):
        self.checks.append((check_fn, name))

    def run_all(self, plan: Dict) -> Dict:
        results = {}
        all_passed = True
        for check_fn, name in self.checks:
            try:
                result = check_fn(plan)
                passed = result if isinstance(result, bool) else result.get("passed", False)
                results[name] = {"passed": passed, "error": None}
                all_passed = all_passed and passed
            except Exception as e:
                results[name] = {"passed": False, "error": str(e)}
                all_passed = False
        return {"passed": all_passed, "check_results": results}

Architecture Decision Trees

Feedforward Control Strategy

Is the task complex enough to warrant planning?
├── Single atomic step
│   └── Yes → Direct execution with pre-flight validation
│
├── 2-8 steps with clear ordering
│   └── Yes → Plan-and-Execute with dependency graph
│       ├── Validate DAG for cycles
│       └── Execute in topological order
│
├── 9+ steps or multi-domain
│   └── Yes → Hierarchical goal decomposition
│       ├── Break into phases (max 4 levels deep)
│       ├── Propagate constraints between phases
│       └── Add checkpoints between phases
│
└── Task has high failure cost
    └── Full OODA loop
        ├── Observe: gather environment state
        ├── Orient: classify intent, assess risk
        ├── Decide: build and validate plan
        └── Act: execute with monitoring

Pre-Flight Check Selection

What type of execution?
├── File modification
│   ├── File exists? → Continue
│   ├── File is writable? → Continue
│   ├── File is not locked? → Continue
│   └── Backup exists? → Continue
│
├── API call
│   ├── Endpoint is reachable? → Continue
│   ├── Rate limit not exceeded? → Continue
│   ├── Auth token valid? → Continue
│   └── Request schema valid? → Continue
│
├── Database operation
│   ├── Connection works? → Continue
│   ├── Migration state compatible? → Continue
│   └── Transaction safe (no long locks)? → Continue
│
└── Deployment
    ├── All tests pass? → Continue
    ├── No secrets in code? → Continue
    ├── Version bumped? → Continue
    └── Rollback plan exists? → Continue

Production Considerations

• Plan serialization for audit: Serialize every plan artifact with version, timestamp, and agent ID. Store for post-hoc analysis of planning failures.
• Adaptive decomposition depth: Start with shallow decomposition (2 levels). If execution fails due to underspecified steps, increase depth on the next attempt.
• Constraint caching: Cache constraint satisfaction results for sub-plans that are reused across tasks. Invalidates on environment state change.
• Plan cost estimation: Maintain historical cost data (token usage, API calls, execution time) per task type. Use to pre-allocate budgets and detect anomalous patterns early.

Security Considerations

• Intent classification bypass: Malicious requests may disguise intent (e.g., "analyze" that is actually "delete"). Use secondary classification and constraint overlap checking.
• Plan injection: Never serialize plans from untrusted sources without validation. A crafted plan could encode dangerous operations that look benign in structure.
• Pre-flight validation as security gate: Treat pre-flight as a security boundary. Operations that fail validation should be logged with full context for security auditing.

Anti-Patterns

Performance Optimization

• Parallel goal decomposition: Decompose independent sub-goals in parallel using async task trees. Merges results at constraint propagation stage.
• Plan caching: Cache plan structures for identical or similar task requests. Hash by task description + file context. 30-50% of coding tasks have similar structure.
• Lazy constraint evaluation: Only evaluate constraints for tasks on the critical path first. Defer non-critical constraint checks to execution time.
• Constraint compilation: Compile constraint expressions into predicate functions at plan time rather than evaluating trees at runtime. Provides 10-100x faster constraint checking.