feat: add reasoning support to conversations

- Introduced a per-conversation reasoning level (`reasoning_effort`) to control the model's thinking capability (off | low | medium | high).
- Updated the database schema to include the `reasoning_effort` column in the `conversations` table.
- Enhanced the `get_llm` function to accept a `reasoning` parameter, which maps to provider-specific thinking parameters.
- Modified conversation creation and update endpoints to handle the new `reasoning_effort` field.
- Updated frontend components to allow users to select a reasoning level and display the reasoning process in the chat interface.
- Implemented logic to stream reasoning content separately from the main response in the chat.

Author: ssbeatty

CLAUDE.md

@@ -413,7 +413,64 @@ def mermaid(diagram: str, *, title: str | None = None) -> None:
     _emit({"type": "artifact", "kind": "mermaid", "code": str(diagram), "title": title})
 
 
-def get_llm(name: str = "default"):
+_REASONING_BUDGET = {"low": 1024, "medium": 4096, "high": 12000}
+
+
+def _norm_reasoning(level) -> "str | None":
+    """Normalise a reasoning/think request to None | 'low' | 'medium' | 'high'.
+
+    Accepts a level string, a bool (True → 'medium'), or falsy/off values → None.
+    """
+    if level is None or level is False:
+        return None
+    if level is True:
+        return "medium"
+    s = str(level).strip().lower()
+    if s in ("", "off", "none", "no", "false", "0", "disabled"):
+        return None
+    if s in ("low", "medium", "high"):
+        return s
+    if s in ("min", "minimal"):
+        return "low"
+    if s in ("max", "maximum"):
+        return "high"
+    return "medium"
+
+
+def _apply_reasoning(extra: dict, provider: str, base_url: "str | None", level: str) -> None:
+    """Translate one shared low/medium/high level into the provider-specific
+    'think' knob (each vendor exposes reasoning differently):
+
+      - anthropic  → thinking={'type':'enabled','budget_tokens': …}
+                     (Claude also needs temperature=1 and max_tokens > budget)
+      - openai o-series / gpt-5 (no base_url) → reasoning_effort=<level>
+      - OpenAI-compatible gateway (base_url set) → extra_body={'enable_thinking': True}
+                     (Qwen3 / GLM / vLLM style — a boolean toggle, so the level only
+                     gates on/off here; tune per-gateway if it supports a budget)
+      - deepseek   → nothing (deepseek-reasoner reasons natively; the text comes
+                     back in additional_kwargs['reasoning_content'])
+      - ollama     → reasoning=True
+    """
+    if provider == "anthropic":
+        budget = _REASONING_BUDGET[level]
+        extra["thinking"] = {"type": "enabled", "budget_tokens": budget}
+        extra["temperature"] = 1
+        if not extra.get("max_tokens") or extra["max_tokens"] <= budget:
+            extra["max_tokens"] = budget + 4096
+    elif provider == "deepseek":
+        return
+    elif provider == "ollama":
+        extra.setdefault("reasoning", True)
+    else:  # openai + OpenAI-compatible gateways
+        if base_url:
+            eb = dict(extra.get("extra_body") or {})
+            eb.setdefault("enable_thinking", True)
+            extra["extra_body"] = eb
+        else:
+            extra.setdefault("reasoning_effort", level)
+
+
+def get_llm(name: str = "default", reasoning=None):
     """
     Return a LangChain chat model.
 
@@ -422,6 +479,13 @@ def get_llm(name: str = "default"):
                                   normalised to `_`). When several channels serve the
                                   same model id, the highest-priority channel wins
                                   (ties → earliest); credentials come from that channel.
+    - `reasoning=`             → turn on the model's thinking/chain-of-thought at a
+                                  shared level: `"low"` / `"medium"` / `"high"` (or
+                                  `True` = medium; `None`/`"off"` = disabled). Mapped
+                                  per provider (Claude thinking budget, OpenAI
+                                  reasoning_effort, gateway enable_thinking, …). The
+                                  reasoning text streams back as `reasoning_content`
+                                  or a `<think>` block — see the chat templates.
 
     Available model ids can be enumerated with `list_llms()`. Returns None outside
     the platform.
@@ -454,6 +518,10 @@ def get_llm(name: str = "default"):
         extra.setdefault("timeout", 60)
         extra.setdefault("max_retries", 1)
 
+        _level = _norm_reasoning(reasoning)
+        if _level:
+            _apply_reasoning(extra, provider, base_url, _level)
+
         if provider == "anthropic":
             from langchain_anthropic import ChatAnthropic
             return ChatAnthropic(model=model, api_key=api_key, **extra)
@@ -780,6 +848,7 @@ def get_agent(
     system_prompt: str | None = None,
     llm_name: str = "default",
     tools: list | None = None,
+    reasoning=None,
 ):
     """
     Return a ready-to-use ReAct agent (LangGraph create_react_agent).
