From 2d133a3506bb6442d05ad3530ab67481c131ad8e Mon Sep 17 00:00:00 2001
From: Alain Prasquier <alain@supervaize.com>
Date: Wed, 18 Mar 2026 14:21:03 +0200
Subject: [PATCH 1/2] =?UTF-8?q?=E2=9C=A8=20feat:=20custom=5Froutes=20+=20s?=
 =?UTF-8?q?cheduled=20steps=20+=20executor=20+=20workbench=20UI?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Platform extensions for agent-driven orchestration:

- custom_routes: agents mount their own FastAPI routers at /agents/{slug}/api/
- scheduled steps: CaseNodeUpdate gains scheduled_at/method/params/status fields
- executor: background asyncio loop polls every 60s, executes due steps
- workbench UI: countdown display, Execute now / Cancel buttons on pending steps
- routes: POST execute, POST cancel, PATCH reschedule for scheduled steps
- Case.cancel_scheduled_steps() for job stop cascading
- Cases.get_due_scheduled_steps() for executor polling
- dialog_renderer: lightweight markdown rendering for text content
- 433 tests passing
---
 docs/CHANGELOG.md                             |  10 ++
 .../admin/static/js/workbench-form.js         |  37 ++++++
 .../templates/components/dialog_renderer.html |  62 +++++++++-
 .../admin/templates/workbench.html            |   2 +-
 .../admin/templates/workbench_monitor.html    |  26 ++++
 src/supervaizer/admin/workbench_routes.py     | 113 ++++++++++++++++++
 src/supervaizer/agent.py                      |  11 +-
 src/supervaizer/case.py                       |  50 +++++++-
 src/supervaizer/server.py                     |  50 ++++++++
 9 files changed, 355 insertions(+), 6 deletions(-)

diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index 30cc75a..e0169b0 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -21,6 +21,16 @@ All notable changes to this project will be documented in this file.
 
 ### v0.11.0
 
+- **Custom routes** — Agents can mount their own FastAPI routers via `custom_routes` field on Agent. Supervaizer mounts them at `/agents/{slug}/api/` without inspecting or managing the routes. Enables agents to expose tool endpoints, webhooks, or any HTTP API alongside the workbench.
+
+- **Scheduled steps** — CaseNodeUpdate gains `scheduled_at`, `scheduled_method`, `scheduled_params`, `scheduled_status` fields. Steps with `scheduled_at` are deferred until the time arrives. A background executor polls every 60s and calls the agent method. Workbench shows countdown, "Execute now", and "Cancel" controls on pending scheduled steps. Enables time-based orchestration (call scheduling, retries with backoff, follow-up actions).
+  - Model: `CaseNodeUpdate.scheduled_at/scheduled_method/scheduled_params/scheduled_status`
+  - Executor: `_run_scheduled_step_loop` in server.py (local mode)
+  - Routes: `POST/PATCH .../steps/{case_id}/{step_index}/execute|cancel|schedule`
+  - UI: status badges, execute now / cancel buttons in workbench monitor
+  - Case: `cancel_scheduled_steps()` for job stop cascading
+  - Cases: `get_due_scheduled_steps()` for executor polling
+
 - **Job Poll mechanism** — New optional `job_poll` method in `AgentMethods` for manual external service polling. When defined, the workbench shows a "Check for updates" button on active jobs. Clicking it calls the agent's poll handler, which checks external services (email inboxes, call status APIs, etc.) and updates cases accordingly. Enables local development without webhooks — production uses real-time webhooks, local mode uses the poll button.
   - `AgentMethods`: new `job_poll: AgentMethod | None` field
   - Route: `POST /workbench/jobs/{job_id}/poll` triggers the agent's poll handler
diff --git a/src/supervaizer/admin/static/js/workbench-form.js b/src/supervaizer/admin/static/js/workbench-form.js
index f1d3bb8..392ffe7 100644
--- a/src/supervaizer/admin/static/js/workbench-form.js
+++ b/src/supervaizer/admin/static/js/workbench-form.js
@@ -151,6 +151,43 @@ class WorkbenchForm {
         }
     }
 
