✨ feat: custom_routes + scheduled steps + executor + workbench UI

docs/CHANGELOG.md

@@ -19,7 +19,26 @@ All notable changes to this project will be documented in this file.
 
 ## Unreleased
 
-### v0.11.0
+### 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`
+  - 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
+
+### 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

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}} "

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}"'

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()

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"

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;

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 %}

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');

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">

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."""