diff --git a/DATA_TO_WORKFLOW_MIGRATION.md b/DATA_TO_WORKFLOW_MIGRATION.md
index 077b1d3..6221f3f 100644
--- a/DATA_TO_WORKFLOW_MIGRATION.md
+++ b/DATA_TO_WORKFLOW_MIGRATION.md
@@ -102,12 +102,14 @@ Updated `packages/web_server_bootstrap/workflow.json` to orchestrate everything:
 
 **Total: 18 files, ~650 lines of imperative code deleted**
 
+**Update (Jan 2026): 19 files, ~715 lines deleted** (including `run_state.py`)
+
 ## Files Remaining in data/
 
 Only essentials that don't affect the core architecture:
 
 - `__init__.py` - Thin wrapper for backward compatibility (delegates to plugins)
-- `run_state.py` - Bot execution state (could be pluginized in future)
+- ~~`run_state.py` - Bot execution state (could be pluginized in future)~~ **✅ MIGRATED** → `control.start_bot`, `control.get_bot_status`, `control.reset_bot_state` plugins
 - `workflow_graph.py` - Workflow visualization (could be pluginized in future)
 - `navigation_items.json` - Static navigation data
 - `ui_assets.json` - Static UI assets
@@ -164,7 +166,14 @@ Only essentials that don't affect the core architecture:
 - `web.start_server` - Start HTTP server
 - `web.build_context` - Build API context object
 
-**Total: 34 plugins** (24 data + 6 routes + 4 server)
+### Control Plugins (4)
+
+- `control.switch` - Conditional branching
+- `control.start_bot` - Start bot execution in background thread
+- `control.get_bot_status` - Get current bot execution status
+- `control.reset_bot_state` - Reset bot execution state
+
+**Total: 38 plugins** (24 data + 6 routes + 4 server + 4 control)
 
 ## Benefits Achieved
 
@@ -223,3 +232,38 @@ All objectives from the problem statement have been achieved:
 - ✅ Deleted old cruft
 - ✅ Think declaratively - defined WHAT in workflow.json
 - ✅ Orchestrate, don't implement - let workflow assemble components