@@ -805,7 +874,7 @@ async def run(input: dict) -> dict:
     import inspect
     from langgraph.prebuilt import create_react_agent
 
-    llm = get_llm(llm_name)
+    llm = get_llm(llm_name, reasoning=reasoning)
     if llm is None:
         raise RuntimeError(
             "No LLM configured. Add one in AgentFlow Settings before calling get_agent()."
@@ -865,6 +934,7 @@ def get_deep_agent(
     system_prompt: str | None = None,
     llm_name: str = "default",
     tools: list | None = None,
+    reasoning=None,
     **kwargs,
 ):
     """
@@ -899,7 +969,7 @@ async def run(input: dict) -> dict:
             "baseline venv; if it's missing, add 'deepagents' to requirements.txt."
         ) from e
 
-    llm = get_llm(llm_name)
+    llm = get_llm(llm_name, reasoning=reasoning)
     if llm is None:
         raise RuntimeError(
             "No LLM configured. Add one in AgentFlow Settings before calling get_deep_agent()."

backend/agentflow/__init__.py

@@ -0,0 +1,42 @@
+"""add conversations.reasoning_effort
+
+Adds a per-conversation reasoning/think level (off | low | medium | high),
+threaded into each run's input as input["reasoning"] and mapped to the model's
+provider-specific thinking parameter by agentflow.get_llm(reasoning=...).
+
+Plain add_column (a pure add needs no sqlite table-recreate) + inspector guard,
+so a DB that already has the column (e.g. a pre-Alembic DB healed via create_all,
+or one that got it through an interim path) is skipped rather than hitting a
+duplicate-column error. Same defensive shape as 0002/0003.
+"""
+from typing import Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = "0004"
+down_revision: Union[str, None] = "0003"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    insp = sa.inspect(bind)
+    cols = [c["name"] for c in insp.get_columns("conversations")]
+    if "reasoning_effort" not in cols:
+        op.add_column(
+            "conversations",
+            sa.Column(
+                "reasoning_effort",
+                sa.String(length=16),
+                nullable=False,
+                server_default="off",
+            ),
+        )
+
+
+def downgrade() -> None:
+    op.drop_column("conversations", "reasoning_effort")

backend/alembic/versions/0004_add_conversation_reasoning_effort.py

@@ -262,6 +262,10 @@ class Conversation(Base):
     script_id = Column(String, ForeignKey("scripts.id", ondelete="CASCADE"), nullable=False)
     title = Column(String(500), default="New conversation")
     context_turns = Column(Integer, default=10)
+    # Per-conversation reasoning/think level: off | low | medium | high. Threaded
+    # into each run's input as input["reasoning"] and mapped to the model's
+    # provider-specific thinking knob by agentflow.get_llm(reasoning=...).
+    reasoning_effort = Column(String(16), default="off", server_default="off")
     created_at = Column(DateTime, default=datetime.utcnow)
     updated_at = Column(DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)
 

backend/app/models.py

@@ -51,6 +51,7 @@ def create_conversation(body: ConversationCreate, db: Session = Depends(get_db))
         script_id=body.script_id,
         title=body.title,
         context_turns=body.context_turns,
+        reasoning_effort=body.reasoning_effort or "off",
     )
     db.add(conv)
     db.commit()
@@ -75,6 +76,8 @@ def update_conversation(conv_id: str, body: ConversationUpdate, db: Session = De
         conv.title = body.title
     if body.context_turns is not None:
         conv.context_turns = body.context_turns
+    if body.reasoning_effort is not None:
+        conv.reasoning_effort = body.reasoning_effort
     conv.updated_at = datetime.utcnow()
     db.commit()
     db.refresh(conv)
@@ -154,10 +157,15 @@ async def chat_start(conv_id: str, body: ConverseChatStartRequest, db: Session =
     history_slice = prior[-(conv.context_turns * 2):]
     history = [{"role": m.role, "content": m.content} for m in history_slice]
 
-    # Create execution row
+    # Create execution row. Thread the conversation's reasoning level into the
+    # input so the script can pass it to get_llm(reasoning=input.get("reasoning")).
     exc = Execution(
         script_id=conv.script_id,
-        input_data={"message": body.message, "history": history},
+        input_data={
+            "message": body.message,
+            "history": history,
+            "reasoning": conv.reasoning_effort or "off",
+        },
     )
     db.add(exc)
     db.commit()

backend/app/routers/conversations.py

@@ -524,18 +524,21 @@ class ConversationCreate(BaseModel):
     script_id: str
     title: str = "New conversation"
     context_turns: int = 10
+    reasoning_effort: str = "off"
 
 
 class ConversationUpdate(BaseModel):
     title: Optional[str] = None
     context_turns: Optional[int] = None
+    reasoning_effort: Optional[str] = None
 
 
 class ConversationSummary(BaseModel):
     id: str
     script_id: str
     title: str
     context_turns: int
+    reasoning_effort: str = "off"
     created_at: datetime
     updated_at: datetime
 

backend/app/schemas.py

@@ -17,6 +17,7 @@
 import subprocess
 import sys
 import threading
+import time
 from collections import deque
 from datetime import datetime
 from pathlib import Path
@@ -34,6 +35,19 @@
 MAX_CONCURRENT: int = int(os.getenv("AGENTFLOW_MAX_CONCURRENT", "5"))
 EXECUTION_TIMEOUT: float = float(os.getenv("AGENTFLOW_EXECUTION_TIMEOUT", "600"))
 
+# Lightweight timing diagnostics printed to the backend console (uvicorn / F5),
+# so a slow run can be split into queue-wait / prep / python-cold-start-imports /
+# script(LLM) without digging through the per-run logs in the DB. The heavy,
+# usually-dominant cost is `cold_import` — every run spawns a fresh python that
+# re-imports the whole langchain/langgraph stack (there is no warm worker pool).
+# Toggle off with AGENTFLOW_PROFILE=0.
+_PROFILE: bool = os.getenv("AGENTFLOW_PROFILE", "1").lower() not in ("0", "false", "no", "")
+
+
+def _prof(execution_id: str, msg: str) -> None:
+    if _PROFILE:
+        print(f"[agentflow] [{execution_id[:8]}] {msg}", flush=True)
+
 # lazy-init so it's created inside the running event loop
 _semaphore: asyncio.Semaphore | None = None
 
@@ -195,6 +209,10 @@ def _emit(d):
 async def _main():
 
     import agentflow as _af
+    # Signal that python cold-start + base imports are done, so the engine can
+    # split process-boot cost from script(LLM) cost. Unknown event type -> the
+    # engine's drain loop ignores it (never shown in the Logs panel).
+    _emit({{"type": "boot"}})
 
     # Zero-intrusion execution tracing: emits __AGENTFLOW__ trace events
     # for every LangGraph node, tool call, and agent action. User scripts
@@ -253,6 +271,8 @@ async def _run():
 async def start_execution(execution_id: str) -> None:
     db = SessionLocal()
     slot_acquired = False
+    _t_enter = time.perf_counter()
+    _t_slot = _t_spawn = _t_enter
     try:
         exc_row: Execution = db.query(Execution).filter_by(id=execution_id).first()
         if not exc_row:
@@ -269,6 +289,7 @@ async def start_execution(execution_id: str) -> None:
 
         await _get_semaphore().acquire()
         slot_acquired = True
+        _t_slot = time.perf_counter()
 
         # re-read: may have been cancelled while waiting in queue
         db.refresh(exc_row)
@@ -493,6 +514,12 @@ def _blob(model: str, ch) -> str:
             **popen_kwargs,
         )
         _procs[execution_id] = proc
+        _t_spawn = time.perf_counter()
+        _prof(execution_id, (
+            f"spawned pid={proc.pid} "
+            f"(queue_wait={_t_slot - _t_enter:.2f}s, prep={_t_spawn - _t_slot:.2f}s, "
+            f"venv={'yes' if venv_exists(exc_row.script_id) else 'no→backend-py'})"
+        ))
 
         def _pump(stream, is_stderr: bool):
             try:
@@ -510,14 +537,22 @@ def _pump(stream, is_stderr: bool):
         result_data: Any = None
         error_data: dict | None = None
         eof_count = 0
+        first_output_at: float | None = None
+        result_at: float | None = None
 
         async def _drain():
-            nonlocal result_data, error_data, eof_count
+            nonlocal result_data, error_data, eof_count, first_output_at, result_at
             while eof_count < 2:
                 is_stderr, line = await queue.get()
                 if line is None:
                     eof_count += 1
                     continue
+                if first_output_at is None:
+                    first_output_at = time.perf_counter()
+                    _prof(execution_id, (
+                        f"first output +{first_output_at - _t_spawn:.2f}s "
+                        f"(python cold-start + imports)"
+                    ))
                 if line.startswith(_PREFIX):
                     try:
                         payload = json.loads(line[len(_PREFIX):])
@@ -572,6 +607,7 @@ async def _drain():
                         await ws_manager.send(execution_id, payload)
                     elif t == "result":
                         result_data = payload.get("data")
+                        result_at = time.perf_counter()
                     elif t == "error":
                         error_data = payload
                         # Persist the crash as a log too, so it shows in the Logs
@@ -607,6 +643,11 @@ async def _drain():
             _procs.pop(execution_id, None)
 
             timeout_msg = f"Execution timed out after {EXECUTION_TIMEOUT:.0f}s"
+            _prof(execution_id, (
+                f"TIMEOUT after {EXECUTION_TIMEOUT:.0f}s | "
+                f"cold_import={'n/a' if first_output_at is None else f'{first_output_at - _t_spawn:.2f}s'} "
+                f"(no result before timeout)"
+            ))
             exc_row = db.query(Execution).filter_by(id=execution_id).first()
             exc_row.status = "failed"
             exc_row.error = timeout_msg
@@ -655,6 +696,19 @@ async def _drain():
                 })
         db.commit()
 
+        _t_end = time.perf_counter()
+        if first_output_at is not None:
+            _cold = first_output_at - _t_spawn
+            _script = (result_at or _t_end) - first_output_at
+        else:
+            _cold = _t_end - _t_spawn      # process produced no output at all
+            _script = 0.0
+        _prof(execution_id, (
+            f"done status={exc_row.status} rc={proc.returncode} | "
+            f"queue_wait={_t_slot - _t_enter:.2f}s prep={_t_spawn - _t_slot:.2f}s "
+            f"cold_import={_cold:.2f}s script={_script:.2f}s total={_t_end - _t_slot:.2f}s"
+        ))
+
         await ws_manager.send(execution_id, {
             "type": "status",
             "status": exc_row.status,