"""Phase A: Multi-Step Action Chain Engine.
Executes chains of actions with conditional branching, wait delays, and error handling.
Each chain is stored as a single action with type='chain' and config.steps array.
Chain execution model:
- Steps reference each other by ID
- Each step routes to next step(s) via on_success, on_failure, on_condition
- Wait steps use RQ enqueue_at for delayed execution
- Conditional branching reuses evaluate_condition() from form_logic
- Parallel fan-out supported via multiple on_success targets
Example chain config:
{
"steps": [
{
"id": "step_1",
"type": "http_request",
"config": { "method": "POST", "url": "https://...", "headers": {} },
"on_success": "step_2",
"on_failure": "fallback_email"
},
{
"id": "step_2",
"type": "wait",
"config": { "duration_seconds": 3600 },
"on_success": "step_3"
},
{
"id": "step_3",
"type": "email",
"config": { "recipients": ["{{email}}"], "body": "..." }
}
]
}
"""
import json
import logging
import os
import re
from datetime import UTC, datetime, timedelta, timezone
from typing import Any, Optional
logger = logging.getLogger("agentforms.chain")
logger.setLevel(logging.INFO)
if not logger.handlers:
_h = logging.StreamHandler(__import__("sys").stderr)
_h.setFormatter(logging.Formatter("[chain] %(message)s"))
logger.addHandler(_h)
# ─── Template variable substitution ─────────────────────────────────────────────
def _interpolate(template: str, context: dict) -> str:
"""Replace {{variable}} placeholders with context values.
Supports nested access: {{data.email}}, {{fields.name}}.
"""
if not isinstance(template, str):
return template
def replacer(match):
key = match.group(1).strip()
# Try direct key first
if key in context:
return str(context[key])
# Try nested access
parts = key.split(".", 1)
if len(parts) == 2 and parts[0] in context:
inner = context[parts[0]]
if isinstance(inner, dict):
return str(inner.get(parts[1], ""))
return ""
return re.sub(r"\{\{([^}]+)\}\}", replacer, template)
def _interpolate_config(config: dict, context: dict) -> dict:
"""Recursively interpolate template variables in config dict."""
if isinstance(config, str):
return _interpolate(config, context)
if isinstance(config, dict):
return {k: _interpolate_config(v, context) for k, v in config.items()}
if isinstance(config, list):
return [_interpolate_config(item, context) for item in config]
return config
# ─── Step execution ─────────────────────────────────────────────────────────────
def _build_step_context(site: dict, submission: dict, chain_context: dict, step_context: dict) -> dict:
"""Build the context dict for template interpolation in a step.
Context includes:
- submission data (flattened fields)
- site info
- previous step results (chain_context)
- current step outputs (step_context)
"""
# Flatten submission data
fields = {}
if submission.get("customer_name"):
fields["name"] = submission["customer_name"]
if submission.get("customer_phone"):
fields["phone"] = submission["customer_phone"]
if submission.get("customer_email"):
fields["email"] = submission["customer_email"]
if submission.get("data"):
try:
fields.update(json.loads(submission["data"]))
except (json.JSONDecodeError, TypeError):
pass
context = {
"site": site,
"submission": {
"id": submission.get("id"),
"submitted_at": submission.get("submitted_at"),
},
"fields": fields,
# Allow direct access to field values
**fields,
# Chain-level context (results from previous steps)
**chain_context,
# Step-level context
**step_context,
}
return context
def execute_step(step: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Execute a single step in a chain.
Returns:
{
"success": bool,
"output": dict, # Step output for template context
"error": str|None, # Error message on failure
"next_step": str|None # ID of next step to execute
}
"""
step_id = step.get("id", "unknown")
step_type = step.get("type", "http_request")
config = _interpolate_config(step.get("config", {}), _build_step_context(site, submission, chain_context, {}))
logger.info("Executing step %s (type=%s)", step_id, step_type)
try:
# Route to step handler
handler = STEP_HANDLERS.get(step_type)
if handler is None:
return {
"success": False,
"output": {},
"error": f"Unknown step type: {step_type}",
"next_step": step.get("on_failure"),
}
result = handler(step_type, config, site, submission, chain_context)
success = result.get("success", False)
if success:
return {
"success": True,
"output": result.get("output", {}),
"error": None,
"next_step": step.get("on_success"),
}
else:
return {
"success": False,
"output": {},
"error": result.get("error", "Step failed"),
"next_step": step.get("on_failure"),
}
except Exception as e:
logger.error("Step %s execution error: %s", step_id, e)
return {
"success": False,
"output": {},
"error": str(e),
"next_step": step.get("on_failure"),
}
# ─── Step Handlers ──────────────────────────────────────────────────────────────
def _handle_http_request(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Execute an HTTP request step.
Config:
method: GET|POST|PUT|PATCH|DELETE (default POST)
url: Target URL
headers: Dict of headers
body: Request body (JSON) or body_template for template strings
json: Dict sent as JSON body (alternative to body/body_template)
"""
from app.services.webhook import _deliver_webhook
url = config.get("url", "")
if not url:
return {"success": False, "output": {}, "error": "Missing URL"}
method = config.get("method", "POST").upper()
headers = config.get("headers", {})
# Build payload
if "json" in config:
payload = config["json"]
elif "body" in config:
payload = config["body"]
else:
payload = {}
# For POST/PUT/PATCH, use _deliver_webhook
if method in ("POST", "PUT", "PATCH"):
success, error = _deliver_webhook(url, payload, headers=headers)
if success:
return {"success": True, "output": {"status": 200, "url": url}}
else:
return {"success": False, "output": {}, "error": error or "Request failed"}
# GET/DELETE
import requests
try:
kwargs = {"timeout": 10, "headers": headers}
if method == "GET":
resp = requests.get(url, **kwargs)
elif method == "DELETE":
resp = requests.delete(url, **kwargs)
else:
resp = requests.request(method, url, json=payload, **kwargs)
if 200 <= resp.status_code < 300:
return {
"success": True,
"output": {
"status": resp.status_code,
"url": url,
"body": resp.text[:1000] if resp.text else None,
},
}
else:
return {"success": False, "output": {}, "error": f"HTTP {resp.status_code}"}
except requests.exceptions.Timeout:
return {"success": False, "output": {}, "error": "Request timeout"}
except Exception as e:
return {"success": False, "output": {}, "error": str(e)}
def _handle_wait(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Wait step — schedules the next step after a delay.
Config:
duration_seconds: Number of seconds to wait (default 60)
duration: Human readable duration string (e.g., "5m", "1h", "30s")
"""
import redis as redis_lib
from rq import Queue
# Parse duration
duration_seconds = config.get("duration_seconds", 60)
# Support human-readable duration string
duration_str = config.get("duration", "")
if duration_str:
duration_seconds = _parse_duration(duration_str)
# Get the next step ID from the parent chain
next_step_id = config.get("_next_step_id", "")
if not next_step_id:
return {"success": False, "output": {}, "error": "No next step to schedule after wait"}
# Schedule via RQ
redis_url = os.environ.get("REDIS_URL")
if not redis_url:
logger.warning("REDIS_URL not set, cannot schedule wait step")
return {"success": False, "output": {}, "error": "Redis not available for wait"}
try:
r = redis_lib.from_url(redis_url)
r.ping()
q = Queue("chains", connection=r)
# Schedule continuation
scheduled_time = datetime.now(UTC) + timedelta(seconds=duration_seconds)
q.enqueue_at(
scheduled_time,
_continue_chain_from_wait,
args=(site["id"], submission["id"], config.get("_chain_id"), next_step_id),
job_timeout=3600,
result_ttl=86400,
)
logger.info(
"Wait step scheduled for site %s submission %s: %ds delay, continues at %s",
site["id"],
submission["id"],
duration_seconds,
next_step_id,
)
return {
"success": True,
"output": {
"wait_seconds": duration_seconds,
"scheduled_at": scheduled_time.isoformat(),
},
}
except Exception as e:
logger.error("Failed to schedule wait step: %s", e)
return {"success": False, "output": {}, "error": f"Failed to schedule: {e}"}
def _handle_email(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Send email via configured SMTP/Resend/etc.
Config:
to: Recipient(s) — string or list
subject: Email subject
body: Email body (HTML or text)
from_name: Sender name
cc: CC recipients
bcc: BCC recipients
"""
from app.services.action_pipeline import _handle_email as pipeline_email
# Build email config for existing handler
recipients = config.get("to", [])
if isinstance(recipients, str):
recipients = [recipients]
subject = config.get("subject", "New Form Submission")
body = config.get("body", "")
# Use existing email handler
email_config = {
"to": recipients,
"subject": subject,
"body": body,
}
if config.get("from_name"):
email_config["from_name"] = config["from_name"]
if config.get("cc"):
email_config["cc"] = config["cc"]
if config.get("bcc"):
email_config["bcc"] = config["bcc"]
result = pipeline_email("email", email_config, site, submission)
return result
def _handle_log(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Log a message to chain execution log.
Config:
message: Log message (supports template variables)
level: Log level (info, warning, error) — default info
"""
message = config.get("message", "Chain step executed")
level = config.get("level", "info")
log_func = logger.info
if level == "warning":
log_func = logger.warning
elif level == "error":
log_func = logger.error
log_func("Chain log: %s (site=%s, submission=%s)", message, site["id"], submission["id"])
return {
"success": True,
"output": {
"message": message,
"level": level,
},
}
def _handle_condition(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Evaluate a condition and route based on result.
Config:
condition: { "field": "...", "operator": "...", "value": "..." }
true_step: Step ID if condition is true
false_step: Step ID if condition is false
Returns success with output containing the evaluated result.
The caller (chain executor) routes to the correct next step.
"""
from app.services.form_logic import evaluate_condition
condition = config.get("condition", {})
context = _build_step_context(site, submission, chain_context, {})
# Evaluate condition — evaluate_condition takes (field_def, all_data)
result = evaluate_condition(condition, context)
return {
"success": True,
"output": {
"condition_result": result,
"true_step": config.get("true_step"),
"false_step": config.get("false_step"),
},
}
def _handle_parallel(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Execute multiple steps in parallel.
Config:
steps: List of step dicts to execute in parallel
All parallel steps execute concurrently. The step succeeds if all sub-steps succeed.
"""
import threading
parallel_steps = config.get("steps", [])
results = {}
errors = {}
def _run_step(step):
step_id = step.get("id", "unknown")
try:
result = execute_step(step, site, submission, chain_context)
results[step_id] = result
if not result["success"]:
errors[step_id] = result.get("error", "Step failed")
except Exception as e:
errors[step_id] = str(e)
threads = []
for step in parallel_steps:
t = threading.Thread(target=_run_step, args=(step,), daemon=True)
threads.append(t)
t.start()
for t in threads:
t.join(timeout=60)
all_succeeded = len(errors) == 0
return {
"success": all_succeeded,
"output": {
"parallel_results": results,
"parallel_errors": errors if errors else None,
},
"error": list(errors.values())[0] if errors else None,
}
def _handle_agent_call(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Call an external agent via HTTP with submission context.
Config:
agent_id: Agent registry ID
prompt: Optional prompt/instructions to include
"""
import time
import requests
from app.models_agents import get_agent_with_api_key, log_agent_call, update_agent_call
agent_id = config.get("agent_id")
if not agent_id:
return {"success": False, "output": {}, "error": "No agent_id specified"}
user_id = site.get("user_id", 0)
agent = get_agent_with_api_key(int(agent_id), user_id)
if not agent:
return {"success": False, "output": {}, "error": f"Agent {agent_id} not found or disabled"}
# Build request payload
payload = {
"type": "agent_call",
"site_id": site.get("id"),
"submission_id": submission.get("id"),
"submission_data": submission,
"chain_context": {k: v for k, v in chain_context.items()},
"prompt": config.get("prompt", ""),
}
# Log the call
call_log_id = None
try:
call_log_id = log_agent_call(
user_id=user_id,
agent_id=int(agent_id),
chain_action_id=config.get("_chain_action_id"),
site_id=site.get("id"),
submission_id=submission.get("id"),
request_payload=payload,
)
except Exception:
pass # Don't fail the step if logging fails
start_time = time.time()
timeout = agent.get("max_timeout_seconds", 60)
headers = {"Content-Type": "application/json"}
if agent.get("api_key"):
headers["Authorization"] = f"Bearer {agent['api_key']}"
try:
resp = requests.post(
agent["endpoint_url"],
json=payload,
headers=headers,
timeout=timeout,
)
duration_ms = int((time.time() - start_time) * 1000)
if resp.status_code == 200:
result = resp.json()
agent_output = result.get("output", result) if isinstance(result, dict) else {"response": result}
# Update call log
try:
if call_log_id:
update_agent_call(call_log_id, "completed", response_payload=agent_output, duration_ms=duration_ms)
except Exception:
pass
return {
"success": True,
"output": {"agent_result": agent_output, "status": resp.status_code, "duration_ms": duration_ms},
}
else:
error_msg = f"Agent returned HTTP {resp.status_code}"
try:
if call_log_id:
update_agent_call(call_log_id, "failed", error_message=error_msg, duration_ms=duration_ms)
except Exception:
pass
return {"success": False, "output": {}, "error": error_msg}
except requests.exceptions.Timeout:
duration_ms = int((time.time() - start_time) * 1000)
error_msg = f"Agent call timed out after {timeout}s"
try:
if call_log_id:
update_agent_call(call_log_id, "timeout", error_message=error_msg, duration_ms=duration_ms)
except Exception:
pass
return {"success": False, "output": {}, "error": error_msg}
except Exception as e:
duration_ms = int((time.time() - start_time) * 1000)
error_msg = f"Agent call failed: {str(e)}"
try:
if call_log_id:
update_agent_call(call_log_id, "failed", error_message=str(e), duration_ms=duration_ms)
except Exception:
pass
return {"success": False, "output": {}, "error": error_msg}
def _handle_workflow(step_type: str, config: dict, site: dict, submission: dict, chain_context: dict) -> dict:
"""Execute a multi-agent workflow via the orchestrator engine.
Config:
phases: List of workflow phases (each with agents, parallel flag, routing)
max_phases: Maximum phases to execute (default 20)
continue_on_error: Continue workflow on phase failure (default false)
"""
from app.services.orchestrator import execute_workflow
workflow = {
"phases": config.get("phases", []),
"max_phases": config.get("max_phases", 20),
"continue_on_error": config.get("continue_on_error", False),
}
try:
result = execute_workflow(workflow, site, submission)
return {
"success": result["success"],
"output": {
"workflow_context": result["context"],
"phases_executed": result["phases_executed"],
"agent_results": result["results"],
},
"error": result.get("error"),
}
except Exception as e:
logger.error("Workflow execution error: %s", e)
return {
"success": False,
"output": {},
"error": f"Workflow failed: {str(e)}",
}
# Registry of step handlers
STEP_HANDLERS = {
"http_request": _handle_http_request,
"webhook": _handle_http_request, # Alias for backward compatibility with DB chain configs
"wait": _handle_wait,
"email": _handle_email,
"log": _handle_log,
"condition": _handle_condition,
"parallel": _handle_parallel,
"agent_call": _handle_agent_call,
"workflow": _handle_workflow,
}
# ─── Chain Execution ────────────────────────────────────────────────────────────
def execute_chain(chain_action: dict, site: dict, submission: dict, field_config: dict | None = None) -> dict:
"""Execute a full chain of steps.
Args:
chain_action: The chain action dict from form_actions
site: Site dict
submission: Submission dict
field_config: Field configuration for template context
Returns:
{
"success": bool,
"steps_executed": int,
"steps_failed": int,
"chain_context": dict, # Accumulated context
"error": str|None,
}
"""
config = chain_action.get("config", {})
# Ensure config is parsed as dict (may be stored as JSON string in DB)
if isinstance(config, str):
try:
config = json.loads(config)
except (json.JSONDecodeError, TypeError):
logger.error("Chain action: invalid config JSON")
return {
"success": False,
"steps_executed": 0,
"steps_failed": 0,
"chain_context": {},
"error": "invalid_config",
}
steps = config.get("steps", [])
max_steps = config.get("max_steps", 20) # Prevent infinite loops
if not steps:
return {"success": True, "steps_executed": 0, "steps_failed": 0, "chain_context": {}}
# Build step lookup
step_map = {step["id"]: step for step in steps}
# Start from first step or configured entry point
entry_step = config.get("entry_step", steps[0]["id"])
current_step_id = entry_step
chain_context = {
"chain_id": chain_action.get("id", "unknown"),
"started_at": datetime.now(UTC).isoformat(),
}
steps_executed = 0
steps_failed = 0
logger.info("Starting chain execution for site %s submission %s", site["id"], submission["id"])
while current_step_id and steps_executed < max_steps:
step = step_map.get(current_step_id)
if not step:
logger.warning("Step %s not found, ending chain", current_step_id)
break
# Execute step
result = execute_step(step, site, submission, chain_context)
if result["success"]:
chain_context.update(result["output"])
steps_executed += 1
logger.info("Step %s completed successfully", current_step_id)
else:
steps_failed += 1
logger.warning("Step %s failed: %s", current_step_id, result.get("error"))
# Handle condition steps specially — they route based on evaluation
if step["type"] == "condition" and result["success"]:
condition_result = result["output"].get("condition_result")
if condition_result:
current_step_id = step.get("on_success", step.get("config", {}).get("true_step"))
else:
current_step_id = step.get("on_failure", step.get("config", {}).get("false_step"))
else:
# Normal routing
current_step_id = result["next_step"]
# Handle parallel fan-out (multiple next steps)
if isinstance(current_step_id, list):
for next_id in current_step_id:
if next_id and next_id in step_map:
# Fork execution — schedule parallel branches
_fork_chain_branch(next_id, step_map, site, submission, chain_context.copy())
current_step_id = None # Main chain ends
break
# Store step result in chain context
chain_context[f"step_{current_step_id}"] = result
logger.info(
"Chain execution complete: %d executed, %d failed",
steps_executed,
steps_failed,
)
return {
"success": steps_failed == 0,
"steps_executed": steps_executed,
"steps_failed": steps_failed,
"chain_context": chain_context,
"error": None if steps_failed == 0 else f"{steps_failed} step(s) failed",
}
def _fork_chain_branch(next_step_id: str, step_map: dict, site: dict, submission: dict, chain_context: dict):
"""Execute a parallel branch of a chain.
Forks run in the same chain context but execute independently.
"""
import threading
def _run_branch():
current = next_step_id
while current and current in step_map:
step = step_map[current]
result = execute_step(step, site, submission, chain_context)
if result["success"]:
chain_context.update(result["output"])
current = result["next_step"]
else:
break
thread = threading.Thread(target=_run_branch, daemon=True, name=f"chain-fork-{next_step_id}")
thread.start()
def _continue_chain_from_wait(site_id: int, submission_id: int, chain_id: str, next_step_id: str):
"""Continue chain execution after a wait step completes.
Called by RQ scheduled job.
"""
from app.models import get_action_by_id, get_site, get_submission
from app.services.form_logic import get_site_field_config
site = get_site(site_id)
if not site:
logger.error("Site %s not found for chain continuation", site_id)
return
submission = get_submission(submission_id)
if not submission:
logger.error("Submission %s not found for chain continuation", submission_id)
return
# Re-fetch the chain action
chain_action = get_action_by_id(chain_id)
if not chain_action:
logger.error("Chain action %s not found for continuation", chain_id)
return
field_config = get_site_field_config(site_id)
# Execute remaining chain from next_step_id
config = chain_action.get("config", {})
if isinstance(config, str):
try:
config = json.loads(config)
except (json.JSONDecodeError, TypeError):
logger.error("Chain action %s: invalid config JSON, cannot continue", chain_id)
return
steps = config.get("steps", [])
step_map = {step["id"]: step for step in steps}
# Execute from next_step_id
current_step_id = next_step_id
chain_context = {
"chain_id": chain_id,
"resumed_at": datetime.now(UTC).isoformat(),
}
max_steps = config.get("max_steps", 20)
steps_executed = 0
while current_step_id and steps_executed < max_steps:
step = step_map.get(current_step_id)
if not step:
break
result = execute_step(step, site, submission, chain_context)
if result["success"]:
chain_context.update(result["output"])
steps_executed += 1
# Normal routing
current_step_id = result["next_step"]
logger.info(
"Chain continuation complete: %d steps executed after wait",
steps_executed,
)
# ─── Helpers ────────────────────────────────────────────────────────────────────
def _parse_duration(duration_str: str) -> int:
"""Parse human-readable duration string to seconds.
Supports: "30s", "5m", "1h", "2d", "1w", "1h30m"
"""
total = 0
pattern = re.compile(r"(\d+)([smhdw])")
for match in pattern.finditer(duration_str):
value = int(match.group(1))
unit = match.group(2)
if unit == "s":
total += value
elif unit == "m":
total += value * 60
elif unit == "h":
total += value * 3600
elif unit == "d":
total += value * 86400
elif unit == "w":
total += value * 604800
return max(total, 1) # Minimum 1 second
# Public alias for testing
parse_wait_duration = _parse_duration
# ─── Validation ─────────────────────────────────────────────────────────────────
def validate_chain(config: dict) -> dict:
"""Validate a chain configuration before saving.
Returns:
{
"valid": bool,
"errors": list[str],
}
"""
errors = []
steps = config.get("steps", [])
if not steps:
errors.append("Chain must have at least one step")
return {"valid": False, "errors": errors}
step_ids = [step.get("id") for step in steps]
entry_step = config.get("entry_step", steps[0].get("id"))
# Check for duplicate IDs
seen_ids = set()
for sid in step_ids:
if sid in seen_ids:
errors.append(f"Duplicate step ID: '{sid}'")
seen_ids.add(sid)
if entry_step not in seen_ids:
errors.append(f"Entry step '{entry_step}' not found in steps")
for step in steps:
step_id = step.get("id", "unknown")
# Validate step type
if step.get("type") not in STEP_HANDLERS:
errors.append(f"Step '{step_id}': unknown type '{step.get('type')}'")
# Validate routing references
for route_key in ("on_success", "on_failure"):
target = step.get(route_key)
if target and target not in seen_ids and target != "end":
errors.append(f"Step '{step_id}': {route_key} '{target}' not found in steps")
return {
"valid": len(errors) == 0,
"errors": errors,
}