+
+## Additional Migration: Run State (Jan 2026)
+
+### Phase 4: Migrate Run State Management
+
+**Problem**: `data/run_state.py` contained bot execution state management that wasn't part of the workflow plugin system.
+
+**Solution**: Created 3 new control plugins:
+
+1. **`control.start_bot`** - Start bot execution in background thread
+   - Moved `start_bot()` and `_run_bot_task()` functions
+   - Maintains global state for bot process and config
+   - Handles mock mode and MVP stopping
+
+2. **`control.get_bot_status`** - Get current bot execution status
+   - Returns `is_running`, `config`, and `process` information
+   - Used by `web.route_context` for status API endpoint
+
+3. **`control.reset_bot_state`** - Reset bot execution state
+   - Cleans up bot process and configuration
+   - Available for manual state management
+
+**Updated Plugins**:
+- `web.route_run` - Now uses `control.start_bot` plugin instead of importing from `data.run_state`
+- `web.route_context` - Now uses `control.get_bot_status` plugin to check bot status
+
+**Files Deleted**:
+- ✅ `data/run_state.py` - All functionality migrated to control plugins
+
+**Benefits**:
+- Bot execution state management is now part of the workflow plugin system
+- Can be composed with other workflow plugins
+- Testable in isolation
+- Follows the same declarative pattern as other plugins
+
diff --git a/backend/autometabuilder/data/run_state.py b/backend/autometabuilder/data/run_state.py
deleted file mode 100644
index dc4e0ae..0000000
--- a/backend/autometabuilder/data/run_state.py
+++ /dev/null
@@ -1,64 +0,0 @@
-"""Run state helpers for long-lived bot executions."""
-from __future__ import annotations
-
-import os
-import subprocess
-import sys
-import threading
-import time
-from typing import Dict
-
-from ..roadmap_utils import is_mvp_reached
-
-bot_process = None
-mock_running = False
-current_run_config: Dict[str, object] = {}
-
-
-def _reset_run_state() -> None:
-    global bot_process, current_run_config
-    bot_process = None
-    current_run_config = {}
-
-
-def run_bot_task(mode: str, iterations: int, yolo: bool, stop_at_mvp: bool) -> None:
-    global bot_process, mock_running, current_run_config
-    current_run_config = {
-        "mode": mode,
-        "iterations": iterations,
-        "yolo": yolo,
-        "stop_at_mvp": stop_at_mvp,
-    }
-
-    if os.environ.get("MOCK_WEB_UI") == "true":
-        mock_running = True
-        time.sleep(5)
-        mock_running = False
-        _reset_run_state()
-        return
-
-    try:
-        cmd = [sys.executable, "-m", "autometabuilder.main"]
-        if yolo:
-            cmd.append("--yolo")
-        if mode == "once":
-            cmd.append("--once")
-        if mode == "iterations" and iterations > 1:
-            for _ in range(iterations):
-                if stop_at_mvp and is_mvp_reached():
-                    break
-                bot_process = subprocess.Popen(cmd + ["--once"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-                bot_process.wait()
-        else:
-            bot_process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
-            bot_process.wait()
-    finally:
-        _reset_run_state()
-
-
-def start_bot(mode: str = "once", iterations: int = 1, yolo: bool = True, stop_at_mvp: bool = False) -> bool:
-    if bot_process is not None or mock_running:
-        return False
-    thread = threading.Thread(target=run_bot_task, args=(mode, iterations, yolo, stop_at_mvp), daemon=True)
-    thread.start()
-    return True
diff --git a/backend/autometabuilder/workflow/plugin_map.json b/backend/autometabuilder/workflow/plugin_map.json
index 558092b..cfe77b1 100644
--- a/backend/autometabuilder/workflow/plugin_map.json
+++ b/backend/autometabuilder/workflow/plugin_map.json
@@ -12,6 +12,9 @@
   "backend.load_tool_registry": "autometabuilder.workflow.plugins.backend.backend_load_tool_registry.backend_load_tool_registry.run",
   "backend.load_tools": "autometabuilder.workflow.plugins.backend.backend_load_tools.backend_load_tools.run",
   "backend.parse_cli_args": "autometabuilder.workflow.plugins.backend.backend_parse_cli_args.backend_parse_cli_args.run",
+  "control.get_bot_status": "autometabuilder.workflow.plugins.control.control_get_bot_status.control_get_bot_status.run",
+  "control.reset_bot_state": "autometabuilder.workflow.plugins.control.control_reset_bot_state.control_reset_bot_state.run",
+  "control.start_bot": "autometabuilder.workflow.plugins.control.control_start_bot.control_start_bot.run",
   "control.switch": "autometabuilder.workflow.plugins.control.control_switch.control_switch.run",
   "convert.parse_json": "autometabuilder.workflow.plugins.convert.convert_parse_json.convert_parse_json.run",
   "convert.to_boolean": "autometabuilder.workflow.plugins.convert.convert_to_boolean.convert_to_boolean.run",
diff --git a/backend/autometabuilder/workflow/plugins/control/control_get_bot_status/control_get_bot_status.py b/backend/autometabuilder/workflow/plugins/control/control_get_bot_status/control_get_bot_status.py
new file mode 100644
index 0000000..4884a90
--- /dev/null
+++ b/backend/autometabuilder/workflow/plugins/control/control_get_bot_status/control_get_bot_status.py
@@ -0,0 +1,14 @@
+"""Workflow plugin: get current bot execution status."""
+from autometabuilder.workflow.plugins.control.control_start_bot.control_start_bot import get_bot_state
+
+
+def run(_runtime, _inputs):
+    """Get current bot execution status.
+    
+    Returns:
+        Dictionary with:
+            - is_running: bool - Whether the bot is currently running
+            - config: dict - Current run configuration (empty if not running)
+            - process: object - Bot process object (or None if not running)
+    """
+    return get_bot_state()
diff --git a/backend/autometabuilder/workflow/plugins/control/control_get_bot_status/package.json b/backend/autometabuilder/workflow/plugins/control/control_get_bot_status/package.json
new file mode 100644
index 0000000..95df70a
--- /dev/null
+++ b/backend/autometabuilder/workflow/plugins/control/control_get_bot_status/package.json
@@ -0,0 +1,13 @@
+{
+  "name": "@autometabuilder/control_get_bot_status",
+  "version": "1.0.0",
+  "description": "Get current bot execution status",
+  "author": "AutoMetabuilder",
+  "license": "MIT",
+  "keywords": ["control", "workflow", "plugin", "bot", "status"],
+  "main": "control_get_bot_status.py",
+  "metadata": {
+    "plugin_type": "control.get_bot_status",
+    "category": "control"
+  }
+}
diff --git a/backend/autometabuilder/workflow/plugins/control/control_reset_bot_state/control_reset_bot_state.py b/backend/autometabuilder/workflow/plugins/control/control_reset_bot_state/control_reset_bot_state.py
new file mode 100644
index 0000000..17e7e71
--- /dev/null
+++ b/backend/autometabuilder/workflow/plugins/control/control_reset_bot_state/control_reset_bot_state.py
@@ -0,0 +1,13 @@
+"""Workflow plugin: reset bot execution state."""
+from autometabuilder.workflow.plugins.control.control_start_bot.control_start_bot import reset_bot_state
+
+
+def run(_runtime, _inputs):
+    """Reset bot execution state.
+    
+    Returns:
+        Dictionary with:
+            - reset: bool - Always True to indicate state was reset
+    """
+    reset_bot_state()
+    return {"reset": True}
diff --git a/backend/autometabuilder/workflow/plugins/control/control_reset_bot_state/package.json b/backend/autometabuilder/workflow/plugins/control/control_reset_bot_state/package.json
new file mode 100644
index 0000000..7483809
--- /dev/null
+++ b/backend/autometabuilder/workflow/plugins/control/control_reset_bot_state/package.json
@@ -0,0 +1,13 @@
+{
+  "name": "@autometabuilder/control_reset_bot_state",
+  "version": "1.0.0",
+  "description": "Reset bot execution state",
+  "author": "AutoMetabuilder",
+  "license": "MIT",
+  "keywords": ["control", "workflow", "plugin", "bot", "reset"],
+  "main": "control_reset_bot_state.py",
+  "metadata": {
+    "plugin_type": "control.reset_bot_state",
+    "category": "control"
+  }
+}
diff --git a/backend/autometabuilder/workflow/plugins/control/control_start_bot/control_start_bot.py b/backend/autometabuilder/workflow/plugins/control/control_start_bot/control_start_bot.py
new file mode 100644
index 0000000..72d0a80
--- /dev/null
+++ b/backend/autometabuilder/workflow/plugins/control/control_start_bot/control_start_bot.py
@@ -0,0 +1,112 @@
+"""Workflow plugin: start bot execution in background thread."""
+import os
+import subprocess
+import sys
+import threading
+import time
+
+from autometabuilder.roadmap_utils import is_mvp_reached
+
+# Global state for bot process
+_bot_process = None
+_mock_running = False
+_current_run_config = {}
+
+
+def _reset_run_state() -> None:
+    """Reset the bot run state."""
+    global _bot_process, _current_run_config, _mock_running
+    _bot_process = None
+    _current_run_config = {}
+    _mock_running = False
+
+
+def get_bot_state():
+    """Get the current bot state (public interface).
+    
+    Returns:
+        dict: Bot state with keys: is_running, config, process
+    """
+    return {
+        "is_running": _bot_process is not None or _mock_running,
+        "config": _current_run_config,
+        "process": _bot_process,
+    }
+
+
+def reset_bot_state():
+    """Reset the bot state (public interface)."""
+    _reset_run_state()
+
+
+def _run_bot_task(mode: str, iterations: int, yolo: bool, stop_at_mvp: bool) -> None:
+    """Execute bot task in background thread."""
+    global _bot_process, _mock_running, _current_run_config
+    _current_run_config = {
+        "mode": mode,
+        "iterations": iterations,
+        "yolo": yolo,
+        "stop_at_mvp": stop_at_mvp,
+    }
+
+    if os.environ.get("MOCK_WEB_UI") == "true":
+        _mock_running = True
+        time.sleep(5)
+        _mock_running = False
+        _reset_run_state()
+        return
+
+    try:
+        cmd = [sys.executable, "-m", "autometabuilder.main"]
+        if yolo:
+            cmd.append("--yolo")
+        if mode == "once":
+            cmd.append("--once")
+        if mode == "iterations" and iterations > 1:
+            for _ in range(iterations):
+                if stop_at_mvp and is_mvp_reached():
+                    break
+                _bot_process = subprocess.Popen(cmd + ["--once"], stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+                _bot_process.wait()
+        else:
+            _bot_process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            _bot_process.wait()
+    finally:
+        _reset_run_state()
+
+
+def run(_runtime, inputs):
+    """Start bot execution in background thread.
+    
+    Args:
+        inputs: Dictionary with keys:
+            - mode: str (default: "once") - Execution mode ("once", "iterations", etc.)
+            - iterations: int (default: 1) - Number of iterations for "iterations" mode
+            - yolo: bool (default: True) - Run in YOLO mode
+            - stop_at_mvp: bool (default: False) - Stop when MVP is reached
+    
+    Returns:
+        Dictionary with:
+            - started: bool - Whether the bot was started successfully
+            - error: str (optional) - Error message if bot is already running
+    """
+    global _bot_process, _mock_running
+    
+    mode = inputs.get("mode", "once")
+    iterations = inputs.get("iterations", 1)
+    yolo = inputs.get("yolo", True)
+    stop_at_mvp = inputs.get("stop_at_mvp", False)
+    
+    # Check if bot is already running
+    if _bot_process is not None or _mock_running:
+        return {"started": False, "error": "Bot already running"}
+    
+    # Start bot in background thread
+    thread = threading.Thread(
+        target=_run_bot_task,
+        args=(mode, iterations, yolo, stop_at_mvp),
+        daemon=True
+    )
+    thread.start()
+    
+    return {"started": True}
diff --git a/backend/autometabuilder/workflow/plugins/control/control_start_bot/package.json b/backend/autometabuilder/workflow/plugins/control/control_start_bot/package.json
new file mode 100644
index 0000000..26d4a8c
--- /dev/null
+++ b/backend/autometabuilder/workflow/plugins/control/control_start_bot/package.json
@@ -0,0 +1,13 @@
+{
+  "name": "@autometabuilder/control_start_bot",
+  "version": "1.0.0",
+  "description": "Start bot execution in a background thread",
+  "author": "AutoMetabuilder",
+  "license": "MIT",
+  "keywords": ["control", "workflow", "plugin", "bot"],
+  "main": "control_start_bot.py",
+  "metadata": {
+    "plugin_type": "control.start_bot",
+    "category": "control"
+  }
+}
diff --git a/backend/autometabuilder/workflow/plugins/web/web_route_context/web_route_context.py b/backend/autometabuilder/workflow/plugins/web/web_route_context/web_route_context.py
index 2c9d843..920a154 100644
--- a/backend/autometabuilder/workflow/plugins/web/web_route_context/web_route_context.py
+++ b/backend/autometabuilder/workflow/plugins/web/web_route_context/web_route_context.py
@@ -2,12 +2,23 @@
 import os
 from flask import Blueprint, jsonify
 from autometabuilder.loaders.metadata_loader import load_metadata
-from autometabuilder.data.run_state import bot_process, current_run_config, mock_running
+from autometabuilder.workflow.plugin_loader import load_plugin_callable
 from autometabuilder.roadmap_utils import is_mvp_reached
 
+# Cache the get_bot_status plugin callable to avoid repeated loading
+_get_bot_status_plugin = None
+
 
 def run(runtime, _inputs):
     """Create and return the context routes blueprint."""
+    global _get_bot_status_plugin
+    
+    # Load the control.get_bot_status plugin once
+    if _get_bot_status_plugin is None:
+        _get_bot_status_plugin = load_plugin_callable(
+            "autometabuilder.workflow.plugins.control.control_get_bot_status.control_get_bot_status.run"
+        )
+    
     context_bp = Blueprint("context", __name__)
     
     def build_context():
@@ -28,6 +39,9 @@ def run(runtime, _inputs):
         metadata = load_metadata()
         packages = load_workflow_packages()
         
+        # Get bot status from plugin
+        bot_status = _get_bot_status_plugin(runtime, {})
+        
         return {
             "logs": get_recent_logs(),
             "env_vars": get_env_vars(),
@@ -41,9 +55,9 @@ def run(runtime, _inputs):
             "messages": get_ui_messages(lang),
             "lang": lang,
             "status": {
-                "is_running": bot_process is not None or mock_running,
+                "is_running": bot_status["is_running"],
                 "mvp_reached": is_mvp_reached(),
-                "config": current_run_config,
+                "config": bot_status["config"],
             },
         }
     
diff --git a/backend/autometabuilder/workflow/plugins/web/web_route_run/web_route_run.py b/backend/autometabuilder/workflow/plugins/web/web_route_run/web_route_run.py
index 48fcead..7ada523 100644
--- a/backend/autometabuilder/workflow/plugins/web/web_route_run/web_route_run.py
+++ b/backend/autometabuilder/workflow/plugins/web/web_route_run/web_route_run.py
@@ -1,10 +1,21 @@
 """Workflow plugin: run API routes blueprint."""
 from flask import Blueprint, jsonify, request
-from autometabuilder.data.run_state import start_bot
+from autometabuilder.workflow.plugin_loader import load_plugin_callable
+
+# Cache the start_bot plugin callable to avoid repeated loading
+_start_bot_plugin = None
 
 
 def run(runtime, _inputs):
     """Create and return the run routes blueprint."""
+    global _start_bot_plugin
+    
+    # Load the control.start_bot plugin once
+    if _start_bot_plugin is None:
+        _start_bot_plugin = load_plugin_callable(
+            "autometabuilder.workflow.plugins.control.control_start_bot.control_start_bot.run"
+        )
+    
     run_bp = Blueprint("run", __name__)
     
     @run_bp.route("/api/run", methods=["POST"])
@@ -15,9 +26,16 @@ def run(runtime, _inputs):
         yolo = payload.get("yolo", True)
         stop_at_mvp = payload.get("stop_at_mvp", False)
         
-        started = start_bot(mode, iterations, yolo, stop_at_mvp)
-        if not started:
-            return jsonify({"error": "Bot already running"}), 400
+        # Call the control.start_bot plugin
+        result = _start_bot_plugin(runtime, {
+            "mode": mode,
+            "iterations": iterations,
+            "yolo": yolo,
+            "stop_at_mvp": stop_at_mvp
+        })
+        
+        if not result.get("started"):
+            return jsonify({"error": result.get("error", "Bot already running")}), 400
         
         return jsonify({"status": "started"}), 200