
Goal Seeking Agent Pattern
- 148 installs
- 70 repo stars
- Updated July 26, 2026
- rysweet/amplihack
Implement goal-seeking agent loops that plan, act, observe outcomes, and iterate until measurable objectives are met within guardrails.
About
Documents and implements the goal-seeking agent pattern: decompose a target outcome, select tools, execute steps, evaluate progress, and loop until success or limits. Provides a reusable architecture for autonomous Claude agents with explicit success criteria, failure handling, and iteration control instead of unstructured multi-turn improvisation.
- Objective decomposition
- Observe-plan-act loops
- Tool routing and evaluation
- Termination and guardrails
- Reusable autonomous agent template
Goal Seeking Agent Pattern by the numbers
- 148 all-time installs (skills.sh)
- +1 installs in the week ending Jul 26, 2026 (Skillselion tracking)
- Ranked #3,348 of 16,556 AI & Agent Building skills by installs in the Skillselion catalog
- Data as of Aug 2, 2026 (Skillselion catalog sync)
npx skills add https://github.com/rysweet/amplihack --skill goal-seeking-agent-patternAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 148 |
|---|---|
| repo stars | ★ 70 |
| Last updated | July 26, 2026 |
| Repository | rysweet/amplihack ↗ |
What it does
Implement goal-seeking agent loops that plan, act, observe outcomes, and iterate until measurable objectives are met within guardrails.
Files
Goal-Seeking Agent Pattern Skill
1. What Are Goal-Seeking Agents?
Goal-seeking agents are autonomous AI agents that execute multi-phase objectives by:
1. Understanding High-Level Goals: Accept natural language objectives without explicit step-by-step instructions 2. Planning Execution: Break goals into phases with dependencies and success criteria 3. Autonomous Execution: Make decisions and adapt behavior based on intermediate results 4. Self-Assessment: Evaluate progress against success criteria and adjust approach 5. Resilient Operation: Handle failures gracefully and explore alternative solutions
Core Characteristics
Autonomy: Agents decide HOW to achieve goals, not just follow prescriptive steps
Adaptability: Adjust strategy based on runtime conditions and intermediate results
Goal-Oriented: Focus on outcomes (what to achieve) rather than procedures (how to achieve)
Multi-Phase: Complex objectives decomposed into manageable phases with dependencies
Self-Monitoring: Track progress, detect failures, and course-correct autonomously
Distinction from Traditional Agents
| Traditional Agent | Goal-Seeking Agent |
|---|---|
| Follows fixed workflow | Adapts workflow to context |
| Prescriptive steps | Outcome-oriented objectives |
| Human intervention on failure | Autonomous recovery attempts |
| Single-phase execution | Multi-phase with dependencies |
| Rigid decision tree | Dynamic strategy adjustment |
When Goal-Seeking Makes Sense
Goal-seeking agents excel when:
- Problem space is large: Many possible paths to success
- Context varies: Runtime conditions affect optimal approach
- Failures are expected: Need autonomous recovery without human intervention
- Objectives are clear: Success criteria well-defined but path is flexible
- Multi-step complexity: Requires coordination across phases with dependencies
When to Avoid Goal-Seeking
Use traditional agents or scripts when:
- Single deterministic path: Only one way to achieve goal
- Latency-critical: Need fastest possible execution (no decision overhead)
- Safety-critical: Human verification required at each step
- Simple workflow: Complexity of goal-seeking exceeds benefit
- Audit requirements: Need deterministic, reproducible execution
2. When to Use This Pattern
Problem Indicators
Use goal-seeking agents when you observe these patterns:
Pattern 1: Workflow Variability
Indicators:
- Same objective requires different approaches based on context
- Manual decisions needed at multiple points
- "It depends" answers when mapping workflow
Example: Release workflow that varies by:
- Environment (staging vs production)
- Change type (hotfix vs feature)
- Current system state (healthy vs degraded)
Solution: Goal-seeking agent evaluates context and adapts workflow
Pattern 2: Multi-Phase Complexity
Indicators:
- Objective requires 3-5+ distinct phases
- Phases have dependencies (output of phase N feeds phase N+1)
- Parallel execution opportunities exist
- Success criteria differ per phase
Example: Data pipeline with phases:
1. Data collection (multiple sources, parallel) 2. Transformation (depends on collection results) 3. Validation (depends on transformation output) 4. Publishing (conditional on validation pass)
Solution: Goal-seeking agent orchestrates phases, handles dependencies
Pattern 3: Autonomous Recovery Needed
Indicators:
- Failures are expected and recoverable
- Multiple retry/fallback strategies exist
- Human intervention is expensive or slow
- Can verify success programmatically
Example: CI diagnostic workflow:
- Test failures (retry with different approach)
- Environment issues (reconfigure and retry)
- Dependency conflicts (resolve and rerun)
Solution: Goal-seeking agent tries strategies until success or escalation
Pattern 4: Adaptive Decision Making
Indicators:
- Need to evaluate trade-offs at runtime
- Multiple valid solutions with different characteristics
- Optimization objectives (speed vs quality vs cost)
- Context-dependent best practices
Example: Fix agent pattern matching:
- QUICK mode for obvious issues
- DIAGNOSTIC mode for unclear problems
- COMPREHENSIVE mode for complex solutions
Solution: Goal-seeking agent selects strategy based on problem analysis
Pattern 5: Domain Expertise Required
Indicators:
- Requires specialized knowledge to execute
- Multiple domain-specific tools/approaches
- Best practices vary by domain
- Coordination of specialized sub-agents
Example: AKS SRE automation:
- Azure-specific operations (ARM, CLI)
- Kubernetes expertise (kubectl, YAML)
- Networking knowledge (CNI, ingress)
- Security practices (RBAC, Key Vault)
Solution: Goal-seeking agent with domain expertise coordinates specialized actions
Decision Framework
Use this 5-question framework to evaluate goal-seeking applicability:
Question 1: Is the objective well-defined but path flexible?
YES if:
- Clear success criteria exist
- Multiple valid approaches
- Runtime context affects optimal path
NO if:
- Only one correct approach
- Path is deterministic
- Success criteria ambiguous
Example YES: "Ensure AKS cluster is production-ready" (many paths, clear criteria) Example NO: "Run specific kubectl command" (one path, prescriptive)
Question 2: Are there multiple phases with dependencies?
YES if:
- Objective naturally decomposes into 3-5+ phases
- Phase outputs feed subsequent phases
- Some phases can execute in parallel
- Failures in one phase affect downstream phases
NO if:
- Single-phase execution sufficient
- No inter-phase dependencies
- Purely sequential with no branching
Example YES: Data pipeline (collect → transform → validate → publish) Example NO: Format code with ruff (single atomic operation)
Question 3: Is autonomous recovery valuable?
YES if:
- Failures are common and expected
- Multiple recovery strategies exist
- Human intervention is expensive/slow
- Can verify success automatically
NO if:
- Failures are rare edge cases
- Manual investigation always required
- Safety-critical (human verification needed)
- Cannot verify success programmatically
Example YES: CI diagnostic workflow (try multiple fix strategies) Example NO: Deploy to production (human approval required)
Question 4: Does context significantly affect approach?
YES if:
- Environment differences change strategy
- Current system state affects decisions
- Trade-offs vary by situation (speed vs quality vs cost)
- Domain-specific best practices apply
NO if:
- Same approach works for all contexts
- No environmental dependencies
- No trade-off decisions needed
Example YES: Fix agent (quick vs diagnostic vs comprehensive based on issue) Example NO: Generate UUID (context-independent)
Question 5: Is the complexity justified?
YES if:
- Problem is repeated frequently (2+ times/week)
- Manual execution takes 30+ minutes
- High value from automation
- Maintenance cost is acceptable
NO if:
- One-off or rare problem
- Quick manual execution (< 5 minutes)
- Simple script suffices
- Maintenance cost exceeds benefit
Example YES: CI failure diagnosis (frequent, time-consuming, high value) Example NO: One-time data migration (rare, script sufficient)
Decision Matrix
| All 5 YES | Use Goal-Seeking Agent | | 4 YES, 1 NO | Probably use Goal-Seeking Agent | | 3 YES, 2 NO | Consider simpler agent or hybrid | | 2 YES, 3 NO | Traditional agent likely better | | 0-1 YES | Script or simple automation |
3. Architecture Pattern
Component Architecture
Goal-seeking agents have four core components:
# Component 1: Goal Definition
class GoalDefinition:
"""Structured representation of objective"""
raw_prompt: str # Natural language goal
goal: str # Extracted primary objective
domain: str # Problem domain (security, data, automation, etc.)
constraints: list[str] # Technical/operational constraints
success_criteria: list[str] # How to verify success
complexity: str # simple, moderate, complex
context: dict # Additional metadata
# Component 2: Execution Plan
class ExecutionPlan:
"""Multi-phase plan with dependencies"""
goal_id: uuid.UUID
phases: list[PlanPhase]
total_estimated_duration: str
required_skills: list[str]
parallel_opportunities: list[list[str]] # Phases that can run parallel
risk_factors: list[str]
# Component 3: Plan Phase
class PlanPhase:
"""Individual phase in execution plan"""
name: str
description: str
required_capabilities: list[str]
estimated_duration: str
dependencies: list[str] # Names of prerequisite phases
parallel_safe: bool # Can execute in parallel
success_indicators: list[str] # How to verify phase completion
# Component 4: Skill Definition
class SkillDefinition:
"""Capability needed for execution"""
name: str
description: str
capabilities: list[str]
implementation_type: str # "native" or "delegated"
delegation_target: str # Agent to delegate toExecution Flow
┌─────────────────────────────────────────────────────────────┐
│ 1. GOAL ANALYSIS │
│ │
│ Input: Natural language objective │
│ Process: Extract goal, domain, constraints, criteria │
│ Output: GoalDefinition │
│ │
│ [PromptAnalyzer.analyze_text(prompt)] │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 2. PLANNING │
│ │
│ Input: GoalDefinition │
│ Process: Decompose into phases, identify dependencies │
│ Output: ExecutionPlan │
│ │
│ [ObjectivePlanner.generate_plan(goal_definition)] │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 3. SKILL SYNTHESIS │
│ │
│ Input: ExecutionPlan │
│ Process: Map capabilities to skills, identify agents │
│ Output: list[SkillDefinition] │
│ │
│ [SkillSynthesizer.synthesize(execution_plan)] │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 4. AGENT ASSEMBLY │
│ │
│ Input: GoalDefinition, ExecutionPlan, Skills │
│ Process: Combine into executable bundle │
│ Output: GoalAgentBundle │
│ │
│ [AgentAssembler.assemble(goal, plan, skills)] │
└─────────────────────────────────────────────────────────────┘
↓
┌─────────────────────────────────────────────────────────────┐
│ 5. EXECUTION (Auto-Mode) │
│ │
│ Input: GoalAgentBundle │
│ Process: Execute phases, monitor progress, adapt │
│ Output: Success or escalation │
│ │
│ [Auto-mode with initial_prompt from bundle] │
└─────────────────────────────────────────────────────────────┘Phase Dependency Management
Phases can have three relationship types:
Sequential Dependency: Phase B depends on Phase A completion
Phase A → Phase B → Phase CParallel Execution: Phases can run concurrently
Phase A ──┬→ Phase B ──┐
└→ Phase C ──┴→ Phase DConditional Branching: Phase selection based on results
Phase A → [Decision] → Phase B (success path)
└→ Phase C (recovery path)State Management
Goal-seeking agents maintain state across phases:
class AgentState:
"""Runtime state for goal-seeking agent"""
current_phase: str
completed_phases: list[str]
phase_results: dict[str, Any] # Output from each phase
failures: list[FailureRecord] # Track what didn't work
retry_count: int
total_duration: timedelta
context: dict # Shared context across phasesError Handling
Three error recovery strategies:
Retry with Backoff: Same approach, exponential delay
for attempt in range(MAX_RETRIES):
try:
result = execute_phase(phase)
break
except RetryableError as e:
wait_time = INITIAL_DELAY * (2 ** attempt)
sleep(wait_time)Alternative Strategy: Different approach to same goal
for strategy in STRATEGIES:
try:
result = execute_phase(phase, strategy)
break
except StrategyFailedError:
continue # Try next strategy
else:
escalate_to_human("All strategies exhausted")Graceful Degradation: Accept partial success
try:
result = execute_phase_optimal(phase)
except OptimalFailedError:
result = execute_phase_fallback(phase) # Lower quality but works4. Integration with goal_agent_generator
The goal_agent_generator module provides the implementation for goal-seeking agents. Here's how to integrate:
Core API
from amplihack.goal_agent_generator import (
PromptAnalyzer,
ObjectivePlanner,
SkillSynthesizer,
AgentAssembler,
GoalAgentPackager,
)
# Step 1: Analyze natural language goal
analyzer = PromptAnalyzer()
goal_definition = analyzer.analyze_text("""
Automate AKS cluster production readiness verification.
Check security, networking, monitoring, and compliance.
Generate report with actionable recommendations.
""")
# Step 2: Generate execution plan
planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_definition)
# Step 3: Synthesize required skills
synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)
# Step 4: Assemble complete agent
assembler = AgentAssembler()
agent_bundle = assembler.assemble(
goal_definition=goal_definition,
execution_plan=execution_plan,
skills=skills,
bundle_name="aks-readiness-checker"
)
# Step 5: Package for deployment
packager = GoalAgentPackager()
packager.package(
bundle=agent_bundle,
output_dir=".claude/agents/goal-driven/aks-readiness-checker"
)CLI Integration
# Generate agent from prompt file
amplihack goal-agent-generator create \
--prompt ./prompts/aks-readiness.md \
--output .claude/agents/goal-driven/aks-readiness-checker
# Generate agent from inline prompt
amplihack goal-agent-generator create \
--inline "Automate CI failure diagnosis and fix iteration" \
--output .claude/agents/goal-driven/ci-fixer
# List generated agents
amplihack goal-agent-generator list
# Test agent execution
amplihack goal-agent-generator test \
--agent-path .claude/agents/goal-driven/ci-fixer \
--dry-runPromptAnalyzer Details
Extracts structured information from natural language:
from amplihack.goal_agent_generator import PromptAnalyzer
from pathlib import Path
analyzer = PromptAnalyzer()
# From file
goal_def = analyzer.analyze(Path("./prompts/my-goal.md"))
# From text
goal_def = analyzer.analyze_text("Deploy and monitor microservices to AKS")
# GoalDefinition contains:
print(goal_def.goal) # "Deploy and monitor microservices to AKS"
print(goal_def.domain) # "deployment"
print(goal_def.constraints) # ["Zero downtime", "Rollback capability"]
print(goal_def.success_criteria) # ["All pods running", "Metrics visible"]
print(goal_def.complexity) # "moderate"
print(goal_def.context) # {"priority": "high", "scale": "medium"}Domain classification:
data-processing: Data transformation, analysis, ETLsecurity-analysis: Vulnerability scanning, auditsautomation: Workflow automation, schedulingtesting: Test generation, validationdeployment: Release, publishing, distributionmonitoring: Observability, alertingintegration: API connections, webhooksreporting: Dashboards, metrics, summaries
Complexity determination:
simple: Single-phase, < 50 words, basic operationsmoderate: 2-4 phases, 50-150 words, some coordinationcomplex: 5+ phases, > 150 words, sophisticated orchestration
ObjectivePlanner Details
Generates multi-phase execution plans:
from amplihack.goal_agent_generator import ObjectivePlanner
planner = ObjectivePlanner()
plan = planner.generate_plan(goal_definition)
# ExecutionPlan contains:
for i, phase in enumerate(plan.phases, 1):
print(f"Phase {i}: {phase.name}")
print(f" Description: {phase.description}")
print(f" Duration: {phase.estimated_duration}")
print(f" Capabilities: {', '.join(phase.required_capabilities)}")
print(f" Dependencies: {', '.join(phase.dependencies)}")
print(f" Parallel Safe: {phase.parallel_safe}")
print(f" Success Indicators: {phase.success_indicators}")
print(f"\nTotal Duration: {plan.total_estimated_duration}")
print(f"Required Skills: {', '.join(plan.required_skills)}")
print(f"Parallel Opportunities: {plan.parallel_opportunities}")
print(f"Risk Factors: {plan.risk_factors}")Phase templates by domain:
- data-processing: Collection → Transformation → Analysis → Reporting
- security-analysis: Reconnaissance → Vulnerability Detection → Risk Assessment → Reporting
- automation: Setup → Workflow Design → Execution → Validation
- testing: Test Planning → Implementation → Execution → Results Analysis
- deployment: Pre-deployment → Deployment → Verification → Post-deployment
- monitoring: Setup Monitors → Data Collection → Analysis → Alerting
SkillSynthesizer Details
Maps capabilities to skills:
from amplihack.goal_agent_generator import SkillSynthesizer
synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)
# list[SkillDefinition]
for skill in skills:
print(f"Skill: {skill.name}")
print(f" Description: {skill.description}")
print(f" Capabilities: {', '.join(skill.capabilities)}")
print(f" Type: {skill.implementation_type}")
if skill.implementation_type == "delegated":
print(f" Delegates to: {skill.delegation_target}")Capability mapping:
data-*→data-processorskillsecurity-*,vulnerability-*→security-analyzerskilltest-*→testerskilldeploy-*→deployerskillmonitor-*,alert-*→monitorskillreport-*,document-*→documenterskill
AgentAssembler Details
Combines components into executable bundle:
from amplihack.goal_agent_generator import AgentAssembler
assembler = AgentAssembler()
bundle = assembler.assemble(
goal_definition=goal_definition,
execution_plan=execution_plan,
skills=skills,
bundle_name="custom-agent" # Optional, auto-generated if omitted
)
# GoalAgentBundle contains:
print(bundle.id) # UUID
print(bundle.name) # "custom-agent" or auto-generated
print(bundle.version) # "1.0.0"
print(bundle.status) # "ready"
print(bundle.auto_mode_config) # Configuration for auto-mode execution
print(bundle.metadata) # Domain, complexity, skills, etc.
# Auto-mode configuration
config = bundle.auto_mode_config
print(config["max_turns"]) # Based on complexity
print(config["initial_prompt"]) # Generated execution prompt
print(config["success_criteria"]) # From goal definition
print(config["constraints"]) # From goal definitionAuto-mode configuration:
max_turns: 5 (simple), 10 (moderate), 15 (complex), +20% per extra phaseinitial_prompt: Full markdown prompt with goal, plan, success criteriaworking_dir: Current directorysdk: "claude" (default)ui_mode: False (headless by default)
GoalAgentPackager Details
Packages bundle for deployment:
from amplihack.goal_agent_generator import GoalAgentPackager
from pathlib import Path
packager = GoalAgentPackager()
packager.package(
bundle=agent_bundle,
output_dir=Path(".claude/agents/goal-driven/my-agent")
)
# Creates:
# .claude/agents/goal-driven/my-agent/
# ├── agent.md # Agent definition
# ├── prompt.md # Initial prompt
# ├── metadata.json # Bundle metadata
# ├── plan.yaml # Execution plan
# └── skills.yaml # Required skills5. Recent Amplihack Examples
Real goal-seeking agents from the amplihack project:
Example 1: AKS SRE Automation (Issue #1293)
Problem: Manual AKS cluster operations are time-consuming and error-prone
Goal-Seeking Solution:
# Goal: Automate AKS production readiness verification
goal = """
Verify AKS cluster production readiness:
- Security: RBAC, network policies, Key Vault integration
- Networking: Ingress, DNS, load balancers
- Monitoring: Container Insights, alerts, dashboards
- Compliance: Azure Policy, resource quotas
Generate actionable report with recommendations.
"""
# Agent decomposes into phases:
# 1. Security Audit (parallel): RBAC check, network policies, Key Vault
# 2. Networking Validation (parallel): Ingress test, DNS resolution, LB health
# 3. Monitoring Verification (parallel): Metrics, logs, alerts configured
# 4. Compliance Check (depends on 1-3): Azure Policy, quotas, best practices
# 5. Report Generation (depends on 4): Markdown report with findings
# Agent adapts based on findings:
# - If security issues found: Suggest fixes, offer to apply
# - If monitoring missing: Generate alert templates
# - If compliance violations: List remediation stepsKey Characteristics:
- Autonomous: Checks multiple systems without step-by-step instructions
- Adaptive: Investigation depth varies by findings
- Multi-Phase: Parallel security/networking/monitoring, sequential reporting
- Domain Expert: Azure + Kubernetes knowledge embedded
- Self-Assessing: Validates each check, aggregates results
Implementation:
# Located in: .claude/agents/amplihack/specialized/azure-kubernetes-expert.md
# Uses knowledge base: .claude/data/azure_aks_expert/
# Integrates with goal_agent_generator:
from amplihack.goal_agent_generator import (
PromptAnalyzer, ObjectivePlanner, AgentAssembler
)
analyzer = PromptAnalyzer()
goal_def = analyzer.analyze_text(goal)
planner = ObjectivePlanner()
plan = planner.generate_plan(goal_def) # Generates 5-phase plan
# Domain-specific customization:
plan.phases[0].required_capabilities = [
"rbac-audit", "network-policy-check", "key-vault-integration"
]Lessons Learned:
- Domain expertise critical for complex infrastructure
- Parallel execution significantly reduces total time
- Actionable recommendations increase agent value
- Comprehensive knowledge base (Q&A format) enables autonomous decisions
Example 2: CI Diagnostic Workflow
Problem: CI failures require manual diagnosis and fix iteration
Goal-Seeking Solution:
# Goal: Diagnose CI failure and iterate fixes until success
goal = """
CI pipeline failing after push.
Diagnose failures, apply fixes, push updates, monitor CI.
Iterate until all checks pass.
Stop at mergeable state without auto-merging.
"""
# Agent decomposes into phases:
# 1. CI Status Monitoring: Check current CI state
# 2. Failure Diagnosis: Analyze logs, compare environments
# 3. Fix Application: Apply fixes based on failure patterns
# 4. Push and Wait: Commit fixes, push, wait for CI re-run
# 5. Success Verification: Confirm all checks pass
# Iterative loop:
# Phases 2-4 repeat until success or max iterations (5)Key Characteristics:
- Iterative: Repeats fix cycle until success
- Autonomous Recovery: Tries multiple fix strategies
- State Management: Tracks attempted fixes, avoids repeating failures
- Pattern Matching: Recognizes common CI failure types
- Escalation: Reports to user after max iterations
Implementation:
# Located in: .claude/agents/amplihack/specialized/ci-diagnostic-workflow.md
# Fix iteration loop:
MAX_ITERATIONS = 5
iteration = 0
while iteration < MAX_ITERATIONS:
status = check_ci_status()
if status["conclusion"] == "success":
break
# Diagnose failures
failures = analyze_ci_logs(status)
# Apply pattern-matched fixes
for failure in failures:
if "test" in failure["type"]:
fix_test_failure(failure)
elif "lint" in failure["type"]:
fix_lint_failure(failure)
elif "type" in failure["type"]:
fix_type_failure(failure)
# Commit and push
git_commit_and_push(f"fix: CI iteration {iteration + 1}")
# Wait for CI re-run
wait_for_ci_completion()
iteration += 1
if iteration >= MAX_ITERATIONS:
escalate_to_user("CI still failing after 5 iterations")Lessons Learned:
- Iteration limits prevent infinite loops
- Pattern matching (test/lint/type) enables targeted fixes
- Smart waiting (exponential backoff) reduces wait time
- Never auto-merge: human approval always required
Example 3: Pre-Commit Diagnostic Workflow
Problem: Pre-commit hooks fail with unclear errors
Goal-Seeking Solution:
# Goal: Fix pre-commit hook failures before commit
goal = """
Pre-commit hooks failing.
Diagnose issues (formatting, linting, type checking).
Apply fixes locally, re-run hooks.
Ensure all hooks pass before allowing commit.
"""
# Agent decomposes into phases:
# 1. Hook Failure Analysis: Identify which hooks failed
# 2. Environment Check: Compare local vs pre-commit versions
# 3. Targeted Fixes: Apply fixes per hook type
# 4. Hook Re-run: Validate fixes, iterate if needed
# 5. Commit Readiness: Confirm all hooks passKey Characteristics:
- Pre-Push Focus: Fixes issues before pushing to CI
- Tool Version Management: Ensures local matches pre-commit config
- Hook-Specific Fixes: Tailored approach per hook type
- Fast Iteration: No wait for CI, immediate feedback
Implementation:
# Located in: .claude/agents/amplihack/specialized/pre-commit-diagnostic.md
# Hook failure patterns:
HOOK_FIXES = {
"ruff": lambda: subprocess.run(["ruff", "check", "--fix", "."]),
"black": lambda: subprocess.run(["black", "."]),
"mypy": lambda: add_type_ignores(),
"trailing-whitespace": lambda: subprocess.run(["pre-commit", "run", "trailing-whitespace", "--all-files"]),
}
# Execution:
failed_hooks = detect_failed_hooks()
for hook in failed_hooks:
if hook in HOOK_FIXES:
HOOK_FIXES[hook]()
else:
generic_fix(hook)
# Re-run to verify
rerun_result = subprocess.run(["pre-commit", "run", "--all-files"])
if rerun_result.returncode == 0:
print("All hooks passing, ready to commit!")Lessons Learned:
- Pre-commit fixes are faster than CI iteration
- Tool version mismatches are common culprit
- Automated fixes for 80% of cases
- Remaining 20% escalate with clear diagnostics
Example 4: Fix-Agent Pattern Matching
Problem: Different issues require different fix approaches
Goal-Seeking Solution:
# Goal: Select optimal fix strategy based on problem context
goal = """
Analyze issue and select fix mode:
- QUICK: Obvious fixes (< 5 min)
- DIAGNOSTIC: Unclear root cause (investigation)
- COMPREHENSIVE: Complex issues (full workflow)
"""
# Agent decomposes into phases:
# 1. Issue Analysis: Classify problem type and complexity
# 2. Mode Selection: Choose QUICK/DIAGNOSTIC/COMPREHENSIVE
# 3. Fix Execution: Apply mode-appropriate strategy
# 4. Validation: Verify fix resolves issueKey Characteristics:
- Context-Aware: Selects strategy based on problem analysis
- Multi-Mode: Three fix modes for different complexity levels
- Pattern Recognition: Learns from past fixes
- Adaptive: Escalates complexity if initial mode fails
Implementation:
# Located in: .claude/agents/amplihack/specialized/fix-agent.md
# Mode selection logic:
def select_fix_mode(issue: Issue) -> FixMode:
if issue.is_obvious() and issue.scope == "single-file":
return FixMode.QUICK
elif issue.root_cause_unclear():
return FixMode.DIAGNOSTIC
elif issue.is_complex() or issue.requires_architecture_change():
return FixMode.COMPREHENSIVE
else:
return FixMode.DIAGNOSTIC # Default to investigation
# Pattern frequency (from real usage):
FIX_PATTERNS = {
"import": 0.15, # Import errors (15%)
"config": 0.12, # Configuration issues (12%)
"test": 0.18, # Test failures (18%)
"ci": 0.20, # CI/CD problems (20%)
"quality": 0.25, # Code quality (linting, types) (25%)
"logic": 0.10, # Logic errors (10%)
}
# Template-based fixes for common patterns:
if issue.pattern == "import":
apply_template("import-fix-template", issue)
elif issue.pattern == "config":
apply_template("config-fix-template", issue)
# ... etcLessons Learned:
- Pattern matching enables template-based fixes (80% coverage)
- Mode selection reduces over-engineering (right-sized approach)
- Diagnostic mode critical for unclear issues (root cause analysis)
- Usage data informs template priorities
6. Design Checklist
Use this checklist when designing goal-seeking agents:
Goal Definition
- [ ] Objective is clear and well-defined
- [ ] Success criteria are measurable and verifiable
- [ ] Constraints are explicit (time, resources, safety)
- [ ] Domain is identified (impacts phase templates)
- [ ] Complexity is estimated (simple/moderate/complex)
Phase Design
- [ ] Decomposed into 3-5 phases (not too granular, not too coarse)
- [ ] Phase dependencies are explicit
- [ ] Parallel execution opportunities identified
- [ ] Each phase has clear success indicators
- [ ] Phase durations are estimated
Skill Mapping
- [ ] Required capabilities identified per phase
- [ ] Skills mapped to existing agents or tools
- [ ] Delegation targets specified
- [ ] No missing capabilities
Error Handling
- [ ] Retry strategies defined (max attempts, backoff)
- [ ] Alternative strategies identified
- [ ] Escalation criteria clear (when to ask for help)
- [ ] Graceful degradation options (fallback approaches)
State Management
- [ ] State tracked across phases
- [ ] Phase results stored for downstream use
- [ ] Failure history maintained
- [ ] Context shared appropriately
Testing
- [ ] Success scenarios tested
- [ ] Failure recovery tested
- [ ] Edge cases identified
- [ ] Performance validated (duration, resource usage)
Documentation
- [ ] Goal clearly documented
- [ ] Phase descriptions complete
- [ ] Usage examples provided
- [ ] Integration points specified
Philosophy Compliance
- [ ] Ruthless simplicity (no unnecessary complexity)
- [ ] Single responsibility per phase
- [ ] No over-engineering (right-sized solution)
- [ ] Regeneratable (clear specifications)
7. Agent SDK Integration (Future)
When the Agent SDK Skill is integrated, goal-seeking agents can leverage:
Enhanced Autonomy
# Agent SDK provides enhanced context management
from claude_agent_sdk import AgentContext, Tool
class GoalSeekingAgent:
def __init__(self, context: AgentContext):
self.context = context
self.state = {}
async def execute_phase(self, phase: PlanPhase):
# SDK provides tools, memory, delegation
tools = self.context.get_tools(phase.required_capabilities)
memory = self.context.get_memory()
# Execute with SDK support
result = await phase.execute(tools, memory)
# Store in context for downstream phases
self.context.store_result(phase.name, result)Tool Discovery
# SDK enables dynamic tool discovery
available_tools = context.discover_tools(capability="data-processing")
# Select optimal tool for task
tool = context.select_tool(
capability="data-transformation",
criteria={"performance": "high", "accuracy": "required"}
)Memory Management
# SDK provides persistent memory across sessions
context.memory.store("deployment-history", deployment_record)
previous = context.memory.retrieve("deployment-history")
# Enables learning from past executions
if previous and previous.failed:
# Avoid previous failure strategy
strategy = select_alternative_strategy(previous.failure_reason)Agent Delegation
# SDK simplifies agent-to-agent delegation
result = await context.delegate(
agent="security-analyzer",
task="audit-rbac-policies",
input={"cluster": cluster_name}
)
# Parallel delegation
results = await context.delegate_parallel([
("security-analyzer", "audit-rbac-policies"),
("network-analyzer", "validate-ingress"),
("monitoring-validator", "check-metrics")
])Observability
# SDK provides built-in tracing and metrics
with context.trace("data-transformation"):
result = transform_data(input_data)
context.metrics.record("transformation-duration", duration)
context.metrics.record("transformation-accuracy", accuracy)Integration Example
from claude_agent_sdk import AgentContext, create_agent
from amplihack.goal_agent_generator import GoalAgentBundle
# Create SDK-enabled goal-seeking agent
def create_goal_agent(bundle: GoalAgentBundle) -> Agent:
context = AgentContext(
name=bundle.name,
version=bundle.version,
capabilities=bundle.metadata["required_capabilities"]
)
# Register phases as agent tasks
for phase in bundle.execution_plan.phases:
context.register_task(
name=phase.name,
capabilities=phase.required_capabilities,
executor=create_phase_executor(phase)
)
# Create agent with SDK
agent = create_agent(context)
# Execute goal
return agent
# Usage:
agent = create_goal_agent(agent_bundle)
result = await agent.execute(bundle.auto_mode_config["initial_prompt"])8. Trade-Off Analysis
Goal-Seeking vs Traditional Agents
| Dimension | Goal-Seeking Agent | Traditional Agent |
|---|---|---|
| Flexibility | High - adapts to context | Low - fixed workflow |
| Development Time | Moderate - define goals & phases | Low - script steps |
| Execution Time | Higher - decision overhead | Lower - direct execution |
| Maintenance | Lower - self-adapting | Higher - manual updates |
| Debuggability | Harder - dynamic behavior | Easier - predictable flow |
| Reusability | High - same agent, different contexts | Low - context-specific |
| Failure Handling | Autonomous recovery | Manual intervention |
| Complexity | Higher - multi-phase coordination | Lower - linear execution |
When to Choose Each
Choose Goal-Seeking when:
- Problem space is large with many valid approaches
- Context varies significantly across executions
- Autonomous recovery is valuable
- Reusability across contexts is important
- Development time investment is justified
Choose Traditional when:
- Single deterministic path exists
- Performance is critical (low latency required)
- Simplicity is paramount
- One-off or rare execution
- Debugging and auditability are critical
Cost-Benefit Analysis
Goal-Seeking Costs:
- Higher development time (define goals, phases, capabilities)
- Increased execution time (decision overhead)
- More complex testing (dynamic behavior)
- Harder debugging (non-deterministic paths)
Goal-Seeking Benefits:
- Autonomous operation (less human intervention)
- Adaptive to context (works in varied conditions)
- Reusable across problems (same agent, different goals)
- Self-recovering (handles failures gracefully)
Break-Even Point: Goal-seeking justified when problem is:
- Repeated 2+ times per week, OR
- Takes 30+ minutes manual execution, OR
- Requires expert knowledge hard to document, OR
- High value from autonomous recovery
9. When to Escalate
Goal-seeking agents should escalate to humans when:
Hard Limits Reached
Max Iterations Exceeded:
if iteration_count >= MAX_ITERATIONS:
escalate(
reason="Reached maximum iterations without success",
context={
"iterations": iteration_count,
"attempted_strategies": attempted_strategies,
"last_error": last_error
}
)Timeout Exceeded:
if elapsed_time > MAX_DURATION:
escalate(
reason="Execution time exceeded limit",
context={
"elapsed": elapsed_time,
"max_allowed": MAX_DURATION,
"completed_phases": completed_phases
}
)Safety Boundaries
Destructive Operations:
if operation.is_destructive() and not operation.has_approval():
escalate(
reason="Destructive operation requires human approval",
operation=operation.description,
impact=operation.estimate_impact()
)Production Changes:
if target_environment == "production":
escalate(
reason="Production deployments require human verification",
changes=proposed_changes,
rollback_plan=rollback_strategy
)Uncertainty Detection
Low Confidence:
if decision_confidence < CONFIDENCE_THRESHOLD:
escalate(
reason="Confidence below threshold for autonomous decision",
decision=decision_description,
confidence=decision_confidence,
alternatives=alternative_options
)Conflicting Strategies:
if len(viable_strategies) > 1 and not clear_winner:
escalate(
reason="Multiple viable strategies, need human judgment",
strategies=viable_strategies,
trade_offs=strategy_trade_offs
)Unexpected Conditions
Unrecognized Errors:
if error_type not in KNOWN_ERROR_PATTERNS:
escalate(
reason="Encountered unknown error pattern",
error=error_details,
context=execution_context,
recommendation="Manual investigation required"
)Environment Mismatch:
if detected_environment != expected_environment:
escalate(
reason="Environment mismatch detected",
expected=expected_environment,
detected=detected_environment,
risk="Potential for incorrect behavior"
)Escalation Best Practices
Provide Context:
- What was attempted
- What failed and why
- What alternatives were considered
- Current system state
Suggest Actions:
- Recommend next steps
- Provide diagnostic commands
- Offer manual intervention points
- Suggest rollback if needed
Enable Recovery:
- Save execution state
- Document failures
- Provide resume capability
- Offer manual override
Example Escalation:
escalate(
reason="CI failure diagnosis unsuccessful after 5 iterations",
context={
"iterations": 5,
"attempted_fixes": [
"Import path corrections (iteration 1)",
"Type annotation fixes (iteration 2)",
"Test environment setup (iteration 3)",
"Dependency version pins (iteration 4)",
"Mock configuration (iteration 5)"
],
"persistent_failures": [
"test_integration.py::test_api_connection - Timeout",
"test_models.py::test_validation - Assertion error"
],
"system_state": "2 of 25 tests still failing",
"ci_logs": "https://github.com/.../actions/runs/123456"
},
recommendations=[
"Review test_api_connection timeout - may need increased timeout or mock",
"Examine test_validation assertion - data structure may have changed",
"Consider running tests locally with same environment as CI",
"Check if recent changes affected integration test setup"
],
next_steps={
"manual_investigation": "Run failing tests locally with verbose output",
"rollback_option": "git revert HEAD~5 if fixes made things worse",
"resume_point": "Fix failures and run /amplihack:ci-diagnostic to resume"
}
)10. Example Workflow
Complete example: Building a goal-seeking agent for data pipeline automation
Step 1: Define Goal
# Goal: Automate Multi-Source Data Pipeline
## Objective
Collect data from multiple sources (S3, database, API), transform to common schema, validate quality, publish to data warehouse.
## Success Criteria
- All sources successfully ingested
- Data transformed to target schema
- Quality checks pass (completeness, accuracy)
- Data published to warehouse
- Pipeline completes within 30 minutes
## Constraints
- Must handle source unavailability gracefully
- No data loss (failed records logged)
- Idempotent (safe to re-run)
- Resource limits: 8GB RAM, 4 CPU cores
## Context
- Daily execution (automated schedule)
- Priority: High (blocking downstream analytics)
- Scale: Medium (100K-1M records per source)Step 2: Analyze with PromptAnalyzer
from amplihack.goal_agent_generator import PromptAnalyzer
analyzer = PromptAnalyzer()
goal_definition = analyzer.analyze_text(goal_text)
# Result:
# goal_definition.goal = "Automate Multi-Source Data Pipeline"
# goal_definition.domain = "data-processing"
# goal_definition.complexity = "moderate"
# goal_definition.constraints = [
# "Must handle source unavailability gracefully",
# "No data loss (failed records logged)",
# "Idempotent (safe to re-run)",
# "Resource limits: 8GB RAM, 4 CPU cores"
# ]
# goal_definition.success_criteria = [
# "All sources successfully ingested",
# "Data transformed to target schema",
# "Quality checks pass (completeness, accuracy)",
# "Data published to warehouse",
# "Pipeline completes within 30 minutes"
# ]Step 3: Generate Plan with ObjectivePlanner
from amplihack.goal_agent_generator import ObjectivePlanner
planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_definition)
# Result: 4-phase plan
# Phase 1: Data Collection (parallel)
# - Collect from S3 (parallel-safe)
# - Collect from database (parallel-safe)
# - Collect from API (parallel-safe)
# Duration: 15 minutes
# Success: All sources attempted, failures logged
#
# Phase 2: Data Transformation (depends on Phase 1)
# - Parse raw data
# - Transform to common schema
# - Handle missing fields
# Duration: 15 minutes
# Success: All records transformed or logged as failed
#
# Phase 3: Quality Validation (depends on Phase 2)
# - Completeness check
# - Accuracy validation
# - Consistency verification
# Duration: 5 minutes
# Success: Quality thresholds met
#
# Phase 4: Data Publishing (depends on Phase 3)
# - Load to warehouse
# - Update metadata
# - Generate report
# Duration: 10 minutes
# Success: Data in warehouse, report generatedStep 4: Synthesize Skills
from amplihack.goal_agent_generator import SkillSynthesizer
synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)
# Result: 3 skills
# Skill 1: data-collector
# Capabilities: ["s3-read", "database-query", "api-fetch"]
# Implementation: "native" (built-in)
#
# Skill 2: data-transformer
# Capabilities: ["parsing", "schema-mapping", "validation"]
# Implementation: "native" (built-in)
#
# Skill 3: data-publisher
# Capabilities: ["warehouse-load", "metadata-update", "reporting"]
# Implementation: "delegated" (delegates to warehouse tool)Step 5: Assemble Agent
from amplihack.goal_agent_generator import AgentAssembler
assembler = AgentAssembler()
agent_bundle = assembler.assemble(
goal_definition=goal_definition,
execution_plan=execution_plan,
skills=skills,
bundle_name="multi-source-data-pipeline"
)
# Result: GoalAgentBundle
# - Name: multi-source-data-pipeline
# - Max turns: 12 (moderate complexity, 4 phases)
# - Initial prompt: Full execution plan with phases
# - Status: "ready"Step 6: Package Agent
from amplihack.goal_agent_generator import GoalAgentPackager
from pathlib import Path
packager = GoalAgentPackager()
packager.package(
bundle=agent_bundle,
output_dir=Path(".claude/agents/goal-driven/multi-source-data-pipeline")
)
# Creates agent package:
# .claude/agents/goal-driven/multi-source-data-pipeline/
# ├── agent.md # Agent definition
# ├── prompt.md # Execution prompt
# ├── metadata.json # Bundle metadata
# ├── plan.yaml # Execution plan (4 phases)
# └── skills.yaml # 3 required skillsStep 7: Execute Agent (Auto-Mode)
# Execute via CLI
amplihack goal-agent-generator execute \
--agent-path .claude/agents/goal-driven/multi-source-data-pipeline \
--auto-mode \
--max-turns 12
# Or programmatically:from claude_code import execute_auto_mode
result = execute_auto_mode(
initial_prompt=agent_bundle.auto_mode_config["initial_prompt"],
max_turns=agent_bundle.auto_mode_config["max_turns"],
working_dir=agent_bundle.auto_mode_config["working_dir"]
)Step 8: Monitor Execution
Agent executes autonomously:
Phase 1: Data Collection [In Progress]
├── S3 Collection: ✓ COMPLETED (50K records, 5 minutes)
├── Database Collection: ✓ COMPLETED (75K records, 8 minutes)
└── API Collection: ✗ FAILED (timeout, retrying...)
└── Retry 1: ✓ COMPLETED (25K records, 4 minutes)
Phase 1: ✓ COMPLETED (150K records total, 3 sources, 17 minutes)
Phase 2: Data Transformation [In Progress]
├── Parsing: ✓ COMPLETED (150K records parsed)
├── Schema Mapping: ✓ COMPLETED (148K records mapped, 2K failed)
└── Missing Fields: ✓ COMPLETED (defaults applied)
Phase 2: ✓ COMPLETED (148K records ready, 2K logged as failed, 12 minutes)
Phase 3: Quality Validation [In Progress]
├── Completeness: ✓ PASS (98.7% complete, threshold 95%)
├── Accuracy: ✓ PASS (99.2% accurate, threshold 98%)
└── Consistency: ✓ PASS (100% consistent)
Phase 3: ✓ COMPLETED (All checks passed, 4 minutes)
Phase 4: Data Publishing [In Progress]
├── Warehouse Load: ✓ COMPLETED (148K records loaded)
├── Metadata Update: ✓ COMPLETED (pipeline_run_id: 12345)
└── Report Generation: ✓ COMPLETED (report.html)
Phase 4: ✓ COMPLETED (Data published, 8 minutes)
Total Execution: ✓ SUCCESS (41 minutes, all success criteria met)Step 9: Review Results
# Pipeline Execution Report
## Summary
- **Status**: SUCCESS
- **Duration**: 41 minutes (estimated: 30 minutes)
- **Records Processed**: 150K ingested, 148K published
- **Success Rate**: 98.7%
## Phase Results
### Phase 1: Data Collection
- S3: 50K records (5 min)
- Database: 75K records (8 min)
- API: 25K records (4 min, 1 retry)
### Phase 2: Data Transformation
- Successfully transformed: 148K records
- Failed transformations: 2K records (logged to failed_records.log)
- Failure reasons: Schema mismatch (1.5K), Invalid data (500)
### Phase 3: Quality Validation
- Completeness: 98.7% ✓
- Accuracy: 99.2% ✓
- Consistency: 100% ✓
### Phase 4: Data Publishing
- Warehouse load: Success
- Pipeline run ID: 12345
- Report: report.html
## Issues Encountered
1. API timeout (Phase 1): Resolved with retry
2. 2K transformation failures: Logged for manual review
## Recommendations
1. Investigate schema mismatches in API data
2. Add validation for API data format
3. Consider increasing timeout for API callsStep 10: Iteration (If Needed)
If pipeline fails, agent adapts:
# Example: API source completely unavailable
if phase1_result["api"]["status"] == "unavailable":
# Agent adapts: continues with partial data
log_warning("API source unavailable, continuing with S3 + database")
proceed_to_phase2_with_partial_data()
# Report notes partial data
add_to_report("Data incomplete: API source unavailable")
# Example: Quality validation fails
if phase3_result["completeness"] < THRESHOLD:
# Agent tries recovery: fetch missing data
missing_records = identify_missing_records()
retry_collection_for_missing(missing_records)
rerun_transformation()
rerun_validation()
# If still fails after retry, escalate
if still_below_threshold:
escalate("Quality threshold not met after retry")11. Related Patterns
Goal-seeking agents relate to and integrate with other patterns:
Debate Pattern (Multi-Agent Decision Making)
When to Combine:
- Goal-seeking agent faces complex decision with trade-offs
- Multiple valid approaches exist
- Need consensus from different perspectives
Example:
# Goal-seeking agent reaches decision point
if len(viable_strategies) > 1:
# Invoke debate pattern
result = invoke_debate(
question="Which data transformation approach?",
perspectives=["performance", "accuracy", "simplicity"],
context=current_state
)
# Use debate result to select strategy
selected_strategy = result.consensusN-Version Pattern (Redundant Implementation)
When to Combine:
- Goal-seeking agent executing critical phase
- Error cost is high
- Multiple independent implementations possible
Example:
# Critical security validation phase
if phase.is_critical():
# Generate N versions
results = generate_n_versions(
phase=phase,
n=3,
independent=True
)
# Use voting or comparison to select result
validated_result = compare_and_validate(results)Cascade Pattern (Fallback Strategies)
When to Combine:
- Goal-seeking agent has preferred approach but needs fallbacks
- Quality/performance trade-offs exist
- Graceful degradation desired
Example:
# Data transformation with fallback
try:
# Optimal: ML-based transformation
result = ml_transform(data)
except MLModelUnavailable:
try:
# Pragmatic: Rule-based transformation
result = rule_based_transform(data)
except RuleEngineError:
# Minimal: Manual templates
result = template_transform(data)Investigation Workflow (Knowledge Discovery)
When to Combine:
- Goal requires understanding existing system
- Need to discover architecture or patterns
- Knowledge excavation before execution
Example:
# Before automating deployment, understand current system
if goal.requires_system_knowledge():
# Run investigation workflow
investigation = run_investigation_workflow(
scope="deployment pipeline",
depth="comprehensive"
)
# Use findings to inform goal-seeking execution
adapt_plan_based_on_investigation(investigation.findings)Document-Driven Development (Specification First)
When to Combine:
- Goal-seeking agent generates or modifies code
- Clear specifications prevent drift
- Documentation is single source of truth
Example:
# Goal: Implement new feature
if goal.involves_code_changes():
# DDD Phase 1: Generate specifications
specs = generate_specifications(goal)
# DDD Phase 2: Review and approve specs
await human_review(specs)
# Goal-seeking agent implements from specs
implementation = execute_from_specifications(specs)Pre-Commit / CI Diagnostic (Quality Gates)
When to Combine:
- Goal-seeking agent makes code changes
- Need to ensure quality before commit/push
- Automated validation and fixes
Example:
# After goal-seeking agent generates code
if changes_made:
# Run pre-commit diagnostic
pre_commit_result = run_pre_commit_diagnostic()
if pre_commit_result.has_failures():
# Agent fixes issues
apply_pre_commit_fixes(pre_commit_result.failures)
# After push, run CI diagnostic
ci_result = run_ci_diagnostic_workflow()
if ci_result.has_failures():
# Agent iterates fixes
iterate_ci_fixes_until_pass(ci_result)12. Quality Standards
Goal-seeking agents must meet these quality standards:
Correctness
Success Criteria Verification:
- [ ] Agent verifies all success criteria before completion
- [ ] Intermediate phase results validated
- [ ] No silent failures (all errors logged and handled)
Testing Coverage:
- [ ] Happy path tested (all success criteria met)
- [ ] Failure scenarios tested (phase failures, retries)
- [ ] Edge cases identified and tested
- [ ] Integration with real systems validated
Resilience
Error Handling:
- [ ] Retry logic with exponential backoff
- [ ] Alternative strategies for common failures
- [ ] Graceful degradation when optimal path unavailable
- [ ] Clear escalation criteria
State Management:
- [ ] State persisted across phase boundaries
- [ ] Resume capability after failures
- [ ] Idempotent execution (safe to re-run)
- [ ] Cleanup on abort
Performance
Efficiency:
- [ ] Phases execute in parallel when possible
- [ ] No unnecessary work (skip completed phases on retry)
- [ ] Resource usage within limits (memory, CPU, time)
- [ ] Timeout limits enforced
Latency:
- [ ] Decision overhead acceptable for use case
- [ ] No blocking waits (async where possible)
- [ ] Progress reported (no black box periods)
Observability
Logging:
- [ ] Phase transitions logged
- [ ] Decisions logged with reasoning
- [ ] Errors logged with context
- [ ] Results logged with metrics
Metrics:
- [ ] Duration per phase tracked
- [ ] Success/failure rates tracked
- [ ] Resource usage monitored
- [ ] Quality metrics reported
Tracing:
- [ ] Execution flow traceable
- [ ] Correlations across phases maintained
- [ ] Debugging information sufficient
Usability
Documentation:
- [ ] Goal clearly stated
- [ ] Success criteria documented
- [ ] Usage examples provided
- [ ] Integration guide complete
User Experience:
- [ ] Clear progress reporting
- [ ] Actionable error messages
- [ ] Human-readable outputs
- [ ] Easy to invoke and monitor
Philosophy Compliance
Ruthless Simplicity:
- [ ] No unnecessary phases or complexity
- [ ] Simplest approach that works
- [ ] No premature optimization
Single Responsibility:
- [ ] Each phase has one clear job
- [ ] No overlapping responsibilities
- [ ] Clean phase boundaries
Modularity:
- [ ] Skills are reusable across agents
- [ ] Phases are independent
- [ ] Clear interfaces (inputs/outputs)
Regeneratable:
- [ ] Can be rebuilt from specifications
- [ ] No hardcoded magic values
- [ ] Configuration externalized
13. Getting Started
Quick Start: Build Your First Goal-Seeking Agent
Step 1: Install amplihack (if not already)
pip install amplihackStep 2: Write a goal prompt
cat > my-goal.md << 'EOF'
# Goal: Automated Security Audit
Check application for common security issues:
- SQL injection vulnerabilities
- XSS vulnerabilities
- Insecure dependencies
- Missing security headers
Generate report with severity levels and remediation steps.
EOFStep 3: Generate agent
amplihack goal-agent-generator create \
--prompt my-goal.md \
--output .claude/agents/goal-driven/security-auditorStep 4: Review generated plan
cat .claude/agents/goal-driven/security-auditor/plan.yamlStep 5: Execute agent
amplihack goal-agent-generator execute \
--agent-path .claude/agents/goal-driven/security-auditor \
--auto-modeCommon Use Cases
Use Case 1: Workflow Automation
# Create release automation agent
echo "Automate release workflow: tag, build, test, deploy to staging" | \
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/release-automatorUse Case 2: Data Pipeline
# Create ETL pipeline agent
echo "Extract from sources, transform to schema, validate quality, load to warehouse" | \
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/etl-pipelineUse Case 3: Diagnostic Workflow
# Create performance diagnostic agent
echo "Diagnose application performance issues, identify bottlenecks, suggest optimizations" | \
amplihack goal-agent-generator create --inline --output .claude/agents/goal-driven/perf-diagnosticLearning Resources
Documentation:
- Review examples in
~/.amplihack/.claude/skills/goal-seeking-agent-pattern/examples/ - Read real agent implementations in
~/.amplihack/.claude/agents/amplihack/specialized/ - Check integration guide in
~/.amplihack/.claude/skills/goal-seeking-agent-pattern/templates/integration_guide.md
Practice:
1. Start simple: Build single-phase agent (e.g., file formatter) 2. Add complexity: Build multi-phase agent (e.g., test generator + runner) 3. Add autonomy: Build agent with error recovery (e.g., CI fixer) 4. Build production: Build full goal-seeking agent (e.g., deployment pipeline)
Get Help:
- Review decision framework (Section 2)
- Check design checklist (Section 6)
- Study real examples (Section 5)
- Ask architect agent for guidance
Next Steps
After building your first goal-seeking agent:
1. Test thoroughly: Cover success, failure, and edge cases 2. Monitor in production: Track metrics, logs, failures 3. Iterate: Refine based on real usage 4. Document learnings: Update DISCOVERIES.md with insights 5. Share patterns: Add successful approaches to PATTERNS.md
Success Indicators:
- Agent completes goal autonomously 80%+ of time
- Failures escalate with clear context
- Execution time is acceptable
- Users trust agent to run autonomously
---
Remember: Goal-seeking agents should be ruthlessly simple, focused on clear objectives, and adaptive to context. Start simple, add complexity only when justified, and always verify against success criteria.
Example: Adaptive Testing with Goal-Seeking Agents
Scenario: Intelligent Test Generation and Execution
Problem Statement
Manual test creation and maintenance is:
- Time-consuming: 30-60 minutes to write comprehensive tests for new features
- Coverage-incomplete: Easy to miss edge cases and error paths
- Maintenance-heavy: Tests break when code changes, require manual updates
- Context-unaware: Same test strategy regardless of code complexity
- Flaky: Tests fail intermittently, require manual investigation
Is Goal-Seeking Appropriate?
Apply the 5-question decision framework:
Q1: Well-defined objective but flexible path?
- YES: Objective is clear (generate and run comprehensive tests)
- Multiple paths:
- Simple functions: Basic unit tests
- Complex logic: Property-based tests, edge cases
- APIs: Integration tests with mocks
- Flaky tests: Retry strategies, better assertions
- Success criteria: ≥ 80% coverage, all tests pass
Q2: Multiple phases with dependencies?
- YES: 4 phases with dependencies
1. Code Analysis (understand what to test) 2. Test Generation (create tests based on analysis) 3. Test Execution (run tests, handle failures) 4. Coverage Analysis (verify completeness, suggest improvements)
Q3: Autonomous recovery valuable?
- YES: Test failures are common and often fixable
- Flaky tests: Retry with better waits
- Import errors: Auto-fix imports
- Assertion errors: Suggest better assertions
- Mock setup issues: Auto-configure mocks
Q4: Context affects approach?
- YES: Test strategy varies by:
- Code complexity (simple vs complex algorithms)
- Code type (pure functions vs stateful classes vs APIs)
- Existing coverage (gaps vs comprehensive)
- Flakiness history (stable vs intermittent failures)
Q5: Complexity justified?
- YES: High-value automation
- Frequency: Every feature (50-100 times per year)
- Manual time: 30-60 minutes per feature
- Value: 25-100 hours saved per year
- Quality: Catches more edge cases than manual testing
Conclusion: All 5 YES → Goal-seeking agent is appropriate
Goal-Seeking Agent Design
Goal Definition
# Goal: Intelligent Test Generation and Execution
## Objective
Analyze code to understand functionality, generate comprehensive tests
covering happy paths and edge cases, execute tests with intelligent retry
for flaky failures, and verify coverage thresholds are met.
## Success Criteria
- Code analyzed to identify test requirements
- Tests generated for all public functions/methods
- Coverage ≥ 80% (line coverage)
- All tests pass (or failures are investigated and fixed)
- Edge cases identified and tested
- Flaky tests handled with retries or better assertions
## Constraints
- Must preserve existing tests (don't overwrite)
- Test framework: pytest (Python)
- Max 5 test iterations (prevent infinite loops)
- Tests must run in < 5 minutes
- No external dependencies (use mocks)
## Context
- Language: Python
- Framework: pytest
- Target: New feature functions or classes
- Priority: High (blocking feature merge)Execution Plan
from amplihack.goal_agent_generator import PromptAnalyzer, ObjectivePlanner
analyzer = PromptAnalyzer()
goal_def = analyzer.analyze_text(goal_text)
planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_def)
# Result: 4-phase planPhase 1: Code Analysis (5 minutes)
- Parse source code (AST analysis)
- Identify functions, classes, methods
- Extract function signatures, docstrings
- Detect complexity (cyclomatic complexity)
- Identify dependencies (imports, external calls)
Dependencies: None Success indicators:
- All public functions identified
- Signatures extracted
- Complexity assessed
- Dependencies mapped
Phase 2: Test Generation (10 minutes, depends on Phase 1)
- Generate unit tests for simple functions
- Generate property-based tests for complex logic
- Generate integration tests for APIs
- Create fixtures and mocks
- Add edge case tests
Dependencies: Phase 1 (needs code analysis) Success indicators:
- Tests generated for all functions
- Edge cases covered
- Fixtures/mocks created
- Test files organized properly
Phase 3: Test Execution (15 minutes, depends on Phase 2)
- Run pytest on generated tests
- Capture failures and analyze
- Apply fixes for common failures
- Retry flaky tests with better strategies
- Re-run until all pass or max iterations
Dependencies: Phase 2 (needs generated tests) Success indicators:
- All tests executed
- Failures analyzed and fixed
- Flaky tests identified and handled
- Final run: all tests pass
Phase 4: Coverage Analysis (5 minutes, depends on Phase 3)
- Run pytest with coverage
- Analyze coverage report
- Identify uncovered lines
- Suggest additional tests for gaps
- Verify coverage threshold met
Dependencies: Phase 3 (needs passing tests) Success indicators:
- Coverage ≥ 80%
- Coverage gaps identified
- Suggestions for improvements
- Report generated
Total Duration: 35 minutes (estimated)
Implementation
from amplihack.goal_agent_generator import (
PromptAnalyzer,
ObjectivePlanner,
SkillSynthesizer,
AgentAssembler,
GoalAgentPackager,
)
from pathlib import Path
import ast
import subprocess
from typing import List, Dict, Any
# Goal definition
goal_text = """
Analyze code, generate comprehensive tests (unit + edge cases),
execute tests with intelligent retry for flaky failures,
verify coverage ≥ 80%.
"""
# Create agent
analyzer = PromptAnalyzer()
goal_def = analyzer.analyze_text(goal_text)
planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_def)
synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)
assembler = AgentAssembler()
agent_bundle = assembler.assemble(
goal_definition=goal_def,
execution_plan=execution_plan,
skills=skills,
bundle_name="adaptive-test-generator"
)
packager = GoalAgentPackager()
packager.package(
bundle=agent_bundle,
output_dir=Path(".claude/agents/goal-driven/adaptive-test-generator")
)Adaptive Behavior
The agent adapts based on code characteristics:
Scenario 1: Simple Pure Function (basic unit tests)
# Code to test
def calculate_total(items: List[float]) -> float:
"""Calculate total of item prices."""
return sum(items)
# Agent generates:
def test_calculate_total_happy_path():
"""Test with valid input"""
assert calculate_total([1.0, 2.0, 3.0]) == 6.0
def test_calculate_total_empty_list():
"""Test edge case: empty list"""
assert calculate_total([]) == 0.0
def test_calculate_total_single_item():
"""Test edge case: single item"""
assert calculate_total([5.0]) == 5.0
def test_calculate_total_negative_values():
"""Test edge case: negative values"""
assert calculate_total([-1.0, 2.0]) == 1.0Scenario 2: Complex Logic (property-based tests)
# Code to test
def binary_search(arr: List[int], target: int) -> int:
"""Binary search implementation. Returns index or -1."""
left, right = 0, len(arr) - 1
while left <= right:
mid = (left + right) // 2
if arr[mid] == target:
return mid
elif arr[mid] < target:
left = mid + 1
else:
right = mid - 1
return -1
# Agent detects complexity (cyclomatic = 5) and generates property-based tests:
from hypothesis import given, strategies as st
@given(st.lists(st.integers()).map(sorted), st.integers())
def test_binary_search_property_found(sorted_list, target):
"""Property: If target in list, result should be valid index"""
result = binary_search(sorted_list, target)
if target in sorted_list:
assert result != -1
assert sorted_list[result] == target
@given(st.lists(st.integers()).map(sorted), st.integers())
def test_binary_search_property_not_found(sorted_list, target):
"""Property: If target not in list, result should be -1"""
result = binary_search(sorted_list, target)
if target not in sorted_list:
assert result == -1
def test_binary_search_edge_empty():
"""Edge case: empty list"""
assert binary_search([], 5) == -1
def test_binary_search_edge_single():
"""Edge case: single element"""
assert binary_search([5], 5) == 0
assert binary_search([5], 3) == -1Scenario 3: API Endpoint (integration tests with mocks)
# Code to test
from flask import Flask, jsonify, request
app = Flask(__name__)
@app.route('/api/users', methods=['POST'])
def create_user():
data = request.get_json()
user = database.create_user(data['name'], data['email'])
return jsonify(user), 201
# Agent generates integration tests with mocks:
import pytest
from unittest.mock import patch, MagicMock
@pytest.fixture
def client():
"""Flask test client"""
app.config['TESTING'] = True
return app.test_client()
@pytest.fixture
def mock_database():
"""Mock database"""
with patch('app.database') as mock_db:
yield mock_db
def test_create_user_success(client, mock_database):
"""Test successful user creation"""
mock_database.create_user.return_value = {
'id': 1, 'name': 'John', 'email': 'john@example.com'
}
response = client.post('/api/users', json={
'name': 'John',
'email': 'john@example.com'
})
assert response.status_code == 201
assert response.json['name'] == 'John'
mock_database.create_user.assert_called_once()
def test_create_user_missing_field(client, mock_database):
"""Test error handling: missing required field"""
response = client.post('/api/users', json={'name': 'John'})
assert response.status_code in [400, 422] # Bad request
def test_create_user_invalid_email(client, mock_database):
"""Test validation: invalid email"""
response = client.post('/api/users', json={
'name': 'John',
'email': 'not-an-email'
})
assert response.status_code in [400, 422]Scenario 4: Flaky Test (intelligent retry with better assertions)
# Original flaky test (timing-sensitive)
def test_async_operation():
"""Test async operation completion"""
start_async_operation()
time.sleep(0.1) # Race condition!
result = get_operation_result()
assert result == 'completed'
# Agent detects flakiness and improves:
import pytest
from tenacity import retry, stop_after_attempt, wait_fixed
@retry(stop=stop_after_attempt(3), wait=wait_fixed(1))
def wait_for_operation():
"""Wait for operation with retries"""
result = get_operation_result()
if result != 'completed':
raise AssertionError("Operation not completed")
return result
def test_async_operation_improved():
"""Test async operation with proper waiting"""
start_async_operation()
# Better: Poll until complete or timeout
result = wait_for_operation()
assert result == 'completed'
# Alternative: Use pytest-asyncio for proper async testing
@pytest.mark.asyncio
async def test_async_operation_asyncio():
"""Test async operation with asyncio"""
operation = start_async_operation()
result = await operation
assert result == 'completed'Error Recovery and Fix Strategies
Fix Strategy 1: Import Errors (auto-fix imports)
# Test failure: ModuleNotFoundError: No module named 'calculator'
# Agent analyzes and fixes:
# Original generated test:
from calculator import calculate_total # WRONG PATH
# Agent detects source file location and fixes:
from app.utils.calculator import calculate_total # CORRECT PATH
# Auto-fix applied:
def fix_import_errors(test_file: Path, source_file: Path):
"""Auto-fix import paths"""
module_path = get_module_path(source_file)
test_content = test_file.read_text()
# Replace incorrect imports
test_content = test_content.replace(
f"from {source_file.stem} import",
f"from {module_path} import"
)
test_file.write_text(test_content)Fix Strategy 2: Assertion Errors (improve assertions)
# Test failure: AssertionError (no clear message)
def test_calculate_total():
result = calculate_total([1, 2, 3])
assert result == 7 # Wrong expected value
# Agent detects and suggests fix:
def test_calculate_total_improved():
items = [1, 2, 3]
result = calculate_total(items)
expected = sum(items) # Calculate expected value
assert result == expected, (
f"calculate_total({items}) returned {result}, expected {expected}"
)Fix Strategy 3: Mock Setup (auto-configure mocks)
# Test failure: AttributeError: 'MagicMock' object has no attribute 'return_value'
# Agent detects missing mock setup:
# Original (incomplete):
@patch('app.database')
def test_create_user(mock_db):
result = create_user({'name': 'John'})
# Fails: mock_db not configured
# Agent improves:
@patch('app.database')
def test_create_user_improved(mock_db):
# Agent adds proper mock configuration
mock_db.create_user.return_value = {'id': 1, 'name': 'John'}
result = create_user({'name': 'John'})
assert result['name'] == 'John'
mock_db.create_user.assert_called_once_with('John')Execution Example
Adaptive Test Generator: Starting
Target: src/app/calculator.py (3 functions)
Phase 1: Code Analysis [In Progress]
├── Parsing source code...
│ └── AST analysis: ✓ COMPLETED
├── Identifying functions...
│ ├── calculate_total(items: List[float]) -> float
│ ├── calculate_average(items: List[float]) -> float
│ └── calculate_median(items: List[float]) -> float
│ └── ✓ 3 functions identified
├── Assessing complexity...
│ ├── calculate_total: Cyclomatic 1 (simple)
│ ├── calculate_average: Cyclomatic 2 (simple)
│ └── calculate_median: Cyclomatic 4 (moderate)
│ └── ✓ Complexity assessed
└── Analyzing dependencies...
├── Imports: statistics (standard library)
└── External calls: None
└── ✓ No external dependencies
Phase 1: ✓ COMPLETED (2 minutes)
Phase 2: Test Generation [In Progress]
├── Generating tests for calculate_total...
│ ├── Happy path test: ✓ Generated
│ ├── Edge case (empty list): ✓ Generated
│ ├── Edge case (single item): ✓ Generated
│ └── Edge case (negative values): ✓ Generated
├── Generating tests for calculate_average...
│ ├── Happy path test: ✓ Generated
│ ├── Edge case (empty list): ✓ Generated (expects ZeroDivisionError)
│ ├── Edge case (single item): ✓ Generated
│ └── Edge case (all zeros): ✓ Generated
└── Generating tests for calculate_median...
├── Happy path (odd count): ✓ Generated
├── Happy path (even count): ✓ Generated
├── Edge case (empty list): ✓ Generated
├── Edge case (single item): ✓ Generated
└── Property-based test: ✓ Generated (complexity ≥ 3)
Phase 2: ✓ COMPLETED
- Tests generated: 14
- Test file: tests/test_calculator.py
- Fixtures: 0
- Mocks: 0 (no external dependencies)
- Duration: 8 minutes
Phase 3: Test Execution [In Progress]
Iteration 1:
├── Running pytest... ✓ COMPLETED
├── Results: 13/14 passed, 1 failed
├── Failures:
│ └── test_calculate_average_empty_list:
│ Expected ZeroDivisionError, got statistics.StatisticsError
├── Analyzing failure...
│ └── Root cause: Wrong exception type
├── Applying fix...
│ └── Updated: expect statistics.StatisticsError instead
└── Re-running tests...
Iteration 2:
├── Running pytest... ✓ COMPLETED
└── Results: 14/14 passed ✓
Phase 3: ✓ COMPLETED
- Total tests: 14
- Iterations: 2
- Final status: All tests passing
- Duration: 5 minutes
Phase 4: Coverage Analysis [In Progress]
├── Running pytest with coverage...
│ └── ✓ COMPLETED
├── Coverage report:
│ ├── calculate_total: 100% (4/4 lines)
│ ├── calculate_average: 100% (6/6 lines)
│ └── calculate_median: 87.5% (7/8 lines)
│ └── Overall: 94.4% ✓ (threshold: 80%)
├── Uncovered lines:
│ └── calculator.py:45 (error path in calculate_median)
└── Suggestion:
└── Add test for calculate_median with invalid input type
Phase 4: ✓ COMPLETED (3 minutes)
Test Generation Complete: ✓ SUCCESS (18 minutes)
Summary:
┌──────────────────────────────────────────────────────────────┐
│ Adaptive Test Generator - Summary │
├──────────────────────────────────────────────────────────────┤
│ Target: src/app/calculator.py │
│ Functions analyzed: 3 │
│ Tests generated: 14 │
│ Test iterations: 2 │
│ Final status: All tests passing ✓ │
│ │
│ Coverage: │
│ - Overall: 94.4% ✓ (threshold: 80%) │
│ - calculate_total: 100% │
│ - calculate_average: 100% │
│ - calculate_median: 87.5% │
│ │
│ Test Distribution: │
│ - Happy path: 5 tests │
│ - Edge cases: 8 tests │
│ - Property-based: 1 test │
│ │
│ Improvements: │
│ - Fixed 1 assertion error (wrong exception type) │
│ - Suggested 1 additional test (uncovered error path) │
│ │
│ Next Steps: │
│ - Review generated tests: tests/test_calculator.py │
│ - Run tests: pytest tests/test_calculator.py │
│ - Add suggested test for 100% coverage │
└──────────────────────────────────────────────────────────────┘Failure Scenario: Persistent Test Failures
Phase 3: Test Execution [In Progress]
Iteration 1:
├── Running pytest...
└── Results: 10/14 passed, 4 failed
├── test_api_call: ConnectionError (mock not configured)
├── test_database_query: AttributeError (mock missing method)
├── test_async_operation: TimeoutError (async not awaited)
└── test_complex_logic: AssertionError (edge case not handled)
Iteration 2:
├── Fixing mock configuration...
│ └── ✓ Mock setup improved
├── Fixing async handling...
│ └── ✓ Added pytest.mark.asyncio
├── Re-running pytest...
└── Results: 12/14 passed, 2 failed
├── test_database_query: Still failing (complex mock)
└── test_complex_logic: Still failing (algorithm bug?)
Iteration 3:
├── Analyzing complex failures...
│ ├── test_database_query: Needs multi-level mock
│ └── test_complex_logic: Possible code bug
├── Applying advanced fixes...
│ └── ✓ Multi-level mock configured
├── Re-running pytest...
└── Results: 13/14 passed, 1 failed
└── test_complex_logic: Still failing
Iteration 4:
├── Deep analysis of test_complex_logic...
│ └── Hypothesis: Algorithm bug in code (not test issue)
├── Escalating to human...
Phase 3: ⚠ PARTIAL SUCCESS (Max iterations reached)
┌──────────────────────────────────────────────────────────────┐
│ ⚠ Test Execution Incomplete │
│ │
│ Status: 13/14 tests passing (92.9%) │
│ Remaining failure: test_complex_logic │
│ │
│ Investigation: │
│ - Test appears correct │
│ - Failure is consistent (not flaky) │
│ - Possible bug in source code (calculate_median) │
│ │
│ Recommended Actions: │
│ 1. Review algorithm in calculate_median (line 40-48) │
│ 2. Debug with failing input: [1, 2, 3, 4, 5, 6] │
│ 3. Expected: 3.5, Got: 3.0 │
│ 4. Possible issue: Integer division instead of float │
│ │
│ Fix Suggestion: │
│ Replace: result = (sorted_items[mid] + sorted_items[mid+1]) / 2
│ With: result = (sorted_items[mid] + sorted_items[mid+1]) / 2.0
│ │
│ After fixing code, re-run: │
│ pytest tests/test_calculator.py::test_complex_logic │
└──────────────────────────────────────────────────────────────┘Lessons Learned
Benefits Realized:
1. Time savings: 18 minutes automated vs 30-60 minutes manual 2. Coverage: 94.4% achieved automatically (better than manual) 3. Edge cases: 8 edge case tests generated (often missed manually) 4. Adaptive: Different strategies for simple vs complex code 5. Self-fixing: Fixed 1 assertion error automatically
Challenges Encountered:
1. Complex mocks: Multi-level mocking required manual intervention 2. Code bugs: Test revealed algorithm bug (not test issue) 3. Async testing: Required pytest-asyncio (dependency management) 4. Property-based tests: Hypothesis integration added complexity
Philosophy Compliance:
- Ruthless simplicity: 4 clear phases, straightforward logic
- Single responsibility: Each phase focused (analyze, generate, execute, validate)
- Modularity: Test generators (unit, property-based, integration) are reusable
- Regeneratable: Can rebuild tests from code analysis
When to Use This Pattern:
- New features requiring comprehensive tests
- Legacy code lacking test coverage
- Refactoring (ensure behavior preserved)
- Flaky test investigation and fixing
- CI failures from missing tests
When NOT to Use:
- Simple one-liner functions (manual test faster)
- UI tests (requires different approach)
- Performance tests (needs benchmarking framework)
- Security tests (specialized tools required)
- Tests already exist and are comprehensive
Example: Data Pipeline with Goal-Seeking Agents
Scenario: Multi-Source Data Pipeline Automation
Problem Statement
Manual data pipeline operations are:
- Labor-intensive: 2-3 hours to collect, transform, validate, and publish data
- Source-dependent: Different ingestion strategies for S3, databases, APIs
- Fragile: Source unavailability breaks entire pipeline
- Quality-risky: Data quality issues discovered late (after publishing)
- Non-idempotent: Re-running after failures causes duplicates
Is Goal-Seeking Appropriate?
Apply the 5-question decision framework:
Q1: Well-defined objective but flexible path?
- YES: Objective is clear (ingest, transform, validate, publish data)
- Multiple paths:
- All sources available: Parallel collection
- Some sources unavailable: Partial collection with logging
- Quality issues: Iterative cleansing
- Success criteria: Data in warehouse, quality thresholds met
Q2: Multiple phases with dependencies?
- YES: 4 phases with clear dependencies
1. Data Collection (parallel: S3, database, API) 2. Data Transformation (depends on collection) 3. Quality Validation (depends on transformation) 4. Data Publishing (depends on validation)
Q3: Autonomous recovery valuable?
- YES: Failures are common and recoverable
- Source timeouts: Retry with backoff
- Transformation errors: Log and skip bad records
- Quality failures: Apply automated cleansing
- Publishing errors: Retry with exponential backoff
Q4: Context affects approach?
- YES: Strategy varies by:
- Data volume (100K vs 10M records)
- Source availability (all vs partial)
- Quality requirements (strict vs lenient)
- Time constraints (daily batch vs real-time)
Q5: Complexity justified?
- YES: High-value automation
- Frequency: Daily (365 times per year)
- Manual time: 2-3 hours per run
- Value: 730-1095 hours saved per year
- Risk reduction: Fewer human errors
Conclusion: All 5 YES → Goal-seeking agent is appropriate
Goal-Seeking Agent Design
Goal Definition
# Goal: Automate Multi-Source Data Pipeline
## Objective
Collect data from multiple sources (S3 buckets, PostgreSQL database, REST API),
transform to common schema, validate quality, and publish to data warehouse.
## Success Criteria
- All available sources successfully ingested (100% success or logged failures)
- Data transformed to target schema with < 2% transformation failure rate
- Quality checks pass: completeness ≥ 95%, accuracy ≥ 98%, consistency = 100%
- Data published to warehouse without duplicates
- Pipeline completes within 30 minutes
## Constraints
- Must handle source unavailability gracefully (log and continue)
- No data loss (failed records logged for manual review)
- Idempotent (safe to re-run without duplicates)
- Resource limits: 8GB RAM, 4 CPU cores
- Must preserve data lineage (track source → warehouse)
## Context
- Frequency: Daily (automated at 2 AM)
- Priority: High (blocking downstream analytics)
- Scale: Medium (100K-1M records per source)
- Sources: 3 (S3, PostgreSQL, REST API)Execution Plan
from amplihack.goal_agent_generator import PromptAnalyzer, ObjectivePlanner
analyzer = PromptAnalyzer()
goal_def = analyzer.analyze_text(goal_text)
planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_def)
# Result: 4-phase plan with parallel collectionPhase 1: Data Collection (15 minutes, parallel-safe)
- Collect from S3 (100K-500K records, parallel)
- Collect from PostgreSQL (50K-200K records, parallel)
- Collect from REST API (10K-100K records, parallel)
- Handle source failures gracefully
- Log collection metrics
Dependencies: None (all sources collected in parallel) Success indicators:
- All sources attempted
- Successful collections logged
- Failed collections logged with reason
- Raw data stored in staging area
Phase 2: Data Transformation (15 minutes, depends on Phase 1)
- Parse raw data formats (JSON, CSV, Parquet)
- Transform to common schema
- Handle missing fields (apply defaults)
- Normalize data types
- Enrich with metadata
Dependencies: Phase 1 (needs collected data) Success indicators:
- All records attempted transformation
- Success rate ≥ 98%
- Failed records logged
- Transformed data in staging schema
Phase 3: Quality Validation (5 minutes, depends on Phase 2)
- Completeness check (required fields present)
- Accuracy validation (data types, ranges, formats)
- Consistency verification (referential integrity)
- Duplicate detection
- Anomaly detection
Dependencies: Phase 2 (needs transformed data) Success indicators:
- Completeness ≥ 95%
- Accuracy ≥ 98%
- Consistency = 100%
- Duplicates identified and removed
- Anomalies flagged
Phase 4: Data Publishing (10 minutes, depends on Phase 3)
- Load to data warehouse (bulk insert)
- Update metadata tables
- Create data lineage records
- Generate quality report
- Archive raw data
Dependencies: Phase 3 (only publish validated data) Success indicators:
- All validated data in warehouse
- No duplicates
- Metadata updated
- Lineage tracked
- Report generated
Total Duration: 45 minutes (estimated with overhead)
Implementation
from amplihack.goal_agent_generator import (
PromptAnalyzer,
ObjectivePlanner,
SkillSynthesizer,
AgentAssembler,
GoalAgentPackager,
)
from pathlib import Path
import asyncio
from datetime import datetime
from typing import List, Dict, Any
# Goal definition
goal_text = """
Automate multi-source data pipeline:
- Collect from S3, PostgreSQL, REST API (parallel)
- Transform to common schema
- Validate quality (completeness, accuracy, consistency)
- Publish to data warehouse
Handle source failures gracefully, ensure idempotency.
"""
# Create agent
analyzer = PromptAnalyzer()
goal_def = analyzer.analyze_text(goal_text)
planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_def)
synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)
assembler = AgentAssembler()
agent_bundle = assembler.assemble(
goal_definition=goal_def,
execution_plan=execution_plan,
skills=skills,
bundle_name="multi-source-data-pipeline"
)
packager = GoalAgentPackager()
packager.package(
bundle=agent_bundle,
output_dir=Path(".claude/agents/goal-driven/data-pipeline")
)Adaptive Behavior
The agent adapts to different conditions:
Scenario 1: All Sources Available (optimal path)
async def collect_all_sources():
"""Parallel collection when all sources available"""
results = await asyncio.gather(
collect_from_s3(),
collect_from_postgresql(),
collect_from_api(),
return_exceptions=True # Don't fail if one source fails
)
collected_data = []
for source, result in zip(["S3", "PostgreSQL", "API"], results):
if isinstance(result, Exception):
log_source_failure(source, result)
else:
collected_data.append(result)
return collected_dataScenario 2: Source Unavailable (graceful degradation)
async def collect_from_api_with_fallback():
"""Handle API unavailability"""
try:
# Try primary API endpoint
data = await fetch_from_api(primary_endpoint)
return data
except ConnectionError:
try:
# Try fallback endpoint
log_warning("Primary API unavailable, trying fallback")
data = await fetch_from_api(fallback_endpoint)
return data
except ConnectionError:
# Log failure and continue with partial data
log_error("API completely unavailable")
send_alert("Data pipeline: API source unavailable")
return None # Continue with S3 and PostgreSQL dataScenario 3: Large Data Volume (resource optimization)
def transform_data_adaptive(data: List[Dict], volume: int):
"""Adapt transformation strategy based on volume"""
if volume < 100_000:
# Small volume: In-memory transformation
return transform_in_memory(data)
elif volume < 1_000_000:
# Medium volume: Chunked processing
return transform_in_chunks(data, chunk_size=10_000)
else:
# Large volume: Distributed processing
return transform_distributed(data, num_workers=4)Scenario 4: Quality Issues (iterative cleansing)
def validate_and_cleanse(data: List[Dict]) -> List[Dict]:
"""Iterative quality improvement"""
quality_score = calculate_quality(data)
while quality_score < QUALITY_THRESHOLD:
# Apply automated cleansing
data = apply_cleansing_rules(data)
# Recalculate quality
quality_score = calculate_quality(data)
if cleansing_iterations >= MAX_ITERATIONS:
# Escalate if can't meet quality threshold
escalate(
reason="Quality threshold not met after cleansing",
current_quality=quality_score,
threshold=QUALITY_THRESHOLD,
recommendation="Manual data review required"
)
break
return dataError Recovery Strategies
Strategy 1: Retry with Exponential Backoff (transient errors)
import time
from typing import Callable, Any
def retry_with_backoff(
func: Callable[[], Any],
max_attempts: int = 3,
initial_delay: float = 1.0
) -> Any:
"""Retry function with exponential backoff"""
for attempt in range(max_attempts):
try:
return func()
except (ConnectionError, TimeoutError) as e:
if attempt == max_attempts - 1:
raise # Last attempt, re-raise
delay = initial_delay * (2 ** attempt)
log_warning(f"Attempt {attempt + 1} failed, retrying in {delay}s")
time.sleep(delay)
# Usage
s3_data = retry_with_backoff(
lambda: collect_from_s3(),
max_attempts=3,
initial_delay=2.0
)Strategy 2: Partial Success (continue with available data)
def handle_partial_collection(results: List[Any]) -> Dict[str, Any]:
"""Process partial collection results"""
successful_sources = [r for r in results if r is not None]
failed_sources = [i for i, r in enumerate(results) if r is None]
if not successful_sources:
escalate("All data sources failed, cannot proceed")
if failed_sources:
log_warning(f"Partial collection: {len(successful_sources)}/{len(results)} sources")
send_alert(f"Data pipeline: {len(failed_sources)} sources unavailable")
# Continue with available data
return {
"data": successful_sources,
"partial": len(failed_sources) > 0,
"failed_sources": failed_sources
}Strategy 3: Idempotent Execution (safe re-runs)
def publish_to_warehouse_idempotent(data: List[Dict], run_id: str):
"""Ensure idempotent publishing (no duplicates on re-run)"""
# Check if this run already succeeded
if check_run_completed(run_id):
log_info(f"Run {run_id} already completed, skipping")
return
# Use transaction for atomicity
with warehouse_transaction() as txn:
# Delete any partial data from previous failed runs
txn.execute(f"DELETE FROM target_table WHERE run_id = '{run_id}'")
# Insert new data with run_id
for record in data:
record["run_id"] = run_id
record["ingested_at"] = datetime.utcnow()
txn.insert("target_table", record)
# Mark run as completed
txn.insert("pipeline_runs", {
"run_id": run_id,
"status": "completed",
"record_count": len(data),
"completed_at": datetime.utcnow()
})
txn.commit()Execution Example
Multi-Source Data Pipeline: Starting (run_id: 20251116_020000)
Phase 1: Data Collection [In Progress]
├── S3 Collection (parallel)...
│ ├── Bucket: data-lake/raw/2025/11/16
│ ├── Files: 15 Parquet files
│ └── Records: 250,000
│ └── Duration: 8 minutes
│ └── ✓ COMPLETED
├── PostgreSQL Collection (parallel)...
│ ├── Table: transactions_2025_11_16
│ ├── Query: SELECT * WHERE date = '2025-11-16'
│ └── Records: 75,000
│ └── Duration: 6 minutes
│ └── ✓ COMPLETED
└── API Collection (parallel)...
├── Endpoint: https://api.example.com/events?date=2025-11-16
├── Attempt 1: ✗ Timeout
├── Retry in 2s...
├── Attempt 2: ✓ Success
└── Records: 15,000
└── Duration: 4 minutes (1 retry)
└── ✓ COMPLETED
Phase 1: ✓ COMPLETED
- Total records: 340,000
- Sources: 3/3 successful (1 retry)
- Duration: 10 minutes (parallel execution)
Phase 2: Data Transformation [In Progress]
├── Parsing formats...
│ ├── S3 (Parquet): ✓ 250,000 records parsed
│ ├── PostgreSQL (JSON): ✓ 75,000 records parsed
│ └── API (JSON): ✓ 15,000 records parsed
├── Schema transformation...
│ ├── Field mapping: ✓ All fields mapped
│ ├── Type conversion: ✓ Completed (12 type conversions)
│ ├── Default values: ✓ Applied to 2,500 records (0.7%)
│ └── Success rate: 98.5% (5,100 failures logged)
├── Data enrichment...
│ └── Metadata added: source, ingestion_time, run_id
└── Staging schema...
└── ✓ 334,900 records in staging
Phase 2: ✓ COMPLETED
- Total transformed: 334,900 / 340,000 (98.5%)
- Failed transformations: 5,100 (logged to failed_records.log)
- Duration: 14 minutes
Phase 3: Quality Validation [In Progress]
├── Completeness check...
│ ├── Required fields: user_id, event_type, timestamp, amount
│ ├── Missing fields: 3,200 records (0.96%)
│ ├── Completeness: 99.04% ✓ (threshold: 95%)
│ └── ✓ PASS
├── Accuracy check...
│ ├── Data type validation: ✓ 100% correct types
│ ├── Range validation: ✗ 1,500 records out of range
│ │ └── Applying automated correction...
│ │ └── ✓ 1,200 corrected, 300 flagged
│ ├── Format validation: ✓ 100% correct formats
│ └── Accuracy: 99.91% ✓ (threshold: 98%)
├── Consistency check...
│ ├── Referential integrity: ✓ 100%
│ ├── No orphan records: ✓ Verified
│ └── ✓ PASS
├── Duplicate detection...
│ ├── Duplicate records found: 1,800
│ └── ✓ Duplicates removed
└── Anomaly detection...
├── Statistical outliers: 250 flagged
└── ✓ Flagged for review
Phase 3: ✓ COMPLETED
- Final record count: 333,100 (after deduplication)
- Completeness: 99.04% ✓
- Accuracy: 99.91% ✓
- Consistency: 100% ✓
- Duration: 5 minutes
Phase 4: Data Publishing [In Progress]
├── Warehouse load...
│ ├── Target: warehouse.analytics.events
│ ├── Method: Bulk insert (batch size: 10,000)
│ ├── Records: 333,100
│ └── Duration: 6 minutes
│ └── ✓ COMPLETED
├── Metadata update...
│ ├── Table: pipeline_metadata
│ ├── Run ID: 20251116_020000
│ └── ✓ COMPLETED
├── Data lineage...
│ ├── Source → Warehouse mapping created
│ ├── Lineage records: 3 (S3, PostgreSQL, API)
│ └── ✓ COMPLETED
├── Quality report...
│ ├── File: reports/quality_20251116_020000.html
│ ├── Summary: 333,100 records published, 5,100 transformation failures
│ └── ✓ COMPLETED
└── Archive raw data...
├── Location: archive/2025/11/16/
└── ✓ COMPLETED
Phase 4: ✓ COMPLETED (9 minutes)
Pipeline Execution: ✓ SUCCESS (38 minutes total)
Summary Report:
┌──────────────────────────────────────────────────────────────┐
│ Multi-Source Data Pipeline - Execution Summary │
├──────────────────────────────────────────────────────────────┤
│ Run ID: 20251116_020000 │
│ Status: SUCCESS │
│ Duration: 38 minutes │
│ │
│ Data Collection: │
│ - S3: 250,000 records (8 min) │
│ - PostgreSQL: 75,000 records (6 min) │
│ - API: 15,000 records (4 min, 1 retry) │
│ - Total: 340,000 records │
│ │
│ Data Transformation: │
│ - Transformed: 334,900 records (98.5%) │
│ - Failed: 5,100 records (1.5%, logged) │
│ │
│ Quality Validation: │
│ - Completeness: 99.04% ✓ │
│ - Accuracy: 99.91% ✓ │
│ - Consistency: 100% ✓ │
│ - Duplicates: 1,800 removed │
│ │
│ Data Publishing: │
│ - Warehouse: 333,100 records │
│ - Metadata: Updated │
│ - Lineage: Tracked │
│ - Report: Generated │
│ │
│ Issues: │
│ - API timeout (1 retry, resolved) │
│ - 5,100 transformation failures (logged) │
│ - 300 out-of-range values (flagged) │
│ - 250 anomalies (flagged for review) │
│ │
│ Next Steps: │
│ - Review failed transformations (failed_records.log) │
│ - Investigate out-of-range values │
│ - Review anomalies for false positives │
└──────────────────────────────────────────────────────────────┘Failure Scenario: Source Completely Unavailable
Phase 1: Data Collection [In Progress]
├── S3 Collection... ✓ COMPLETED (250,000 records, 8 min)
├── PostgreSQL Collection... ✓ COMPLETED (75,000 records, 6 min)
└── API Collection...
├── Attempt 1: ✗ Connection refused
├── Retry in 2s...
├── Attempt 2: ✗ Connection refused
├── Retry in 4s...
├── Attempt 3: ✗ Connection refused
└── ✗ FAILED (all retries exhausted)
└── Logging failure: API unavailable
└── Sending alert: Data pipeline - API source down
└── Continuing with partial data (S3 + PostgreSQL)
Phase 1: ⚠ COMPLETED WITH WARNINGS
- Total records: 325,000 (2/3 sources)
- Failed sources: API (connection refused)
- Duration: 10 minutes
[Pipeline continues with S3 + PostgreSQL data]
Summary Report:
┌──────────────────────────────────────────────────────────────┐
│ ⚠ PARTIAL SUCCESS │
│ │
│ Data collected from 2/3 sources (API unavailable) │
│ Published: 318,500 records │
│ Missing: ~15,000 records (API source) │
│ │
│ Action Required: │
│ - Investigate API connectivity │
│ - Re-run pipeline for API data once resolved │
│ - Command: amplihack goal-agent-generator execute \ │
│ --agent data-pipeline \ │
│ --source-filter API \ │
│ --date 2025-11-16 │
└──────────────────────────────────────────────────────────────┘Lessons Learned
Benefits Realized:
1. Time savings: 38 minutes automated vs 2-3 hours manual 2. Resilience: Handles source failures gracefully (partial success) 3. Quality: Automated validation catches 99% of issues 4. Idempotency: Safe to re-run without duplicates 5. Observability: Detailed logging and quality reports
Challenges Encountered:
1. Resource tuning: Initial 8GB RAM insufficient for 1M records, increased to 12GB 2. Quality thresholds: Initial 100% completeness too strict, relaxed to 95% 3. API reliability: Frequent timeouts required fallback endpoint 4. Duplicate detection: Needed to add run_id for proper deduplication
Philosophy Compliance:
- Ruthless simplicity: 4 clear phases, no unnecessary steps
- Single responsibility: Each phase has one job (collect, transform, validate, publish)
- Modularity: Collectors (S3, PostgreSQL, API) are reusable
- Regeneratable: Can rebuild pipeline from goal definition
When to Use This Pattern:
- Multiple data sources with different collection strategies
- Quality validation is critical
- Source failures are common (need graceful handling)
- Idempotency is required (safe re-runs)
- High frequency (daily or more)
When NOT to Use:
- Single data source (simple script suffices)
- Real-time streaming (use streaming framework)
- No quality requirements (direct load acceptable)
- One-time data migration (script sufficient)
Example: Workflow Automation with Goal-Seeking Agents
Scenario: Release Workflow Automation
Problem Statement
Manual release workflows are:
- Time-consuming: 2-3 hours per release
- Error-prone: Easy to miss steps (tagging, changelogs, notifications)
- Context-dependent: Different steps for hotfixes vs features
- Environment-specific: Staging and production have different requirements
- Recovery-intensive: Failures require manual rollback
Is Goal-Seeking Appropriate?
Apply the 5-question decision framework:
Q1: Well-defined objective but flexible path?
- YES: Objective is clear (release software to production)
- Multiple paths:
- Hotfix: Skip QA staging, direct to production
- Feature: Full staging validation, gradual rollout
- Patch: Automated tests only, fast-track
- Success criteria: Deployed, tested, monitored
Q2: Multiple phases with dependencies?
- YES: 5 phases with clear dependencies
1. Pre-release validation (tests, lint, security scan) 2. Artifact creation (build, tag, changelog) 3. Staging deployment (deploy, smoke tests) 4. Production deployment (gradual rollout, health checks) 5. Post-release (monitoring, notifications, documentation)
Q3: Autonomous recovery valuable?
- YES: Failures are common and recoverable
- Build failures: Retry with clean cache
- Deployment failures: Rollback to previous version
- Test failures: Re-run flaky tests
- Monitoring issues: Wait and retry
- Human intervention is slow (especially after-hours)
Q4: Context affects approach?
- YES: Release strategy varies by:
- Change type (hotfix/feature/patch)
- Environment (staging/production)
- System state (healthy/degraded)
- Time of day (business hours/off-hours)
Q5: Complexity justified?
- YES: Problem is frequent and valuable
- Frequency: 2-3 releases per week
- Manual time: 2-3 hours per release
- Value: 4-6 hours saved per week
- High value: Reduces release anxiety, enables more frequent releases
Conclusion: All 5 YES → Goal-seeking agent is appropriate
Goal-Seeking Agent Design
Goal Definition
# Goal: Automate Software Release Workflow
## Objective
Execute end-to-end release workflow from code freeze to production deployment,
adapting strategy based on change type (hotfix/feature/patch) and environment health.
## Success Criteria
- All pre-release validations pass (tests, lint, security)
- Artifacts created and versioned (Docker image, Git tag, changelog)
- Staging deployment successful with smoke tests passing
- Production deployment completes with health checks passing
- Post-release monitoring shows no regressions
- Stakeholders notified of release status
## Constraints
- Zero downtime for production deployment
- Rollback capability at any phase
- Must complete within 60 minutes
- Security scans must pass before production
- Requires human approval for production (no auto-deploy)
## Context
- Frequency: 2-3 times per week
- Priority: High (blocking features/fixes)
- Scale: Medium (10-20 microservices)Execution Plan
from amplihack.goal_agent_generator import ObjectivePlanner, PromptAnalyzer
# Analyze goal
analyzer = PromptAnalyzer()
goal_def = analyzer.analyze_text(goal_text)
# Generate plan
planner = ObjectivePlanner()
plan = planner.generate_plan(goal_def)
# Result: 5-phase execution planPhase 1: Pre-Release Validation (15 minutes, parallel)
- Run test suite (unit, integration, e2e)
- Execute linters and formatters
- Run security vulnerability scan
- Validate API contracts
- Check database migration scripts
Dependencies: None Parallel-safe: Yes (tests/lint/security can run concurrently) Success indicators:
- All tests pass (100%)
- No linting violations
- No security vulnerabilities (or approved exceptions)
- API contracts compatible
- Migrations validated
Phase 2: Artifact Creation (10 minutes, depends on Phase 1)
- Build Docker images
- Create Git tag (semantic versioning)
- Generate changelog from commits
- Package release notes
- Upload artifacts to registry
Dependencies: Phase 1 (validation must pass) Parallel-safe: No (artifacts depend on validation) Success indicators:
- Docker images built and pushed
- Git tag created
- Changelog generated
- Artifacts in registry
Phase 3: Staging Deployment (15 minutes, depends on Phase 2)
- Deploy to staging environment
- Run smoke tests
- Execute integration tests
- Monitor metrics for 5 minutes
- Verify health checks
Dependencies: Phase 2 (artifacts must exist) Parallel-safe: No (production waits for staging validation) Success indicators:
- Staging deployment successful
- Smoke tests pass
- Integration tests pass
- Metrics within normal range
- Health checks passing
Phase 4: Production Deployment (15 minutes, depends on Phase 3)
- Gradual rollout (10% → 50% → 100%)
- Monitor metrics at each stage
- Verify health checks continuously
- Run production smoke tests
- Confirm zero errors
Dependencies: Phase 3 (staging must pass) Parallel-safe: No (production is final critical phase) Success indicators:
- Gradual rollout completes
- No error rate increase
- Latency within SLA
- Health checks passing
- Smoke tests pass
Phase 5: Post-Release (5 minutes, depends on Phase 4)
- Update documentation
- Send notifications (Slack, email)
- Create release announcement
- Update monitoring dashboards
- Archive release artifacts
Dependencies: Phase 4 (deployment must succeed) Parallel-safe: Yes (documentation/notifications can run concurrently) Success indicators:
- Documentation updated
- Stakeholders notified
- Release announcement published
- Dashboards updated
Total Duration: 60 minutes (estimated)
Implementation
from amplihack.goal_agent_generator import (
PromptAnalyzer,
ObjectivePlanner,
SkillSynthesizer,
AgentAssembler,
GoalAgentPackager,
)
from pathlib import Path
# Step 1: Define goal
goal_text = """
Automate software release workflow:
- Pre-release validation (tests, lint, security)
- Artifact creation (build, tag, changelog)
- Staging deployment with smoke tests
- Production deployment with gradual rollout
- Post-release monitoring and notifications
Adapt strategy based on change type (hotfix/feature/patch).
"""
# Step 2: Analyze and plan
analyzer = PromptAnalyzer()
goal_def = analyzer.analyze_text(goal_text)
planner = ObjectivePlanner()
execution_plan = planner.generate_plan(goal_def)
# Step 3: Synthesize skills
synthesizer = SkillSynthesizer()
skills = synthesizer.synthesize(execution_plan)
# Result: 5 skills identified
# - validator: Pre-release validation
# - builder: Artifact creation
# - deployer: Staging/production deployment
# - monitor: Health checks and metrics
# - documenter: Post-release tasks
# Step 4: Assemble agent
assembler = AgentAssembler()
agent_bundle = assembler.assemble(
goal_definition=goal_def,
execution_plan=execution_plan,
skills=skills,
bundle_name="release-workflow-agent"
)
# Step 5: Package for deployment
packager = GoalAgentPackager()
packager.package(
bundle=agent_bundle,
output_dir=Path(".claude/agents/goal-driven/release-workflow-agent")
)
print(f"Agent created: {agent_bundle.name}")
print(f"Phases: {len(execution_plan.phases)}")
print(f"Skills: {[s.name for s in skills]}")
print(f"Estimated duration: {execution_plan.total_estimated_duration}")Adaptive Behavior
The agent adapts based on context:
Hotfix Release (urgent bug fix):
# Phase 1: Minimal validation (skip long-running tests)
if release_type == "hotfix":
validation_scope = "critical-tests-only"
# Run only tests related to fix
# Skip full integration suite
# Phase 3: Skip staging (go direct to production)
if release_type == "hotfix" and severity == "critical":
skip_staging = True
# Deploy directly to production with extra monitoring
# Phase 4: Faster rollout (urgency vs safety trade-off)
if release_type == "hotfix":
rollout_schedule = [50, 100] # Skip 10% stageFeature Release (standard):
# Phase 1: Full validation
if release_type == "feature":
validation_scope = "comprehensive"
# Run all tests, security scans, performance tests
# Phase 3: Full staging validation
if release_type == "feature":
staging_duration = 15 # minutes
# Run extensive smoke tests and integration tests
# Phase 4: Gradual rollout
if release_type == "feature":
rollout_schedule = [10, 25, 50, 100] # Cautious rolloutDegraded Environment (system issues):
# If environment is degraded, delay release
if environment_health < HEALTH_THRESHOLD:
escalate(
reason="Environment health below threshold",
current_health=environment_health,
recommendation="Wait for health to improve or proceed with extra caution"
)
# If proceeding, add extra monitoring
if user_approves_risky_release:
monitoring_intensity = "high"
rollback_readiness = "immediate"Error Recovery
The agent implements three recovery strategies:
Strategy 1: Retry with Backoff (transient failures)
# Build failures (network issues, resource contention)
@retry(max_attempts=3, backoff=exponential)
def build_artifacts():
try:
docker_build()
docker_push()
except NetworkError:
# Retry automatically
raise
except DiskSpaceError:
# Not transient, escalate
escalate("Disk space exhausted, cannot build")Strategy 2: Alternative Strategy (approach failures)
# Deployment strategy failures
def deploy_to_production():
strategies = [
gradual_rollout_strategy,
blue_green_deployment_strategy,
rolling_update_strategy
]
for strategy in strategies:
try:
return strategy.execute()
except StrategyFailedError as e:
log_failure(strategy, e)
continue # Try next strategy
escalate("All deployment strategies failed")Strategy 3: Rollback (safety mechanism)
# Automatic rollback on health check failures
def monitor_deployment_health():
for check in health_checks:
if not check.passing():
# Initiate automatic rollback
rollback_deployment()
escalate(
reason="Health checks failing after deployment",
failed_checks=[check.name for check in health_checks if not check.passing()],
action_taken="Automatic rollback completed"
)Execution Example
Release Workflow Agent: Starting
Phase 1: Pre-Release Validation [In Progress]
├── Running test suite... ✓ COMPLETED (458 tests, 0 failures, 12 minutes)
├── Executing linters... ✓ COMPLETED (0 violations, 2 minutes)
├── Security scan... ✗ FAILED (1 medium vulnerability found)
│ └── Analyzing vulnerability...
│ └── Checking for approved exceptions...
│ └── Found approved exception (CVE-2023-12345, expires 2025-12-01)
│ └── ✓ PASS (approved exception)
└── Validating migrations... ✓ COMPLETED (3 migrations validated, 1 minute)
Phase 1: ✓ COMPLETED (15 minutes)
Phase 2: Artifact Creation [In Progress]
├── Building Docker image... ✓ COMPLETED (image: app:v1.2.3, 6 minutes)
├── Creating Git tag... ✓ COMPLETED (tag: v1.2.3, 10 seconds)
├── Generating changelog... ✓ COMPLETED (15 commits since v1.2.2, 30 seconds)
└── Pushing to registry... ✓ COMPLETED (registry.example.com/app:v1.2.3, 3 minutes)
Phase 2: ✓ COMPLETED (10 minutes)
Phase 3: Staging Deployment [In Progress]
├── Deploying to staging... ✓ COMPLETED (3 replicas, 4 minutes)
├── Running smoke tests... ✓ COMPLETED (12/12 tests passed, 3 minutes)
├── Integration tests... ✓ COMPLETED (25/25 tests passed, 5 minutes)
└── Monitoring metrics... ✓ PASS (error rate: 0%, latency: 45ms avg, 3 minutes)
Phase 3: ✓ COMPLETED (15 minutes)
Phase 4: Production Deployment [HUMAN APPROVAL REQUIRED]
┌─────────────────────────────────────────────────────────────────┐
│ Production Deployment Ready │
│ │
│ Release: v1.2.3 │
│ Changes: 15 commits (3 features, 2 bug fixes) │
│ Staging: All tests passed │
│ Health: Environment healthy │
│ │
│ Rollout Plan: │
│ - 10% of traffic (2 minutes) │
│ - 50% of traffic (5 minutes) │
│ - 100% of traffic (final) │
│ │
│ Approve deployment? (yes/no) │
└─────────────────────────────────────────────────────────────────┘
User: yes
Phase 4: Production Deployment [In Progress]
├── Rollout stage 1 (10%)... ✓ COMPLETED (1 replica, 2 minutes)
│ └── Metrics: error rate 0%, latency 48ms avg
├── Rollout stage 2 (50%)... ✓ COMPLETED (5 replicas, 5 minutes)
│ └── Metrics: error rate 0%, latency 46ms avg
└── Rollout stage 3 (100%)... ✓ COMPLETED (10 replicas, 5 minutes)
└── Metrics: error rate 0%, latency 47ms avg
Phase 4: ✓ COMPLETED (12 minutes)
Phase 5: Post-Release [In Progress]
├── Updating documentation... ✓ COMPLETED (CHANGELOG.md, README.md, 1 minute)
├── Sending notifications...
│ ├── Slack (#releases): ✓ SENT
│ └── Email (stakeholders): ✓ SENT
├── Creating announcement... ✓ COMPLETED (blog post draft, 1 minute)
└── Updating dashboards... ✓ COMPLETED (Grafana dashboard updated, 30 seconds)
Phase 5: ✓ COMPLETED (3 minutes)
Release Workflow: ✓ SUCCESS (55 minutes total)
Summary:
- Version: v1.2.3
- Changes: 15 commits (3 features, 2 bug fixes)
- Tests: 495 passed, 0 failures
- Staging: All smoke and integration tests passed
- Production: Gradual rollout completed, all metrics healthy
- Documentation: Updated and stakeholders notified
Next steps:
- Monitor production for next 24 hours
- Address any user-reported issues
- Plan next releaseFailure Scenario: Staging Tests Fail
Phase 3: Staging Deployment [In Progress]
├── Deploying to staging... ✓ COMPLETED (3 replicas, 4 minutes)
├── Running smoke tests... ✗ FAILED (2/12 tests failed)
│ └── Failed tests:
│ - test_user_login: Connection timeout
│ - test_payment_flow: 500 Internal Server Error
│ └── Diagnosing failures...
│ └── Root cause: Database connection pool exhausted
│ └── Applying fix: Increasing connection pool size
│ └── Redeploying with fix...
│ └── Re-running smoke tests... ✓ COMPLETED (12/12 tests passed, 3 minutes)
├── Integration tests... ✓ COMPLETED (25/25 tests passed, 5 minutes)
└── Monitoring metrics... ✓ PASS (error rate: 0%, latency: 45ms avg, 3 minutes)
Phase 3: ✓ COMPLETED (18 minutes, 1 retry)Failure Scenario: Production Rollout Issues
Phase 4: Production Deployment [In Progress]
├── Rollout stage 1 (10%)... ✓ COMPLETED (1 replica, 2 minutes)
│ └── Metrics: error rate 0%, latency 48ms avg
├── Rollout stage 2 (50%)... ✗ FAILED (error rate spike detected)
│ └── Error rate: 5% (threshold: 1%)
│ └── Initiating automatic rollback...
│ └── Rollback completed: All traffic on v1.2.2
│ └── Investigating root cause...
│ └── Root cause: Missing environment variable in production config
│ └── Escalating to human for investigation
Phase 4: ✗ FAILED (5 minutes, automatic rollback completed)
┌─────────────────────────────────────────────────────────────────┐
│ Production Deployment Failed - Rolled Back │
│ │
│ Release: v1.2.3 │
│ Failure: Error rate spike during 50% rollout │
│ Root Cause: Missing environment variable (DB_POOL_SIZE) │
│ Action Taken: Automatic rollback to v1.2.2 │
│ │
│ Current State: │
│ - Production: Running v1.2.2 (stable) │
│ - Staging: Running v1.2.3 (working) │
│ - Error rate: 0% (normal) │
│ │
│ Recommended Actions: │
│ 1. Add DB_POOL_SIZE to production environment config │
│ 2. Re-run staging tests with production-like config │
│ 3. Retry deployment after fix │
│ │
│ Resume command: │
│ amplihack goal-agent-generator execute \ │
│ --agent-path .claude/agents/goal-driven/release-workflow-agent \
│ --resume-from-phase 4 │
└─────────────────────────────────────────────────────────────────┘Lessons Learned
Benefits Realized:
1. Time savings: 55 minutes automated vs 2-3 hours manual 2. Consistency: Same workflow every time, no missed steps 3. Autonomous recovery: Automatic retries and rollbacks 4. Adaptability: Different strategies for hotfixes vs features 5. Safety: Human approval gate for production, automatic rollback on failures
Challenges Encountered:
1. Initial setup: 4 hours to define goal, phases, and error handling 2. Testing: Needed to test failure scenarios (staging failures, production rollbacks) 3. Environment differences: Staging and production configs diverged 4. Monitoring integration: Required connecting to multiple monitoring systems
Philosophy Compliance:
- Ruthless simplicity: 5 clear phases, no unnecessary complexity
- Single responsibility: Each phase has one job (validate, build, deploy, monitor, notify)
- Modularity: Skills (validator, builder, deployer, monitor, documenter) are reusable
- Regeneratable: Can rebuild agent from goal definition
When to Use This Pattern:
- Repeated frequently (2-3+ times per week)
- High value from automation (hours saved, reduced errors)
- Multiple execution paths (hotfix, feature, patch)
- Recoverable failures (retry, alternative strategies, rollback)
- Clear success criteria (tests pass, metrics healthy, stakeholders notified)
When NOT to Use:
- One-time releases (simple script suffices)
- Fully manual process required (compliance, audit)
- Execution time is negligible (< 10 minutes manual)
- No variation in workflow (same steps every time)
- Failures always require human investigation