# OPENCLAW-WRAP: This module will be wrapped around OpenClaw. Preserve public API. """Workflow runner for executing Rex workflows. This module provides the WorkflowRunner class that executes workflows step by step, handling policy checks, approvals, idempotency, and persistence. The runner: - Evaluates preconditions before each step - Consults the PolicyEngine to gate tool calls - Creates approvals for steps that require user review - Executes allowed tool calls via the tool router - Evaluates postconditions after execution - Persists state to disk after each step for recovery - Supports dry-run mode for previewing actions - Supports resume from blocked state after approval Usage: from rex.workflow import Workflow, WorkflowStep from rex.workflow_runner import WorkflowRunner from rex.contracts import ToolCall workflow = Workflow( title="Example", steps=[ WorkflowStep( description="Get time", tool_call=ToolCall(tool="time_now", args={"location": "Dallas, TX"}) ) ] ) runner = WorkflowRunner(workflow) result = runner.run() """ from __future__ import annotations import json import logging import subprocess import sys from collections.abc import Callable from dataclasses import dataclass, field from datetime import UTC, datetime from pathlib import Path from typing import Any from rex.audit import AuditLogger, LogEntry, get_audit_logger from rex.contracts import ToolCall from rex.openclaw.tool_executor import ( execute_tool, ) from rex.policy import PolicyDecision from rex.policy_engine import PolicyEngine, get_policy_engine from rex.workflow import ( DEFAULT_APPROVAL_DIR, DEFAULT_WORKFLOW_DIR, StepResult, Workflow, WorkflowApproval, WorkflowStep, generate_approval_id, get_condition, validate_workflow_steps, ) logger = logging.getLogger(__name__) @dataclass class DryRunStepResult: """Result of a dry-run step evaluation.""" step_id: str description: str tool: str | None would_execute: bool policy_decision: str # "allowed", "requires_approval", "denied" reason: str precondition_passed: bool precondition_name: str | None @dataclass class DryRunResult: """Summary of a dry-run workflow execution.""" workflow_id: str title: str total_steps: int steps: list[DryRunStepResult] would_complete: bool blocking_reason: str | None @dataclass class RunResult: """Result of running a workflow.""" workflow_id: str status: str steps_executed: int steps_total: int error: str | None blocking_approval_id: str | None class WorkflowRunner: """Runner for executing workflows step by step. The WorkflowRunner handles the execution lifecycle of a workflow: 1. Load or create workflow state 2. Iterate through steps from current_step_index 3. For each step: - Evaluate preconditions (skip if False) - Check idempotency (skip if already executed) - Consult policy engine for tool calls - Handle approvals for gated actions - Execute tool calls - Evaluate postconditions - Update and persist state 4. Handle completion, failure, and blocking states The runner uses dependency injection for policy engine and audit logger, making it testable and configurable. """ def __init__( self, workflow: Workflow, *, policy_engine: PolicyEngine | None = None, audit_logger: AuditLogger | None = None, workflow_dir: Path | str | None = None, approval_dir: Path | str | None = None, default_context: dict[str, Any] | None = None, pre_step_hook: Callable[[WorkflowStep], None] | None = None, before_tool_call_hook: Callable[[WorkflowStep], None] | None = None, after_tool_call_hook: Callable[[WorkflowStep, StepResult], None] | None = None, ) -> None: """Initialize the workflow runner. Args: workflow: The workflow to execute. policy_engine: Optional policy engine. Uses singleton if not provided. audit_logger: Optional audit logger. Uses singleton if not provided. workflow_dir: Directory for workflow persistence. approval_dir: Directory for approval persistence. default_context: Default context for tool execution. """ self.workflow = workflow self.policy_engine = policy_engine or get_policy_engine() self.audit_logger = audit_logger or get_audit_logger() self.workflow_dir = Path(workflow_dir) if workflow_dir else DEFAULT_WORKFLOW_DIR self.approval_dir = Path(approval_dir) if approval_dir else DEFAULT_APPROVAL_DIR self.default_context = default_context or {} self.pre_step_hook = pre_step_hook self.before_tool_call_hook = before_tool_call_hook self.after_tool_call_hook = after_tool_call_hook def run(self) -> RunResult: """Run the workflow from its current state. Executes steps starting from current_step_index until: - All steps are completed (status -> "completed") - A step fails (status -> "failed") - A step requires approval (status -> "blocked") State is persisted after each step to enable recovery. Returns: RunResult with execution summary. """ steps_executed = 0 # Validate all step inputs before executing validation_errors = validate_workflow_steps(self.workflow) if validation_errors: all_errors = [e for errs in validation_errors.values() for e in errs] error_msg = "Step validation failed: " + "; ".join(all_errors) logger.error(error_msg) self.workflow.mark_failed(error_msg) self._save_workflow() return RunResult( workflow_id=self.workflow.workflow_id, status="failed", steps_executed=0, steps_total=len(self.workflow.steps), error=error_msg, blocking_approval_id=None, ) # Mark as running if not already blocked if self.workflow.status != "blocked": self.workflow.mark_running() self._save_workflow() logger.info( "Running workflow %s from step %d of %d", self.workflow.workflow_id, self.workflow.current_step_index, len(self.workflow.steps), ) while not self.workflow.is_finished() and not self.workflow.is_blocked(): step = self.workflow.current_step() if step is None: # No more steps, workflow is complete self.workflow.status = "completed" self.workflow.completed_at = datetime.now(UTC) self.workflow.updated_at = datetime.now(UTC) self._save_workflow() break try: if self.pre_step_hook: self.pre_step_hook(step) step_result = self._execute_step(step) steps_executed += 1 if step_result.success or step_result.skipped: # Advance to next step self.workflow.advance() else: # Step failed, mark workflow as failed self.workflow.mark_failed(step_result.error or f"Step {step.step_id} failed") except ApprovalBlockedError as e: # Step is blocked pending approval logger.info( "Workflow %s blocked at step %s pending approval %s", self.workflow.workflow_id, step.step_id, e.approval_id, ) # Don't increment steps_executed for blocked step break self._save_workflow() result = RunResult( workflow_id=self.workflow.workflow_id, status=self.workflow.status, steps_executed=steps_executed, steps_total=len(self.workflow.steps), error=self.workflow.error, blocking_approval_id=self.workflow.blocking_approval_id, ) logger.info( "Workflow %s finished with status=%s, executed=%d/%d steps", self.workflow.workflow_id, self.workflow.status, steps_executed, len(self.workflow.steps), ) return result def resume(self) -> RunResult: """Resume a blocked workflow after approval. Checks the approval status and resumes execution if approved. If the approval is denied, marks the workflow as failed. If the approval is still pending, returns immediately. Returns: RunResult with execution summary. Raises: ValueError: If the workflow is not in blocked state. """ if self.workflow.status != "blocked": raise ValueError( f"Cannot resume workflow in status '{self.workflow.status}', " "expected 'blocked'" ) if not self.workflow.blocking_approval_id: raise ValueError("Workflow is blocked but has no blocking approval ID") # Load the approval approval = WorkflowApproval.load( self.workflow.blocking_approval_id, approval_dir=self.approval_dir, ) if approval is None: raise ValueError(f"Approval {self.workflow.blocking_approval_id} not found") logger.info( "Checking approval %s status=%s for workflow %s", approval.approval_id, approval.status, self.workflow.workflow_id, ) if approval.status == "pending": # Still waiting for approval return RunResult( workflow_id=self.workflow.workflow_id, status="blocked", steps_executed=0, steps_total=len(self.workflow.steps), error=None, blocking_approval_id=self.workflow.blocking_approval_id, ) if approval.status == "denied": self.workflow.mark_failed(f"Approval denied: {approval.reason or 'No reason provided'}") self._save_workflow() return RunResult( workflow_id=self.workflow.workflow_id, status="failed", steps_executed=0, steps_total=len(self.workflow.steps), error=self.workflow.error, blocking_approval_id=None, ) if approval.status == "expired": self.workflow.mark_failed("Approval expired") self._save_workflow() return RunResult( workflow_id=self.workflow.workflow_id, status="failed", steps_executed=0, steps_total=len(self.workflow.steps), error=self.workflow.error, blocking_approval_id=None, ) # Approval is approved, resume execution logger.info( "Approval %s granted, resuming workflow %s", approval.approval_id, self.workflow.workflow_id, ) # Update the step to mark approval received current_step = self.workflow.current_step() if current_step and current_step.step_id == approval.step_id: current_step.approval_id = approval.approval_id # Clear blocked state and run self.workflow.mark_running() self._save_workflow() return self.run() def dry_run(self) -> DryRunResult: """Preview workflow execution without making changes. Evaluates all steps, checking preconditions and policies, but does not execute any tool calls. Returns: DryRunResult with preview of what would happen. """ steps: list[DryRunStepResult] = [] would_complete = True blocking_reason: str | None = None # Temporarily save current state original_step_index = self.workflow.current_step_index for i, step in enumerate(self.workflow.steps): # Skip already executed steps in dry run preview if i < original_step_index: continue # Evaluate precondition precondition_passed = True precondition_name = step.precondition if step.precondition: condition_func = get_condition(step.precondition) if condition_func: try: precondition_passed = condition_func(self.workflow.state) except Exception as e: logger.warning("Precondition %s failed: %s", step.precondition, e) precondition_passed = False # Check policy for tool calls policy_decision = "allowed" reason = "No tool call" would_execute = precondition_passed if step.tool_call and precondition_passed: decision = self._evaluate_policy(step.tool_call) if decision.denied: policy_decision = "denied" reason = decision.reason would_execute = False would_complete = False blocking_reason = f"Step {step.step_id} would be denied: {reason}" elif decision.requires_approval: policy_decision = "requires_approval" reason = decision.reason # Would still execute after approval else: policy_decision = "allowed" reason = decision.reason if not precondition_passed: reason = f"Precondition '{step.precondition}' would fail" would_execute = False steps.append( DryRunStepResult( step_id=step.step_id, description=step.description, tool=step.tool_call.tool if step.tool_call else None, would_execute=would_execute, policy_decision=policy_decision, reason=reason, precondition_passed=precondition_passed, precondition_name=precondition_name, ) ) return DryRunResult( workflow_id=self.workflow.workflow_id, title=self.workflow.title, total_steps=len(self.workflow.steps), steps=steps, would_complete=would_complete, blocking_reason=blocking_reason, ) def _execute_step(self, step: WorkflowStep) -> StepResult: """Execute a single workflow step. Args: step: The step to execute. Returns: StepResult with execution outcome. Raises: ApprovalBlockedError: If the step requires approval. """ logger.debug("Executing step %s: %s", step.step_id, step.description) # Check idempotency if step.idempotency_key: executed_keys = self.workflow.get_executed_idempotency_keys() if step.idempotency_key in executed_keys: logger.info( "Skipping step %s: idempotency key %s already executed", step.step_id, step.idempotency_key, ) result = StepResult( step_id=step.step_id, success=True, skipped=True, skip_reason=f"Idempotency key '{step.idempotency_key}' already executed", ) step.result = result return result # Evaluate precondition if step.precondition: condition_func = get_condition(step.precondition) if condition_func: try: passed = condition_func(self.workflow.state) except Exception as e: logger.warning("Precondition %s raised exception: %s", step.precondition, e) passed = False if not passed: logger.info( "Skipping step %s: precondition '%s' returned False", step.step_id, step.precondition, ) result = StepResult( step_id=step.step_id, success=True, skipped=True, skip_reason=f"Precondition '{step.precondition}' returned False", ) step.result = result return result else: logger.warning( "Precondition '%s' not found in registry, skipping check", step.precondition, ) # If no tool call, step is just a checkpoint if step.tool_call is None: logger.debug("Step %s has no tool call, marking success", step.step_id) result = StepResult( step_id=step.step_id, success=True, output=None, ) step.result = result return result # Check policy decision = self._evaluate_policy(step.tool_call) if decision.denied: logger.warning( "Step %s denied by policy: %s", step.step_id, decision.reason, ) self._log_step_to_audit(step, "denied", None, decision.reason) result = StepResult( step_id=step.step_id, success=False, error=f"Denied by policy: {decision.reason}", ) step.result = result return result if decision.requires_approval: # Check if step already has an approved approval if step.approval_id: approval = WorkflowApproval.load(step.approval_id, self.approval_dir) if approval and approval.status == "approved": logger.debug( "Step %s already approved via %s", step.step_id, step.approval_id, ) # Continue to execution else: # Re-block if approval not found or not approved raise self._create_approval_and_block(step, decision.reason) else: # Create new approval and block raise self._create_approval_and_block(step, decision.reason) # Execute the tool call if self.before_tool_call_hook: self.before_tool_call_hook(step) try: tool_result = execute_tool( {"tool": step.tool_call.tool, "args": step.tool_call.args}, self.default_context, policy_engine=self.policy_engine, skip_policy_check=True, # Already checked above task_id=self.workflow.workflow_id, requested_by=f"workflow:{self.workflow.workflow_id}", ) except Exception as e: logger.error("Step %s tool execution failed: %s", step.step_id, e) result = StepResult( step_id=step.step_id, success=False, error=str(e), ) step.result = result if self.after_tool_call_hook: self.after_tool_call_hook(step, result) return result # Check for error in result if "error" in tool_result: error_info = tool_result["error"] error_msg = ( error_info.get("message", str(error_info)) if isinstance(error_info, dict) else str(error_info) ) result = StepResult( step_id=step.step_id, success=False, output=tool_result, error=error_msg, ) step.result = result if self.after_tool_call_hook: self.after_tool_call_hook(step, result) return result lifecycle = tool_result.get("lifecycle") if isinstance(tool_result, dict) else None lifecycle_state = lifecycle.get("state") if isinstance(lifecycle, dict) else None if lifecycle_state and lifecycle_state not in {"completed", "verified"}: detail = str(tool_result.get("detail") or f"Tool lifecycle ended in {lifecycle_state}.") result = StepResult( step_id=step.step_id, success=False, output=tool_result, error=detail, ) step.result = result if self.after_tool_call_hook: self.after_tool_call_hook(step, result) return result if isinstance(tool_result, dict) and tool_result.get("success") is False: detail = str(tool_result.get("detail") or "Tool reported an unsuccessful result.") result = StepResult( step_id=step.step_id, success=False, output=tool_result, error=detail, ) step.result = result if self.after_tool_call_hook: self.after_tool_call_hook(step, result) return result # Store output in workflow state self.workflow.state[f"step_{step.step_id}_output"] = tool_result # Evaluate postcondition if step.postcondition: condition_func = get_condition(step.postcondition) if condition_func: try: passed = condition_func(self.workflow.state) except Exception as e: logger.warning("Postcondition %s raised exception: %s", step.postcondition, e) passed = False if not passed: logger.warning( "Step %s postcondition '%s' failed", step.step_id, step.postcondition, ) result = StepResult( step_id=step.step_id, success=False, output=tool_result, error=f"Postcondition '{step.postcondition}' failed", ) step.result = result if self.after_tool_call_hook: self.after_tool_call_hook(step, result) return result # Success! result = StepResult( step_id=step.step_id, success=True, output=tool_result, ) step.result = result if self.after_tool_call_hook: self.after_tool_call_hook(step, result) self._log_step_to_audit(step, "allowed", tool_result, None) return result def _evaluate_policy(self, tool_call: ToolCall) -> PolicyDecision: """Evaluate policy for a tool call. Args: tool_call: The tool call to evaluate. Returns: PolicyDecision from the policy engine. """ metadata = self._extract_policy_metadata(tool_call.args) return self.policy_engine.decide(tool_call, metadata) def _extract_policy_metadata(self, args: dict[str, Any]) -> dict[str, Any]: """Extract metadata from tool arguments for policy evaluation.""" metadata: dict[str, Any] = {} for key in ("to", "recipient", "email", "address"): if key in args and isinstance(args[key], str): metadata["recipient"] = args[key] break for key in ("domain", "host"): if key in args and isinstance(args[key], str): metadata["domain"] = args[key] break if "url" in args and isinstance(args["url"], str): url = args["url"] if "://" in url: domain_part = url.split("://", 1)[1].split("/", 1)[0] domain_part = domain_part.split(":")[0] metadata["domain"] = domain_part return metadata def _create_approval_and_block( self, step: WorkflowStep, reason: str, ) -> ApprovalBlockedError: """Create an approval request and block the workflow. Args: step: The step requiring approval. reason: Reason approval is required. Returns: ApprovalBlockedError to be raised. """ tool_summary = None if step.tool_call: tool_summary = f"{step.tool_call.tool}({step.tool_call.args})" approval = WorkflowApproval( approval_id=generate_approval_id(), workflow_id=self.workflow.workflow_id, step_id=step.step_id, status="pending", reason=reason, requested_by="workflow_runner", step_description=step.description, tool_call_summary=tool_summary, ) approval.save(self.approval_dir) step.requires_approval = True step.approval_id = approval.approval_id self.workflow.mark_blocked(approval.approval_id) self._save_workflow() self._log_step_to_audit(step, "requires_approval", None, reason) logger.info( "Created approval %s for step %s, workflow blocked", approval.approval_id, step.step_id, ) return ApprovalBlockedError(approval.approval_id, step.step_id) def _log_step_to_audit( self, step: WorkflowStep, decision: str, result: dict[str, Any] | None, error: str | None, ) -> None: """Log a step execution to the audit log. Args: step: The executed step. decision: Policy decision ("allowed", "denied", "requires_approval"). result: Tool result if any. error: Error message if any. """ if step.tool_call is None: return try: entry = LogEntry( action_id=f"wf_{self.workflow.workflow_id}_{step.step_id}", task_id=self.workflow.workflow_id, tool=step.tool_call.tool, tool_call_args=step.tool_call.args, policy_decision=decision, # type: ignore[arg-type] tool_result=result, error=error, requested_by=f"workflow:{self.workflow.workflow_id}", ) self.audit_logger.log(entry) except Exception as e: logger.warning("Failed to log step to audit: %s", e) def _save_workflow(self) -> None: """Save the workflow to disk.""" try: self.workflow.save(self.workflow_dir) except Exception as e: logger.error("Failed to save workflow: %s", e) class ApprovalBlockedError(Exception): """Raised when a step is blocked pending approval.""" def __init__(self, approval_id: str, step_id: str) -> None: self.approval_id = approval_id self.step_id = step_id super().__init__(f"Step '{step_id}' blocked pending approval '{approval_id}'") def approve_workflow( approval_id: str, *, decided_by: str = "user", reason: str | None = None, approval_dir: Path | str | None = None, ) -> bool: """Approve a pending workflow approval. This is a helper function for programmatically approving workflow steps. Args: approval_id: ID of the approval to approve. decided_by: Who approved (default "user"). reason: Optional approval reason. approval_dir: Directory containing approvals. Returns: True if approval was updated, False if not found. """ if approval_dir is None: approval_dir = DEFAULT_APPROVAL_DIR approval = WorkflowApproval.load(approval_id, approval_dir) if approval is None: return False approval.status = "approved" approval.decided_by = decided_by approval.reason = reason approval.decided_at = datetime.now(UTC) approval.save(approval_dir) logger.info("Approved workflow approval %s by %s", approval_id, decided_by) return True def deny_workflow( approval_id: str, *, decided_by: str = "user", reason: str | None = None, approval_dir: Path | str | None = None, ) -> bool: """Deny a pending workflow approval. Args: approval_id: ID of the approval to deny. decided_by: Who denied (default "user"). reason: Optional denial reason. approval_dir: Directory containing approvals. Returns: True if approval was updated, False if not found. """ if approval_dir is None: approval_dir = DEFAULT_APPROVAL_DIR approval = WorkflowApproval.load(approval_id, approval_dir) if approval is None: return False approval.status = "denied" approval.decided_by = decided_by approval.reason = reason approval.decided_at = datetime.now(UTC) approval.save(approval_dir) logger.info("Denied workflow approval %s by %s", approval_id, decided_by) return True def list_pending_approvals( approval_dir: Path | str | None = None, ) -> list[WorkflowApproval]: """List all pending workflow approvals. Args: approval_dir: Directory containing approvals. Returns: List of pending WorkflowApproval objects. """ if approval_dir is None: approval_dir = DEFAULT_APPROVAL_DIR approval_dir = Path(approval_dir) if not approval_dir.exists(): return [] pending = [] for file_path in approval_dir.glob("*.json"): try: with open(file_path, encoding="utf-8") as f: approval = WorkflowApproval.model_validate_json(f.read()) if approval.status == "pending": pending.append(approval) except Exception as e: logger.warning("Failed to load approval from %s: %s", file_path, e) return sorted(pending, key=lambda a: a.requested_at) __all__ = [ "WorkflowRunner", "RunResult", "DryRunResult", "DryRunStepResult", "ApprovalBlockedError", "approve_workflow", "deny_workflow", "list_pending_approvals", # Scheduled task runner (US-325) "ScheduledTask", "ScheduledTaskRunner", ] # ============================================================================= # Scheduled Task Runner (US-325) # ============================================================================= # Reads simple name/schedule/action task definitions from config and runs # due tasks when the voice loop or daemon calls run_due_tasks(). _REPO_ROOT = Path(__file__).parent.parent _DEFAULT_CONFIG_PATH = _REPO_ROOT / "config" / "rex_config.json" _DEFAULT_STATE_PATH = _REPO_ROOT / "data" / "workflow_runner_state.json" # Built-in example tasks shown when no tasks are configured. All disabled # by default — users opt-in by adding them to rex_config.json. _EXAMPLE_TASKS: list[dict[str, Any]] = [ { "name": "daily_weather", "schedule": "interval:86400", "action": 'chat --no-tts --prompt "What is the weather today?"', "enabled": False, } ] @dataclass class ScheduledTask: """A scheduled workflow task read from config. Attributes: name: Unique task identifier. schedule: When to run, e.g. ``"interval:86400"`` (daily). action: Rex CLI arguments to pass to ``python -m rex``. enabled: Whether the task is active. """ name: str schedule: str action: str enabled: bool = True def interval_seconds(self) -> int | None: """Return interval in seconds for ``interval:`` schedules.""" s = self.schedule.strip().lower() if s.startswith("interval:"): try: return int(s.split(":", 1)[1]) except (ValueError, IndexError): return None return None def is_due(self, last_run: datetime | None) -> bool: """Return True if enough time has elapsed since *last_run*.""" interval = self.interval_seconds() if interval is None: logger.warning( "ScheduledTask '%s' has unsupported schedule '%s'", self.name, self.schedule ) return False if last_run is None: return True return (datetime.now(UTC) - last_run).total_seconds() >= interval @dataclass class ScheduledTaskRunner: """Reads task definitions from config and executes due tasks. Config location: ``config/rex_config.json`` under ``workflows.tasks``. Example config:: { "workflows": { "tasks": [ { "name": "daily_weather", "schedule": "interval:86400", "action": "chat --no-tts --prompt \\"What is the weather today?\\"", "enabled": true } ] } } State (last-run timestamps) is persisted to ``data/workflow_runner_state.json``. """ config_path: Path = field(default_factory=lambda: _DEFAULT_CONFIG_PATH) state_path: Path = field(default_factory=lambda: _DEFAULT_STATE_PATH) def __post_init__(self) -> None: self._tasks: list[ScheduledTask] = self._load_tasks() self._state: dict[str, str] = self._load_state() # ------------------------------------------------------------------ # Config / state # ------------------------------------------------------------------ def _load_tasks(self) -> list[ScheduledTask]: raw: list[dict[str, Any]] = [] if self.config_path.exists(): try: data = json.loads(self.config_path.read_text(encoding="utf-8")) raw = data.get("workflows", {}).get("tasks", []) except (json.JSONDecodeError, OSError) as exc: logger.warning("ScheduledTaskRunner: failed to read config: %s", exc) if not raw: raw = _EXAMPLE_TASKS tasks: list[ScheduledTask] = [] for item in raw: name = item.get("name") schedule = item.get("schedule") action = item.get("action") if not (name and schedule and action): logger.warning("ScheduledTaskRunner: skipping malformed task: %s", item) continue tasks.append( ScheduledTask( name=str(name), schedule=str(schedule), action=str(action), enabled=bool(item.get("enabled", True)), ) ) return tasks def _load_state(self) -> dict[str, str]: if self.state_path.exists(): try: data = json.loads(self.state_path.read_text(encoding="utf-8")) if isinstance(data, dict): return {str(key): str(value) for key, value in data.items()} except (json.JSONDecodeError, OSError) as exc: logger.warning("ScheduledTaskRunner: failed to read state: %s", exc) return {} def _save_state(self) -> None: self.state_path.parent.mkdir(parents=True, exist_ok=True) try: self.state_path.write_text(json.dumps(self._state, indent=2), encoding="utf-8") except OSError as exc: logger.error("ScheduledTaskRunner: failed to save state: %s", exc) def _last_run(self, name: str) -> datetime | None: ts = self._state.get(name) if ts: try: return datetime.fromisoformat(ts) except ValueError: pass return None # ------------------------------------------------------------------ # Execution # ------------------------------------------------------------------ def _run_task(self, task: ScheduledTask) -> bool: """Execute *task*, returning True on success. Never raises.""" logger.info("ScheduledTaskRunner: executing '%s' (action: %s)", task.name, task.action) try: cmd = [sys.executable, "-m", "rex"] + task.action.split() result = subprocess.run( cmd, capture_output=True, text=True, timeout=120, cwd=str(_REPO_ROOT), ) if result.returncode != 0: logger.error( "ScheduledTaskRunner: task '%s' exited %d: %s", task.name, result.returncode, (result.stderr or "")[:500], ) return False logger.info("ScheduledTaskRunner: task '%s' succeeded", task.name) return True except subprocess.TimeoutExpired: logger.error("ScheduledTaskRunner: task '%s' timed out after 120s", task.name) return False except Exception as exc: logger.error("ScheduledTaskRunner: task '%s' raised: %s", task.name, exc) return False def run_due_tasks(self) -> list[str]: """Run all enabled tasks that are due. Failed tasks are logged but do not block subsequent tasks. Returns: Names of tasks that were triggered (regardless of outcome). """ triggered: list[str] = [] for task in self._tasks: if not task.enabled: continue if not task.is_due(self._last_run(task.name)): continue triggered.append(task.name) # Persist last-run before executing so crashes don't cause retries. self._state[task.name] = datetime.now(UTC).isoformat() self._save_state() try: self._run_task(task) except Exception as exc: logger.error( "ScheduledTaskRunner: task '%s' raised unexpectedly: %s", task.name, exc ) return triggered def tasks(self) -> list[ScheduledTask]: """Return all loaded tasks (enabled or not).""" return list(self._tasks)