+    async executeStepNow(caseId, stepIndex) {
+        if (!this.activeJobId) return;
+        try {
+            const response = await fetch(
+                `${this.basePath}/jobs/${this.activeJobId}/steps/${caseId}/${stepIndex}/execute`,
+                { method: 'POST', headers: { 'X-API-Key': this.getApiKey() } },
+            );
+            const result = await response.json();
+            if (!response.ok) {
+                this.onError(result.detail || 'Execute failed');
+            } else {
+                this.onError('');
+                this.refreshMonitor(true);
+            }
+        } catch (e) {
+            this.onError(`Network error: ${e.message}`);
+        }
+    }
+
+    async cancelStep(caseId, stepIndex) {
+        if (!this.activeJobId) return;
+        try {
+            const response = await fetch(
+                `${this.basePath}/jobs/${this.activeJobId}/steps/${caseId}/${stepIndex}/cancel`,
+                { method: 'POST', headers: { 'X-API-Key': this.getApiKey() } },
+            );
+            if (!response.ok) {
+                this.onError('Cancel failed');
+            } else {
+                this.onError('');
+                this.refreshMonitor(true);
+            }
+        } catch (e) {
+            this.onError(`Network error: ${e.message}`);
+        }
+    }
+
     async stopJob(jobId) {
         const targetJobId = jobId || this.activeJobId;
         if (!targetJobId) return;
diff --git a/src/supervaizer/admin/templates/components/dialog_renderer.html b/src/supervaizer/admin/templates/components/dialog_renderer.html
index 8075fbe..6fa2651 100644
--- a/src/supervaizer/admin/templates/components/dialog_renderer.html
+++ b/src/supervaizer/admin/templates/components/dialog_renderer.html
@@ -77,7 +77,45 @@
         {% elif content_type == "code" %}
         <pre class="p-3 text-xs text-gray-700 max-h-64 overflow-y-auto bg-gray-50 font-mono whitespace-pre-wrap">{{ content_raw }}</pre>
         {% else %}
-        <div class="p-3 text-sm text-gray-700 max-h-64 overflow-y-auto whitespace-pre-wrap">{{ content_raw }}</div>
+        {# Text/markdown: render with lightweight markdown via Alpine #}
+        <div class="p-3 text-sm text-gray-700 max-h-96 overflow-y-auto prose prose-sm prose-headings:text-gray-800 prose-headings:font-semibold"
+             x-data="{ rendered: '' }"
+             x-init="
+                let raw = {{ content_raw | tojson }};
+                // Unescape literal \\n to real newlines
+                raw = raw.replace(/\\\\n/g, '\n').replace(/\\n/g, '\n');
+                // Lightweight markdown rendering
+                let lines = raw.split('\n');
+                let html = '';
+                let inList = false;
+                for (let line of lines) {
+                    let trimmed = line.trim();
+                    if (trimmed.startsWith('## ')) {
+                        if (inList) { html += '</ul>'; inList = false; }
+                        html += '<h3>' + trimmed.slice(3) + '</h3>';
+                    } else if (trimmed.startsWith('# ')) {
+                        if (inList) { html += '</ul>'; inList = false; }
+                        html += '<h2>' + trimmed.slice(2) + '</h2>';
+                    } else if (trimmed.startsWith('- ')) {
+                        if (!inList) { html += '<ul class=\"list-disc pl-4 my-1\">'; inList = true; }
+                        html += '<li>' + trimmed.slice(2) + '</li>';
+                    } else if (trimmed === '') {
+                        if (inList) { html += '</ul>'; inList = false; }
+                        html += '<br>';
+                    } else {
+                        if (inList) { html += '</ul>'; inList = false; }
+                        html += '<p class=\"my-0.5\">' + trimmed + '</p>';
+                    }
+                }
+                if (inList) html += '</ul>';
+                // Inline formatting: bold, italic, code
+                html = html.replace(/\*\*(.+?)\*\*/g, '<strong>$1</strong>');
+                html = html.replace(/\*(.+?)\*/g, '<em>$1</em>');
+                html = html.replace(/`(.+?)`/g, '<code class=\"text-xs bg-gray-100 px-1 py-0.5 rounded\">$1</code>');
+                rendered = html;
+             "
+             x-html="rendered">
+        </div>
         {% endif %}
     </div>
     {% endif %}
@@ -178,7 +216,27 @@
         </template>
     </div>
     {% else %}
-    <div class="p-3 text-sm text-gray-700 max-h-48 overflow-y-auto whitespace-pre-wrap">{{ content_raw }}</div>
+    <div class="p-3 text-sm text-gray-700 max-h-48 overflow-y-auto prose prose-sm"
+         x-data="{ rendered: '' }"
+         x-init="
+            let raw = {{ content_raw | tojson }};
+            raw = raw.replace(/\\\\n/g, '\n').replace(/\\n/g, '\n');
+            let lines = raw.split('\n');
+            let html = '';
+            for (let line of lines) {
+                let t = line.trim();
+                if (t.startsWith('## ')) html += '<h3>' + t.slice(3) + '</h3>';
+                else if (t.startsWith('# ')) html += '<h2>' + t.slice(2) + '</h2>';
+                else if (t.startsWith('- ')) html += '<p class=\"my-0.5\">• ' + t.slice(2) + '</p>';
+                else if (t === '') html += '<br>';
+                else html += '<p class=\"my-0.5\">' + t + '</p>';
+            }
+            html = html.replace(/\*\*(.+?)\*\*/g, '<strong>$1</strong>');
+            html = html.replace(/`(.+?)`/g, '<code class=\"text-xs bg-gray-100 px-1 rounded\">$1</code>');
+            rendered = html;
+         "
+         x-html="rendered">
+    </div>
     {% endif %}
 </div>
 {% endmacro %}
diff --git a/src/supervaizer/admin/templates/workbench.html b/src/supervaizer/admin/templates/workbench.html
index bdb7ca5..6643c3e 100644
--- a/src/supervaizer/admin/templates/workbench.html
+++ b/src/supervaizer/admin/templates/workbench.html
@@ -389,7 +389,7 @@ <h3 class="text-sm font-semibold text-gray-700">Execution Monitor</h3>
 </div>{# end workbenchPage #}
 
 {# ── Workbench JS ────────────────────────────────────────────────── #}
-<script src="/admin/static/js/workbench-form.js?v=13"></script>
+<script src="/admin/static/js/workbench-form.js?v=14"></script>
 <script>
     function applyConsoleFilter() {
         var container = document.getElementById('console-log-container');
diff --git a/src/supervaizer/admin/templates/workbench_monitor.html b/src/supervaizer/admin/templates/workbench_monitor.html
index 34d6dfb..5f2e13a 100644
--- a/src/supervaizer/admin/templates/workbench_monitor.html
+++ b/src/supervaizer/admin/templates/workbench_monitor.html
@@ -168,6 +168,32 @@
                             {% endif %}
                         </div>
                     </div>
+                    {# Scheduled step controls #}
+                    {% set is_sched = step.scheduled_at is not none if step.scheduled_at is defined else false %}
+                    {% if is_sched %}
+                    <div class="ml-8 mt-1 flex items-center gap-2 text-xs">
+                        {% set sched_status = step.scheduled_status if step.scheduled_status is defined else "pending" %}
+                        {% if sched_status == "pending" %}
+                        <span class="text-blue-600 font-medium">&#9200; Scheduled: {{ step.scheduled_at }}</span>
+                        <button onclick="window.workbenchForm && window.workbenchForm.executeStepNow('{{ case.id }}', {{ loop.index0 }})"
+                                class="bg-blue-50 text-blue-700 px-2 py-0.5 rounded hover:bg-blue-100 transition-colors">
+                            Execute now
+                        </button>
+                        <button onclick="window.workbenchForm && window.workbenchForm.cancelStep('{{ case.id }}', {{ loop.index0 }})"
+                                class="bg-red-50 text-red-600 px-2 py-0.5 rounded hover:bg-red-100 transition-colors">
+                            Cancel
+                        </button>
+                        {% elif sched_status == "executing" %}
+                        <span class="text-amber-600">&#9889; Executing...</span>
+                        {% elif sched_status == "completed" %}
+                        <span class="text-green-600">&#9989; Executed</span>
+                        {% elif sched_status == "failed" %}
+                        <span class="text-red-600">&#10060; Failed</span>
+                        {% elif sched_status == "cancelled" %}
+                        <span class="text-gray-400 line-through">Cancelled</span>
+                        {% endif %}
+                    </div>
+                    {% endif %}
                     {# Show reply content as a blockquote #}
                     {% if step_payload is mapping and step_payload.get("reply") %}
                     <div class="mx-4 mb-2 rounded border border-blue-200 bg-blue-50 p-3">
diff --git a/src/supervaizer/admin/workbench_routes.py b/src/supervaizer/admin/workbench_routes.py
index 764fe33..ba64eb4 100644
--- a/src/supervaizer/admin/workbench_routes.py
+++ b/src/supervaizer/admin/workbench_routes.py
@@ -544,6 +544,119 @@ async def workbench_answer_hitl(
             "message": "HITL answer submitted and dispatched",
         })
 
+    @router.post(
+        "/agents/{slug}/workbench/jobs/{job_id}/steps/{case_id}/{step_index}/execute"
+    )
+    async def workbench_execute_step(
+        request: Request, slug: str, job_id: str, case_id: str, step_index: int
+    ) -> Response:
+        """Execute a scheduled step immediately."""
+        from supervaizer.server import _execute_scheduled_method
+
+        get_agent_by_slug(request, slug)
+
+        case = Cases().get_case(case_id, job_id=job_id)
+        if not case:
+            raise HTTPException(status_code=404, detail=f"Case '{case_id}' not found")
+
+        if step_index < 0 or step_index >= len(case.updates):
+            raise HTTPException(status_code=404, detail="Step not found")
+
+        update = case.updates[step_index]
+        if getattr(update, "scheduled_at", None) is None:
+            raise HTTPException(status_code=400, detail="Step is not a scheduled step")
+        if getattr(update, "scheduled_status", None) != "pending":
+            raise HTTPException(
+                status_code=409,
+                detail=f"Step is not pending (current: {getattr(update, 'scheduled_status', 'unknown')})",
+            )
+
+        object.__setattr__(update, "scheduled_status", "executing")
+        try:
+            if update.scheduled_method:
+                _execute_scheduled_method(
+                    update.scheduled_method,
+                    update.scheduled_params or {},
+                )
+            object.__setattr__(update, "scheduled_status", "completed")
+            return JSONResponse({"status": "completed", "message": "Step executed successfully"})
+        except Exception as e:
+            object.__setattr__(update, "scheduled_status", "failed")
+            log.error(f"[Workbench] Scheduled step execute failed: {e}")
+            return JSONResponse(
+                {"status": "failed", "message": f"Execution failed: {e}"},
+                status_code=500,
+            )
+
+    @router.post(
+        "/agents/{slug}/workbench/jobs/{job_id}/steps/{case_id}/{step_index}/cancel"
+    )
+    async def workbench_cancel_step(
+        request: Request, slug: str, job_id: str, case_id: str, step_index: int
+    ) -> Response:
+        """Cancel a pending scheduled step."""
+        get_agent_by_slug(request, slug)
+
+        case = Cases().get_case(case_id, job_id=job_id)
+        if not case:
+            raise HTTPException(status_code=404, detail=f"Case '{case_id}' not found")
+
+        if step_index < 0 or step_index >= len(case.updates):
+            raise HTTPException(status_code=404, detail="Step not found")
+
+        update = case.updates[step_index]
+        if getattr(update, "scheduled_at", None) is None:
+            raise HTTPException(status_code=400, detail="Step is not a scheduled step")
+        if getattr(update, "scheduled_status", None) != "pending":
+            raise HTTPException(
+                status_code=409,
+                detail=f"Step is not pending (current: {getattr(update, 'scheduled_status', 'unknown')})",
+            )
+
+        object.__setattr__(update, "scheduled_status", "cancelled")
+        return JSONResponse({"status": "cancelled", "message": "Step cancelled"})
+
+    @router.patch(
+        "/agents/{slug}/workbench/jobs/{job_id}/steps/{case_id}/{step_index}/schedule"
+    )
+    async def workbench_reschedule_step(
+        request: Request, slug: str, job_id: str, case_id: str, step_index: int
+    ) -> Response:
+        """Reschedule a pending scheduled step."""
+        get_agent_by_slug(request, slug)
+
+        case = Cases().get_case(case_id, job_id=job_id)
+        if not case:
+            raise HTTPException(status_code=404, detail=f"Case '{case_id}' not found")
+
+        if step_index < 0 or step_index >= len(case.updates):
+            raise HTTPException(status_code=404, detail="Step not found")
+
+        update = case.updates[step_index]
+        if getattr(update, "scheduled_at", None) is None:
+            raise HTTPException(status_code=400, detail="Step is not a scheduled step")
+        if getattr(update, "scheduled_status", None) != "pending":
+            raise HTTPException(
+                status_code=409,
+                detail=f"Step is not pending (current: {getattr(update, 'scheduled_status', 'unknown')})",
+            )
+
+        body = await request.json()
+        new_scheduled_at = body.get("scheduled_at")
+        if not new_scheduled_at:
+            raise HTTPException(status_code=400, detail="scheduled_at is required")
+
+        try:
+            new_dt = datetime.fromisoformat(new_scheduled_at.replace("Z", "+00:00"))
+        except (ValueError, AttributeError):
+            raise HTTPException(status_code=400, detail="Invalid datetime format")
+
+        object.__setattr__(update, "scheduled_at", new_dt)
+        return JSONResponse({
+            "status": "rescheduled",
+            "message": f"Step rescheduled to {new_dt.isoformat()}",
+        })
+
     @router.get("/agents/{slug}/workbench/console", response_class=HTMLResponse)
     async def workbench_console(request: Request, slug: str) -> Response:
         """HTMX partial — returns recent console log entries."""
diff --git a/src/supervaizer/agent.py b/src/supervaizer/agent.py
index 25e8445..7db8892 100644
--- a/src/supervaizer/agent.py
+++ b/src/supervaizer/agent.py
@@ -606,8 +606,15 @@ class AgentAbstract(SvBaseModel):
         default="supervaize_instructions.html",
         description="Path where the supervaize instructions page is served (relative to agent path)",
     )
+    custom_routes: Any | None = Field(
+        default=None,
+        description="Optional FastAPI APIRouter with custom routes for this agent",
+        exclude=True,
+    )
 
-    model_config = cast(ConfigDict, {"reference_group": "Core"})
+    model_config = cast(
+        ConfigDict, {"reference_group": "Core", "arbitrary_types_allowed": True}
+    )
 
 
 class Agent(AgentAbstract):
@@ -629,6 +636,7 @@ def __init__(
         server_agent_onboarding_status: str | None = None,
         server_encrypted_parameters: str | None = None,
         max_execution_time: int = 60 * 60,  # 1 hour (in seconds)
+        custom_routes: Any | None = None,
         **kwargs: Any,
     ) -> None:
         """
@@ -681,6 +689,7 @@ def __init__(
             server_agent_onboarding_status=server_agent_onboarding_status,
             server_encrypted_parameters=server_encrypted_parameters,
             max_execution_time=max_execution_time,
+            custom_routes=custom_routes,
             **kwargs,
         )
 
diff --git a/src/supervaizer/case.py b/src/supervaizer/case.py
index eafc038..881445b 100644
--- a/src/supervaizer/case.py
+++ b/src/supervaizer/case.py
@@ -5,7 +5,7 @@
 # https://mozilla.org/MPL/2.0/.
 
 
-from datetime import datetime
+from datetime import datetime, timezone
 from enum import Enum
 from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
 
@@ -35,6 +35,10 @@ class CaseNodeUpdate(SvBaseModel):
     payload: Optional[Dict[str, Any]] = None
     is_final: bool = False
     error: Optional[str] = None
+    scheduled_at: datetime | None = None  # When to execute (UTC)
+    scheduled_method: str | None = None  # Agent method dotted path
+    scheduled_params: Optional[Dict[str, Any]] = None  # Params for the method
+    scheduled_status: str | None = None  # pending, executing, completed, failed, cancelled
 
     def __init__(
         self,
@@ -44,6 +48,10 @@ def __init__(
         is_final: bool = False,
         index: int | None = None,
         error: Optional[str] = None,
+        scheduled_at: datetime | None = None,
+        scheduled_method: str | None = None,
+        scheduled_params: Dict[str, Any] | None = None,
+        scheduled_status: str | None = None,
     ) -> None:
         """Initialize a CaseNodeUpdate.
 
@@ -85,6 +93,10 @@ def __init__(
             "is_final": is_final,
             "index": index,
             "error": error,
+            "scheduled_at": scheduled_at,
+            "scheduled_method": scheduled_method,
+            "scheduled_params": scheduled_params,
+            "scheduled_status": scheduled_status,
         }
         object.__setattr__(self, "__dict__", {})
         object.__setattr__(self, "__pydantic_fields_set__", set())
@@ -95,6 +107,10 @@ def __init__(
         for key, value in values.items():
             setattr(self, key, value)
 
+    @property
+    def is_scheduled(self) -> bool:
+        return self.scheduled_at is not None
+
     @property
     def registration_info(self) -> Dict[str, Any]:
         """Returns registration info for the case node update"""
@@ -102,7 +118,7 @@ def registration_info(self) -> Dict[str, Any]:
         serialized_payload = (
             self.serialize_value(self.payload) if self.payload else None
         )
-        return {
+        info = {
             "index": self.index,
             "name": self.name,
             "error": self.error,
@@ -110,6 +126,13 @@ def registration_info(self) -> Dict[str, Any]:
             "payload": serialized_payload,
             "is_final": self.is_final,
         }
+        if self.scheduled_at:
+            info["scheduled_at"] = self.scheduled_at.isoformat()
+        if self.scheduled_method:
+            info["scheduled_method"] = self.scheduled_method
+        if self.scheduled_status:
+            info["scheduled_status"] = self.scheduled_status
+        return info
 
 
 class CaseNodeType(Enum):
@@ -289,6 +312,13 @@ def close(
         storage = StorageManager()
         storage.save_object("Case", self.to_dict)
 
+    def cancel_scheduled_steps(self):
+        """Cancel all pending scheduled steps for this case."""
+        for update in self.updates:
+            if (getattr(update, 'scheduled_at', None) is not None
+                    and getattr(update, 'scheduled_status', None) == 'pending'):
+                object.__setattr__(update, 'scheduled_status', 'cancelled')
+
     @property
     def registration_info(self) -> Dict[str, Any]:
         """Returns registration info for the case"""
@@ -425,6 +455,22 @@ def get_job_cases(self, job_id: str) -> dict[str, "Case"]:
         """
         return self.cases_by_job.get(job_id, {})
 
+    def get_due_scheduled_steps(self) -> list[tuple]:
+        """Return all steps where scheduled_at <= now() and status == 'pending'.
+
+        Returns list of (case, step_index, update) tuples.
+        """
+        now = datetime.now(timezone.utc)
+        due = []
+        for job_cases in self.cases_by_job.values():
+            for case_id, case in job_cases.items():
+                for i, update in enumerate(case.updates):
+                    if (getattr(update, 'scheduled_at', None) is not None
+                            and getattr(update, 'scheduled_status', None) == 'pending'
+                            and update.scheduled_at <= now):
+                        due.append((case, i, update))
+        return due
+
     def __contains__(self, case_id: str) -> bool:
         """Check if case exists in any job's registry"""
         return any(case_id in cases for cases in self.cases_by_job.values())
diff --git a/src/supervaizer/server.py b/src/supervaizer/server.py
index 02eb281..1782a92 100644
--- a/src/supervaizer/server.py
+++ b/src/supervaizer/server.py
@@ -4,6 +4,7 @@
 # If a copy of the MPL was not distributed with this file, you can obtain one at
 # https://mozilla.org/MPL/2.0/.
 
+import asyncio
 import os
 import secrets
 import sys
@@ -188,6 +189,42 @@ def get_server_info_from_live(server_instance: "Server") -> ServerInfo:
     )
 
 
+def _execute_scheduled_method(method_path: str, params: dict) -> Any:
+    """Execute a method by its full dotted path (module.func)."""
+    module_name, func_name = method_path.rsplit(".", 1)
+    module = __import__(module_name, fromlist=[func_name])
+    method = getattr(module, func_name)
+    return method(**params)
+
+
+async def _run_scheduled_step_loop(server: "Server") -> None:
+    """Poll for due scheduled steps every 60 seconds and execute them."""
+    from supervaizer.case import Cases
+
+    while True:
+        await asyncio.sleep(60)
+        try:
+            cases = Cases()
+            due_steps = cases.get_due_scheduled_steps()
+            for _case, _step_index, update in due_steps:
+                if not update.scheduled_method:
+                    continue
+                try:
+                    object.__setattr__(update, "scheduled_status", "executing")
+                    log.info(f"[Scheduled step] Executing: {update.name}")
+                    _execute_scheduled_method(
+                        update.scheduled_method,
+                        update.scheduled_params or {},
+                    )
+                    object.__setattr__(update, "scheduled_status", "completed")
+                    log.info(f"[Scheduled step] Completed: {update.name}")
+                except Exception as exc:
+                    object.__setattr__(update, "scheduled_status", "failed")
+                    log.error(f"[Scheduled step] Failed: {update.name}: {exc}")
+        except Exception as exc:
+            log.error(f"[Scheduled step loop] Error: {exc}")
+
+
 class ServerAbstract(SvBaseModel):
     """
     API Server for the Supervaize Controller.
@@ -493,6 +530,14 @@ async def validation_exception_handler(
             log.info("[Server launch] 📢 Deploy A2A routes  ")
             self.app.include_router(create_a2a_routes(self))
 
+        # Mount agent custom routes
+        for agent in self.agents:
+            if agent.custom_routes:
+                self.app.include_router(
+                    agent.custom_routes,
+                    prefix=f"/agents/{agent.slug}/api",
+                )
+
         # Store server instance on app state for admin routes to access
         self.app.state.server = self
 
@@ -555,6 +600,11 @@ async def get_current_server() -> "Server":
         else:
             log.info("[Server launch] API Key authentication disabled")
 
+        # Start the scheduled-step executor loop on app startup
+        @self.app.on_event("startup")
+        async def _start_scheduled_executor() -> None:
+            asyncio.create_task(_run_scheduled_step_loop(self))
+
         if not self.public_url:
             self.public_url = f"{self.scheme}://{self.host}:{self.port}"
 

From ce13b5bf32f74ef7c21edce892b9f474a369021d Mon Sep 17 00:00:00 2001
From: Alain Prasquier <alain@supervaize.com>
Date: Sat, 21 Mar 2026 12:43:42 +0200
Subject: [PATCH 2/2] =?UTF-8?q?=E2=9C=A8=20feat:=20add=20dev-version=20tog?=
 =?UTF-8?q?gle=20and=20refine=20API=20typesAdd=20a=20small=20dev-version?=
 =?UTF-8?q?=20tooling=20and=20a=20few=20typing=20and=20fixes.?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Add tools/dev_version to a PEP440 .devN suffix both src/svaizer/__version__. and pyproject.toml [tool.bumpversion].
 provides `just-dev on|` to enable/disable the .dev0 canonical suffix for local/dev workflows so versions remain in sync without editing changelog or historical docs.
- Addfile recipe version-dev cmd that invokes the new script.
 Document usage and caution about using tag_version while dev mode is on- Bump package VERSION to.11.0.dev0 and keep APItelemetry constants- Export AgentResponse from supervaizer.____ and rebuild model import time so forward refs are resolved all imports.
 Change CaseNode.factory annotation Any (was Optional[Callable...])
 and add a comment explaining this emitting JSON in Pydantic/FastAPI OpenAPI (Pydantic2.12+).

These changes make local dev versioning deterministic and fix schemageneration while a missing response.
---
 docs/CHANGELOG.md              |  15 ++++-
 justfile                       |   6 ++
 pyproject.toml                 |  10 ++-
 src/supervaizer/__init__.py    |   2 +
 src/supervaizer/__version__.py |   2 +-
 src/supervaizer/agent.py       |   3 +-
 src/supervaizer/case.py        |   7 ++-
 tools/dev_version.py           | 109 +++++++++++++++++++++++++++++++++
 8 files changed, 145 insertions(+), 9 deletions(-)
 create mode 100644 tools/dev_version.py

diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md
index e0169b0..a0917a3 100644
--- a/docs/CHANGELOG.md
+++ b/docs/CHANGELOG.md
@@ -19,9 +19,7 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
-### v0.11.0
-
-- **Custom routes** — Agents can mount their own FastAPI routers via `custom_routes` field on Agent. Supervaizer mounts them at `/agents/{slug}/api/` without inspecting or managing the routes. Enables agents to expose tool endpoints, webhooks, or any HTTP API alongside the workbench.
+### Added
 
 - **Scheduled steps** — CaseNodeUpdate gains `scheduled_at`, `scheduled_method`, `scheduled_params`, `scheduled_status` fields. Steps with `scheduled_at` are deferred until the time arrives. A background executor polls every 60s and calls the agent method. Workbench shows countdown, "Execute now", and "Cancel" controls on pending scheduled steps. Enables time-based orchestration (call scheduling, retries with backoff, follow-up actions).
   - Model: `CaseNodeUpdate.scheduled_at/scheduled_method/scheduled_params/scheduled_status`
@@ -31,6 +29,17 @@ All notable changes to this project will be documented in this file.
   - Case: `cancel_scheduled_steps()` for job stop cascading
   - Cases: `get_due_scheduled_steps()` for executor polling
 
+### Fixed
+
+- **OpenAPI / JSON Schema (Pydantic 2.12+)** — Building the full FastAPI schema (`GET /openapi.json`, Swagger UI) could raise `PydanticInvalidForJsonSchema: … CallableSchema`. Causes and fixes:
+  - **`AgentMethodAbstract.model_config["example_dict"]`** — The sample field dict used `"type": str` (Python’s `str` builtin). Pydantic’s JSON Schema generator walks that value and emits a callable schema for builtins. **Fix:** use the string `"str"` (documentation-only example data, not a type annotation).
+  - **`AgentResponse` nested schemas** — `AgentResponse` (which embeds `AgentMethods` → `CaseNodes` → `CaseNode`) must be rebuilt after other models so OpenAPI sees consistent inner definitions. **Fix:** call `AgentResponse.model_rebuild()` after `Case.model_rebuild()` in `supervaizer/__init__.py`.
+  - **`CaseNode.factory`** — Runtime value is `Callable[..., CaseNodeUpdate] | None`, which cannot appear in JSON Schema. `Annotated[..., SkipJsonSchema()]` was insufficient: Pydantic 2.12 still registered a `Callable` core definition for `$ref` resolution and OpenAPI failed. **Fix:** declare the field as `Any` (documented in code); behaviour and `registration_info` are unchanged.
+
+- **Custom routes** — Agents can mount their own FastAPI routers via `custom_routes` field on Agent. Supervaizer mounts them at `/agents/{slug}/api/` without inspecting or managing the routes. Enables agents to expose tool endpoints, webhooks, or any HTTP API alongside the workbench.
+
+## v0.11.0
+
 - **Job Poll mechanism** — New optional `job_poll` method in `AgentMethods` for manual external service polling. When defined, the workbench shows a "Check for updates" button on active jobs. Clicking it calls the agent's poll handler, which checks external services (email inboxes, call status APIs, etc.) and updates cases accordingly. Enables local development without webhooks — production uses real-time webhooks, local mode uses the poll button.
   - `AgentMethods`: new `job_poll: AgentMethod | None` field
   - Route: `POST /workbench/jobs/{job_id}/poll` triggers the agent's poll handler
diff --git a/justfile b/justfile
index e6905d2..297e475 100644
--- a/justfile
+++ b/justfile
@@ -56,6 +56,12 @@ env_sync_all:
 build:
     hatch build
 
+# Toggle PEP 440 .dev0 on canonical version (syncs src/supervaizer/__version__.py + pyproject [tool.bumpversion]).
+# Does not edit CHANGELOG or historical docs. After `on`, avoid `just tag_version` until `off` (tags should be release-only).
+# Usage: just version-dev on | just version-dev off
+version-dev cmd:
+    uv run python tools/dev_version.py {{cmd}}
+
 # Reusable recipe to bump version
 _bump_version bump_type:
     @echo "VERSION BUMP IN CICD - not running: hatch version {{bump_type}} "
diff --git a/pyproject.toml b/pyproject.toml
index 2e6be28..b37a3fb 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -126,16 +126,22 @@ mypy_path = "src"
 disallow_any_expr = false
 
 [tool.bumpversion]
-current_version = "0.10.29"
+current_version = "0.11.0.dev0"
 commit = true
 tag = true
 tag_name = "v{new_version}"
 tag_message = "Release {new_version}"
 message = "Bump version: {current_version} → {new_version}"
-parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)"
+# Optional .devN is stripped on bump (CI publishes X.Y.Z only); use `just version-dev on` locally.
+parse = "(?P<major>\\d+)\\.(?P<minor>\\d+)\\.(?P<patch>\\d+)(\\.dev(?P<dev>\\d+))?"
 serialize = ["{major}.{minor}.{patch}"]
 
 [[tool.bumpversion.files]]
 filename = "src/supervaizer/__version__.py"
 search = 'VERSION = "{current_version}"'
 replace = 'VERSION = "{new_version}"'
+
+[[tool.bumpversion.files]]
+filename = "pyproject.toml"
+search = 'current_version = "{current_version}"'
+replace = 'current_version = "{new_version}"'
diff --git a/src/supervaizer/__init__.py b/src/supervaizer/__init__.py
index c54d6a0..7244bfa 100644
--- a/src/supervaizer/__init__.py
+++ b/src/supervaizer/__init__.py
@@ -14,6 +14,7 @@
     AgentMethodParams,
     AgentMethods,
     AgentMethodField,
+    AgentResponse,
     FieldTypeEnum,
 )
 from supervaizer.case import (
@@ -99,3 +100,4 @@
 
 # Rebuild models to resolve forward references after all imports are done
 Case.model_rebuild()
+AgentResponse.model_rebuild()
diff --git a/src/supervaizer/__version__.py b/src/supervaizer/__version__.py
index 9613d90..400b500 100644
--- a/src/supervaizer/__version__.py
+++ b/src/supervaizer/__version__.py
@@ -5,6 +5,6 @@
 # https://mozilla.org/MPL/2.0/.
 
 
-VERSION = "0.10.29"
+VERSION = "0.11.0.dev0"
 API_VERSION = "v1"
 TELEMETRY_VERSION = "v1"
diff --git a/src/supervaizer/agent.py b/src/supervaizer/agent.py
index 7db8892..5c458d2 100644
--- a/src/supervaizer/agent.py
+++ b/src/supervaizer/agent.py
@@ -208,7 +208,8 @@ class AgentMethodAbstract(BaseModel):
                 "fields": [
                     {
                         "name": "Company to research",
-                        "type": str,
+                        # String label for docs / JSON Schema — not the `str` type object (Pydantic 2.12+ OpenAPI)
+                        "type": "str",
                         "field_type": "CharField",
                         "max_length": 100,
                         "required": True,
diff --git a/src/supervaizer/case.py b/src/supervaizer/case.py
index 881445b..0804975 100644
--- a/src/supervaizer/case.py
+++ b/src/supervaizer/case.py
@@ -7,7 +7,7 @@
 
 from datetime import datetime, timezone
 from enum import Enum
-from typing import TYPE_CHECKING, Any, Callable, Dict, List, Optional
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 import shortuuid
 from pydantic import ConfigDict
@@ -159,7 +159,10 @@ class CaseNode(SvBaseModel):
 
     name: str
     type: CaseNodeType
-    factory: Optional[Callable[..., CaseNodeUpdate]] = None
+    # Runtime type is ``Callable[..., CaseNodeUpdate] | None``. Declared as ``Any`` so Pydantic JSON
+    # Schema / FastAPI OpenAPI never emit ``CallableSchema`` (SkipJsonSchema on Callable is not
+    # enough for $ref / definitions generation in Pydantic 2.12+).
+    factory: Any = None
     description: str = ""
     can_be_confirmed: bool = False
 
diff --git a/tools/dev_version.py b/tools/dev_version.py
new file mode 100644
index 0000000..545ca18
--- /dev/null
+++ b/tools/dev_version.py
@@ -0,0 +1,109 @@
+#!/usr/bin/env python3
+"""Toggle PEP 440 .devN suffix for local dev (keeps __version__.py and bumpversion config in sync)."""
+
+from __future__ import annotations
+
+import argparse
+import re
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+VERSION_FILE = ROOT / "src" / "supervaizer" / "__version__.py"
+PYPROJECT = ROOT / "pyproject.toml"
+
+VERSION_LINE_RE = re.compile(r'^(VERSION\s*=\s*")([^"]+)(")', re.M)
+CURRENT_VERSION_LINE_RE = re.compile(r'^(current_version\s*=\s*")([^"]+)(")', re.M)
+
+RELEASE_RE = re.compile(r"^\d+\.\d+\.\d+$")
+DEV_RE = re.compile(r"^(\d+)\.(\d+)\.(\d+)\.dev(\d+)$")
+
+
+def _read_version_py() -> str:
+    text = VERSION_FILE.read_text(encoding="utf-8")
+    m = VERSION_LINE_RE.search(text)
+    if not m:
+        msg = f"Could not find VERSION = \"...\" line in {VERSION_FILE}"
+        raise SystemExit(msg)
+    return m.group(2)
+
+
+def _write_version_py(new_ver: str) -> None:
+    text = VERSION_FILE.read_text(encoding="utf-8")
+    new_text, n = VERSION_LINE_RE.subn(rf"\g<1>{new_ver}\g<3>", text, count=1)
+    if n != 1:
+        msg = f"Failed to replace VERSION in {VERSION_FILE}"
+        raise SystemExit(msg)
+    VERSION_FILE.write_text(new_text, encoding="utf-8", newline="\n")
+
+
+def _read_pyproject_version() -> str:
+    text = PYPROJECT.read_text(encoding="utf-8")
+    m = CURRENT_VERSION_LINE_RE.search(text)
+    if not m:
+        msg = f"Could not find current_version = \"...\" under [tool.bumpversion] in {PYPROJECT}"
+        raise SystemExit(msg)
+    return m.group(2)
+
+
+def _write_pyproject_version(new_ver: str) -> None:
+    text = PYPROJECT.read_text(encoding="utf-8")
+    new_text, n = CURRENT_VERSION_LINE_RE.subn(rf"\g<1>{new_ver}\g<3>", text, count=1)
+    if n != 1:
+        msg = f"Failed to replace current_version in {PYPROJECT}"
+        raise SystemExit(msg)
+    PYPROJECT.write_text(new_text, encoding="utf-8", newline="\n")
+
+
+def _assert_sync() -> str:
+    v_py = _read_version_py()
+    v_toml = _read_pyproject_version()
+    if v_py != v_toml:
+        msg = (
+            "Version mismatch — fix manually or run one of:\n"
+            f"  {VERSION_FILE.relative_to(ROOT)}: {v_py!r}\n"
+            f"  {PYPROJECT.relative_to(ROOT)}: {v_toml!r}"
+        )
+        raise SystemExit(msg)
+    return v_py
+
+
+def cmd_on() -> None:
+    v = _assert_sync()
+    if DEV_RE.match(v):
+        print(f"Already a dev version ({v!r}); nothing to do.")
+        return
+    if not RELEASE_RE.match(v):
+        msg = f"Expected release X.Y.Z (got {v!r}); cannot add .dev0."
+        raise SystemExit(msg)
+    new_v = f"{v}.dev0"
+    _write_version_py(new_v)
+    _write_pyproject_version(new_v)
+    print(f"Dev on:  {v} → {new_v}")
+
+
+def cmd_off() -> None:
+    v = _assert_sync()
+    m = DEV_RE.match(v)
+    if not m:
+        print(f"No .devN suffix ({v!r}); nothing to do.")
+        return
+    base = f"{m.group(1)}.{m.group(2)}.{m.group(3)}"
+    _write_version_py(base)
+    _write_pyproject_version(base)
+    print(f"Dev off: {v} → {base}")
+
+
+def main() -> None:
+    p = argparse.ArgumentParser(description=__doc__)
+    sub = p.add_subparsers(dest="cmd", required=True)
+    sub.add_parser("on", help="Append .dev0 (X.Y.Z → X.Y.Z.dev0)")
+    sub.add_parser("off", help="Strip .devN (X.Y.Z.devN → X.Y.Z)")
+    args = p.parse_args()
+    if args.cmd == "on":
+        cmd_on()
+    else:
+        cmd_off()
+
+
+if __name__ == "__main__":
+    main()