purpose: Comprehensive guide to closed-loop system architecture in agentic coding
type: technical-documentation
course: Tactical Agentic Coding
lesson: 05-supplemental
difficulty: INTERMEDIATE
dependencies: lesson-05-close-the-loops
last-updated: 2025-09-30
Closed-loop architecture is the foundational pattern for autonomous agent systems. This document provides complete architectural patterns, implementation strategies, and failure modes for building self-correcting feedback systems that enable agents to execute, validate, and correct their work without human intervention.
# CLOSED-LOOP ARCHITECTURE
## WHAT IS A CLOSED-LOOP SYSTEM?
### Definition
**CLOSED-LOOP SYSTEM:**
A system where output is continuously measured and fed back into the system to control and optimize behavior without external intervention.
**IN AGENTIC CODING:**
Agents execute tasks, validate results automatically, analyze failures, correct errors, and repeat until success criteria are met—all without human involvement.
### Core Components
**PURPOSE:** Perform the actual work
**RESPONSIBILITIES:**
- Write code
- Make changes
- Generate artifacts
- Execute commands
**OUTPUT:** Work product (code, files, changes)
**PURPOSE:** Assess quality of work product
**RESPONSIBILITIES:**
- Run tests
- Execute linters
- Check builds
- Verify outputs
**OUTPUT:** Validation results (pass/fail + details)
**PURPOSE:** Understand validation failures
**RESPONSIBILITIES:**
- Parse error messages
- Identify root causes
- Contextualize failures
- Prioritize issues
**OUTPUT:** Analysis report (what failed, why, how to fix)
**PURPOSE:** Generate fixes for failures
**RESPONSIBILITIES:**
- Apply fixes based on analysis
- Retry with new context
- Adapt strategy if needed
- Learn from patterns
**OUTPUT:** Updated work product
**PURPOSE:** Orchestrate the loop
**RESPONSIBILITIES:**
- Manage loop iterations
- Track progress
- Enforce timeouts
- Escalate when needed
**OUTPUT:** Final result or escalation signal
---
## EXECUTE → VALIDATE → CORRECT → REPEAT CYCLE
### The Fundamental Loop
```
START
↓
┌─────────────────────┐
│ 1. EXECUTE │ Agent performs task
│ Generate work │
└──────────┬──────────┘
↓
┌─────────────────────┐
│ 2. VALIDATE │ Run automated checks
│ Test the work │
└──────────┬──────────┘
↓
┌────────┐
│Success?│
└───┬────┘
│
┌─────┴─────┐
│ │
YES NO
│ │
▼ ▼
┌────────┐ ┌─────────────────────┐
│ DONE │ │ 3. ANALYZE │ Understand failures
└────────┘ │ Parse errors │
└──────────┬──────────┘
↓
┌─────────────────────┐
│ 4. CORRECT │ Generate fixes
│ Apply changes │
└──────────┬──────────┘
│
└──────┐
│
(Back to EXECUTE)
```
### Detailed Flow
**STEP 1: EXECUTE**
```python
def execute_phase(task, context):
"""Generate work product based on task and context."""
work_product = agent.execute(
task=task,
context=context,
constraints=task.constraints
)
return work_product
```
**STEP 2: VALIDATE**
```python
def validate_phase(work_product, validations):
"""Run all configured validations."""
results = []
for validation in validations:
result = validation.run(work_product)
results.append(result)
return ValidationSummary(results)
```
**STEP 3: ANALYZE**
```python
def analyze_phase(validation_summary):
"""Parse failures and generate actionable context."""
if validation_summary.all_passed:
return None # No analysis needed
failures = validation_summary.failures
analysis = agent.analyze_errors(
errors=failures,
patterns=known_error_patterns
)
return analysis
```
**STEP 4: CORRECT**
```python
def correct_phase(work_product, analysis):
"""Apply fixes based on analysis."""
fixes = agent.generate_fixes(
original=work_product,
analysis=analysis
)
corrected_product = agent.apply_fixes(
work_product=work_product,
fixes=fixes
)
return corrected_product
```
**STEP 5: REPEAT**
```python
def closed_loop_execution(task, max_iterations=5):
"""Complete closed-loop cycle."""
context = task.initial_context
iteration = 0
while iteration < max_iterations:
# Execute
work_product = execute_phase(task, context)
# Validate
validation_summary = validate_phase(
work_product,
task.validations
)
# Check success
if validation_summary.all_passed:
return work_product # Success!
# Analyze failures
analysis = analyze_phase(validation_summary)
# Update context for next iteration
context = context.with_analysis(analysis)
# Correct (happens in next execute phase)
iteration += 1
# Max iterations reached
raise LoopExhaustionError(
f"Failed after {max_iterations} attempts",
last_analysis=analysis
)
```
---
## ARCHITECTURE PATTERNS FOR CLOSED LOOPS
### Pattern 1: Single Validation Loop
**USE CASE:** Simple tasks with one validation type
**ARCHITECTURE:**
```
Task → Execute → Validate → Pass? → Done
↑ │
└─────Fix──────────┘
```
**EXAMPLE:**
```python
class SingleValidationLoop:
def __init__(self, task, validator):
self.task = task
self.validator = validator
def execute(self):
attempts = 0
while attempts < 5:
result = self.agent.execute(self.task)
validation = self.validator.validate(result)
if validation.passed:
return result
self.task = self.task.with_errors(validation.errors)
attempts += 1
raise Exception("Validation failed after 5 attempts")
```
**PROS:** Simple, fast, easy to understand
**CONS:** Limited to single validation type
### Pattern 2: Sequential Validation Pipeline
**USE CASE:** Multiple validations that must run in order
**ARCHITECTURE:**
```
Execute → Lint → Unit → Integration → Build → Done
↑ │ │ │ │
└────────┴──────┴─────────┴──────────┘
(Retry from start on any failure)
```
**EXAMPLE:**
```python
class SequentialPipeline:
def __init__(self, task, validations):
self.task = task
self.validations = validations # Ordered list
def execute(self):
attempts = 0
while attempts < 5:
result = self.agent.execute(self.task)
# Run validations in sequence
for validation in self.validations:
val_result = validation.run(result)
if not val_result.passed:
# Fail fast - restart from beginning
self.task = self.task.with_errors(val_result.errors)
break
else:
# All validations passed
return result
attempts += 1
raise Exception("Pipeline failed after 5 attempts")
```
**PROS:** Fast failure, clear progression
**CONS:** Must restart from beginning on any failure
### Pattern 3: Parallel Validation with Aggregation
**USE CASE:** Independent validations that can run simultaneously
**ARCHITECTURE:**
```
┌─→ Lint ──┐
│ │
Execute → Distribute├─→ Test ──┤→ Aggregate → All Pass? → Done
↑ │ │ │
│ └─→ Build─┘ │
└────────────────────────────────────────────┘
(Retry with all errors)
```
**EXAMPLE:**
```python
class ParallelValidation:
def __init__(self, task, validations):
self.task = task
self.validations = validations
async def execute(self):
attempts = 0
while attempts < 5:
result = self.agent.execute(self.task)
# Run all validations in parallel
validation_results = await asyncio.gather(*[
validation.run_async(result)
for validation in self.validations
])
# Aggregate results
all_passed = all(v.passed for v in validation_results)
if all_passed:
return result
# Collect all errors for next attempt
all_errors = [
error
for val_result in validation_results
for error in val_result.errors
]
self.task = self.task.with_errors(all_errors)
attempts += 1
raise Exception("Parallel validation failed")
```
**PROS:** Fast (parallel execution), comprehensive feedback
**CONS:** More complex, may produce overwhelming error lists
### Pattern 4: Nested Loop Hierarchy
**USE CASE:** Validations at different scopes (function → module → system)
**ARCHITECTURE:**
```
Outer Loop (System-level validation)
│
└─→ Middle Loop (Module-level validation)
│
└─→ Inner Loop (Function-level validation)
│
└─→ Execute & Validate
```
**EXAMPLE:**
```python
class NestedLoopSystem:
def execute_function_loop(self, func_task):
"""Inner loop: single function validation"""
for attempt in range(3):
code = self.agent.write_function(func_task)
if self.validate_syntax(code):
return code
raise Exception("Function validation failed")
def execute_module_loop(self, module_task):
"""Middle loop: module validation"""
for attempt in range(3):
functions = [
self.execute_function_loop(func)
for func in module_task.functions
]
module = self.combine_functions(functions)
if self.validate_module(module):
return module
raise Exception("Module validation failed")
def execute_system_loop(self, system_task):
"""Outer loop: system validation"""
for attempt in range(3):
modules = [
self.execute_module_loop(mod)
for mod in system_task.modules
]
system = self.combine_modules(modules)
if self.validate_system(system):
return system
raise Exception("System validation failed")
```
**PROS:** Granular validation, early failure detection
**CONS:** Complex, slower due to nested loops
### Pattern 5: Adaptive Loop with Strategy Selection
**USE CASE:** Complex tasks where validation strategy changes based on results
**ARCHITECTURE:**
```
Execute → Validate → Analyze Results
│ │
│ ▼
│ Select Strategy
│ │
│ ┌────┴────┐
│ │ │
│ Fast Path Slow Path
│ │ │
└─────────┴─────────┘
(Adaptive retry)
```
**EXAMPLE:**
```python
class AdaptiveLoop:
def __init__(self, task):
self.task = task
self.strategy = "fast" # Start optimistic
def execute(self):
attempts = 0
while attempts < 10: # More attempts due to adaptation
result = self.agent.execute(self.task, strategy=self.strategy)
validation = self.validate(result)
if validation.passed:
return result
# Adapt strategy based on failure type
self.strategy = self.select_strategy(validation.errors)
self.task = self.task.with_errors(validation.errors)
attempts += 1
def select_strategy(self, errors):
"""Choose strategy based on error patterns."""
if any("syntax" in str(e) for e in errors):
return "careful" # Slower, more careful
elif any("timeout" in str(e) for e in errors):
return "fast" # Faster, risk more errors
elif any("memory" in str(e) for e in errors):
return "minimal" # Minimize resource usage
else:
return "balanced"
```
**PROS:** Intelligent adaptation, handles diverse failures
**CONS:** Most complex, requires sophisticated error analysis
---
## BREAKING OPEN LOOPS (COMMON FAILURE MODE)
### What Makes a Loop "Open"
**CHARACTERISTICS OF OPEN LOOPS:**
1. **Human in the loop**: Requires manual intervention
2. **No automatic feedback**: Agent doesn't see validation results
3. **Manual error reporting**: Human must describe failures
4. **Context loss**: Agent doesn't maintain state between attempts
5. **No retry logic**: Failure requires manual restart
**RESULT:** Bottleneck, slow execution, no scalability
### Common Ways Loops Break Open
**SYMPTOM:** Validation exists but isn't automated
**EXAMPLE:**
```python
# BAD: Open loop
result = agent.execute(task)
# Human manually runs: npm test
# Human manually reports results back
```
**FIX:**
```python
# GOOD: Closed loop
result = agent.execute(task)
test_result = run_command("npm test")
if not test_result.success:
agent.retry_with_errors(test_result.errors)
```
**SYMPTOM:** Agent retries without understanding what failed
**EXAMPLE:**
```python
# BAD: No context
for attempt in range(5):
result = agent.execute(task) # Same input every time
if validate(result):
break
```
**FIX:**
```python
# GOOD: Context accumulation
context = task.context
for attempt in range(5):
result = agent.execute(task, context)
validation = validate(result)
if validation.passed:
break
context = context.with_errors(validation.errors)
```
**SYMPTOM:** Execute without any validation
**EXAMPLE:**
```python
# BAD: No validation
result = agent.execute(task)
return result # Hope it works!
```
**FIX:**
```python
# GOOD: Always validate
result = agent.execute(task)
validation = validate(result)
if not validation.passed:
raise ValidationError(validation.errors)
return result
```
**SYMPTOM:** No exit condition or max attempts
**EXAMPLE:**
```python
# BAD: Infinite loop risk
while True:
result = agent.execute(task)
if validate(result):
break
# What if it never passes?
```
**FIX:**
```python
# GOOD: Max attempts and escalation
max_attempts = 5
for attempt in range(max_attempts):
result = agent.execute(task)
if validate(result):
return result
raise LoopExhaustionError(
"Failed after max attempts",
last_result=result
)
```
**SYMPTOM:** Loop interrupted by external events
**EXAMPLE:**
```python
# BAD: External interruption
result = agent.execute(task)
# Agent waits for human approval here
# Loop broken - loses state
if human_approves:
continue_execution()
```
**FIX:**
```python
# GOOD: Autonomous until complete
result = complete_closed_loop(task)
# Only then involve human for review
if human_wants_review:
show_review(result)
```
---
## LOOP NESTING AND CASCADING
### Nested Loop Strategies
**WHY NEST LOOPS?**
Different validation scopes require different loop levels:
- **Inner loops:** Fast, frequent, small scope (function-level)
- **Middle loops:** Medium speed, module-level
- **Outer loops:** Slow, comprehensive, system-level
**BENEFIT:** Fail fast at inner loops, validate comprehensively at outer loops
### Cascade Pattern
**VALIDATION CASCADE:**
```
Level 1: Syntax validation (milliseconds)
│
├─ Pass → Level 2: Type checking (seconds)
│ │
│ ├─ Pass → Level 3: Unit tests (seconds)
│ │ │
│ │ ├─ Pass → Level 4: Integration (minutes)
│ │ │ │
│ │ │ └─ Pass → Done
│ │ │
│ │ └─ Fail → Return to Level 1
│ │
│ └─ Fail → Return to Level 1
│
└─ Fail → Return to Level 1 (immediately)
```
**RULE:** Failure at any level returns to Level 1 (cheapest validation)
### Implementation
```python
class CascadingValidation:
def __init__(self, task):
self.task = task
self.levels = [
("syntax", self.validate_syntax, "instant"),
("types", self.validate_types, "fast"),
("unit", self.validate_unit_tests, "medium"),
("integration", self.validate_integration, "slow"),
]
def execute(self):
attempts = 0
max_attempts = 5
while attempts < max_attempts:
result = self.agent.execute(self.task)
# Cascade through validation levels
for level_name, validator, speed in self.levels:
validation_result = validator(result)
if not validation_result.passed:
# Fail fast - return to start
self.task = self.task.with_errors(
level=level_name,
errors=validation_result.errors
)
break # Break to outer while loop
else:
# All levels passed
return result
attempts += 1
raise CascadeFailureError(f"Failed after {attempts} attempts")
def validate_syntax(self, result):
"""Level 1: Instant syntax check"""
return run_linter(result)
def validate_types(self, result):
"""Level 2: Type checking"""
return run_type_checker(result)
def validate_unit_tests(self, result):
"""Level 3: Unit test execution"""
return run_unit_tests(result)
def validate_integration(self, result):
"""Level 4: Integration test execution"""
return run_integration_tests(result)
```
---
## TIMEOUT AND ESCAPE CONDITIONS
### Why Timeouts Matter
**WITHOUT TIMEOUTS:**
- Infinite loops on impossible tasks
- Resource exhaustion
- Blocked pipelines
- Wasted compute
**WITH TIMEOUTS:**
- Controlled failure
- Resource preservation
- Clear escalation paths
- Predictable behavior
### Timeout Strategies
**APPROACH:** Maximum number of loop iterations
**EXAMPLE:**
```python
def execute_with_iteration_limit(task, max_iterations=5):
for iteration in range(max_iterations):
result = agent.execute(task)
if validate(result):
return result
task = task.with_previous_attempt(result)
raise MaxIterationsError(
f"Failed after {max_iterations} attempts"
)
```
**PROS:** Simple, predictable
**CONS:** Doesn't account for varying task complexity
**APPROACH:** Maximum elapsed time
**EXAMPLE:**
```python
import time
def execute_with_time_limit(task, max_seconds=300):
start_time = time.time()
while time.time() - start_time < max_seconds:
result = agent.execute(task)
if validate(result):
return result
task = task.with_previous_attempt(result)
raise TimeoutError(
f"Failed to complete in {max_seconds} seconds"
)
```
**PROS:** Accounts for task complexity
**CONS:** May terminate during critical operations
**APPROACH:** Timeout adjusts based on progress
**EXAMPLE:**
```python
def execute_with_adaptive_timeout(task):
max_iterations = 5
base_timeout = 60 # seconds per iteration
for iteration in range(max_iterations):
# Increase timeout if making progress
if iteration > 0 and showing_progress(task):
timeout = base_timeout * (iteration + 1)
else:
timeout = base_timeout
result = agent.execute_with_timeout(task, timeout)
if validate(result):
return result
task = task.with_previous_attempt(result)
raise AdaptiveTimeoutError("No progress after max iterations")
def showing_progress(task):
"""Check if validation errors are decreasing"""
if len(task.history) < 2:
return False
prev_errors = len(task.history[-2].errors)
curr_errors = len(task.history[-1].errors)
return curr_errors < prev_errors
```
**PROS:** Intelligent, rewards progress
**CONS:** Most complex to implement
### Escape Conditions
**WHEN TO BREAK THE LOOP:**
1. **Success:** All validations pass
```python
if validation.all_passed:
return result
```
2. **Max Iterations:** Exhausted retry attempts
```python
if iteration >= max_iterations:
raise MaxIterationsError()
```
3. **No Progress:** Same errors repeatedly
```python
if errors_unchanged_for_n_iterations(3):
raise NoProgressError()
```
4. **Degradation:** Getting worse, not better
```python
if error_count_increasing():
raise DegradationError()
```
5. **External Signal:** User cancellation
```python
if user_cancelled:
raise CancellationError()
```
6. **Resource Exhaustion:** Out of memory/time/compute
```python
if resources_exhausted():
raise ResourceError()
```
### Escalation Strategy
**WHEN LOOPS FAIL, ESCALATE:**
```python
class LoopEscalation:
def execute(self, task):
try:
return self.closed_loop(task)
except MaxIterationsError as e:
return self.escalate_to_human(task, e)
except NoProgressError as e:
return self.escalate_to_different_agent(task, e)
except DegradationError as e:
return self.escalate_to_simplified_task(task, e)
def escalate_to_human(self, task, error):
"""Human intervention required"""
notification.send(
"Agent unable to complete task",
details=error.details,
task=task
)
return await_human_input()
def escalate_to_different_agent(self, task, error):
"""Try different agent/model"""
return specialized_agent.execute(task)
def escalate_to_simplified_task(self, task, error):
"""Break task into smaller pieces"""
subtasks = task.decompose()
return [self.execute(st) for st in subtasks]
```
---
## SUMMARY
**CLOSED-LOOP ARCHITECTURE ESSENTIALS:**
**COMPONENTS:**
- Executor: Performs work
- Validator: Checks quality
- Analyzer: Understands failures
- Corrector: Applies fixes
- Controller: Orchestrates loop
**CYCLE:**
Execute → Validate → Analyze → Correct → Repeat
**PATTERNS:**
1. Single validation loop (simple)
2. Sequential pipeline (ordered validations)
3. Parallel validation (speed)
4. Nested loops (granular scope)
5. Adaptive loops (intelligent strategy)
**FAILURE MODES:**
- Missing automation
- No error context
- Missing validation
- Infinite loops
- Context switching
**ESCAPE CONDITIONS:**
- Success (all validations pass)
- Max iterations
- No progress
- Degradation
- External signal
- Resource exhaustion
**KEY INSIGHT:**
Closed loops transform agents from code generators into autonomous engineers that self-validate and self-correct until production-ready.