feat: Kiln assistant — auto mode, background eval jobs, and sub-agents (runtime, tabs UI, steer/stop)

.gitignore

@@ -42,3 +42,4 @@ reviews/
 # ActRight discover local state
 .act_right
 
+.playwright-mcp/

app/desktop/desktop_server.py

@@ -27,7 +27,10 @@
 from app.desktop.log_config import log_config
 from app.desktop.studio_server.agent_api import connect_agent_api
 from app.desktop.studio_server.batch_plan_api import connect_batch_plan_api
-from app.desktop.studio_server.chat import connect_chat_api
+from app.desktop.studio_server.chat import (
+    connect_chat_api,
+    connect_conversations_api,
+)
 from app.desktop.studio_server.code_tool_api import connect_code_tool_api
 from app.desktop.studio_server.copilot_api import connect_copilot_api
 from app.desktop.studio_server.data_gen_api import connect_data_gen_api
@@ -158,6 +161,7 @@ def make_app(tk_root: tk.Tk | None = None):
     connect_agent_api(app)
     connect_dev_tools(app)
     connect_chat_api(app)
+    connect_conversations_api(app)
     connect_jobs_api(app)
     # Important: webhost must be last, it handles all other URLs
     connect_webhost(app)

app/desktop/dev_env.py

@@ -14,3 +14,7 @@ def set_dev_env_vars() -> None:
     os.environ.setdefault("KILN_SKIP_REMOTE_MODEL_LIST", "true")
     os.environ["DEBUG_EVENT_LOOP"] = "true"
     os.environ["KILN_DEV_MODE"] = "true"
+    # Assistant forensic debug logging on by default in dev (unless explicitly set, e.g.
+    # KILN_CHAT_DEBUG_LOG=0) so the JSONL timeline already exists when an assistant issue
+    # needs investigating. See chat/debug_log.py.
+    os.environ.setdefault("KILN_CHAT_DEBUG_LOG", "1")

app/desktop/git_sync/test_background_sync.py

@@ -28,6 +28,26 @@ def second_clone(git_repos, tmp_path: Path):
     return second_path
 
 
+async def wait_for(predicate, timeout: float = 10.0, interval: float = 0.05) -> None:
+    """Poll until the (sync or async) predicate is truthy, or fail loudly.
+
+    Fixed sleeps flake on loaded CI runners — a real ``git fetch`` on an
+    executor thread can take longer than any constant chosen here. Polling
+    with a generous deadline keeps the tests fast locally and deterministic
+    under load.
+    """
+    deadline = asyncio.get_event_loop().time() + timeout
+    while True:
+        result = predicate()
+        if asyncio.iscoroutine(result):
+            result = await result
+        if result:
+            return
+        if asyncio.get_event_loop().time() > deadline:
+            pytest.fail(f"Condition not met within {timeout}s")
+        await asyncio.sleep(interval)
+
+
 @pytest.mark.asyncio
 async def test_poll_loop_fetches_and_fast_forwards(manager, git_repos, second_clone):
     local_path, _ = git_repos
@@ -40,7 +60,11 @@ async def test_poll_loop_fetches_and_fast_forwards(manager, git_repos, second_cl
     bg = BackgroundSync(manager, poll_interval=0.05, idle_pause_after=60.0)
     await bg.start()
     try:
-        await asyncio.sleep(0.3)
+
+        async def head_advanced() -> bool:
+            return await manager.get_head() != old_head
+
+        await wait_for(head_advanced)
     finally:
         await bg.stop()
 
@@ -72,18 +96,26 @@ async def test_idle_pause_and_resume(manager, git_repos, second_clone):
     bg = BackgroundSync(manager, poll_interval=0.05, idle_pause_after=0.1)
     await bg.start()
 
-    await asyncio.sleep(0.3)
+    try:
+        # Let the loop notice it has been idle for > idle_pause_after and
+        # park. (Parking has no observable flag; two 0.05s ticks past the
+        # 0.1s idle threshold suffice, so 0.5s carries a wide margin.)
+        await asyncio.sleep(0.5)
+
+        commit_in_repo(second_clone, "after_idle.txt", "data", "after idle")
+        push_from(second_clone)
 
-    commit_in_repo(second_clone, "after_idle.txt", "data", "after idle")
-    push_from(second_clone)
+        old_head = await manager.get_head()
+        assert not (local_path / "after_idle.txt").exists()
 
-    old_head = await manager.get_head()
-    assert not (local_path / "after_idle.txt").exists()
+        bg.notify_request()
 
-    bg.notify_request()
-    await asyncio.sleep(0.3)
+        async def head_advanced() -> bool:
+            return await manager.get_head() != old_head
 
-    await bg.stop()
+        await wait_for(head_advanced)
+    finally:
+        await bg.stop()
 
     new_head = await manager.get_head()
     assert new_head != old_head

app/desktop/studio_server/agent_api.py

@@ -551,40 +551,50 @@ async def agent_overview(
             Path(description="The unique identifier of the task within the project."),
         ],
     ) -> AgentOverview:
-        task = task_from_id(project_id, task_id)
-        project = task.parent_project()
-        if project is None:
-            raise HTTPException(status_code=404, detail="Project not found")
-
-        task_run_configs = get_all_run_configs(project_id, task_id)
-
-        instruction_text = truncate_to_words_with_agent_sentinel(task.instruction, 70)
-
-        return AgentOverview(
-            project=AgentOverviewProject(
-                id=project.id,
-                name=project.name,
-                description=project.description,
-            ),
-            task=AgentOverviewTask(
-                id=task.id,
-                name=task.name,
-                description=task.description,
-                instruction=instruction_text or "",
-                input_json_schema=task.input_json_schema,
-                output_json_schema=task.output_json_schema,
-                default_run_config_id=task.default_run_config_id,
-            ),
-            dataset=_dataset_stats(task),
-            docs=_docs_stats(project),
-            search_tools=_search_tools_block(project),
-            prompts=_prompts_block(task, project, task_run_configs),
-            specs=_specs_block(task),
-            evals=_evals_block(task),
-            tool_servers=_tool_servers_block(project),
-            run_configs=_run_configs_block(task, task_run_configs),
-            fine_tunes=_fine_tunes_block(task),
-            prompt_optimization_jobs=_prompt_optimization_jobs_block(task),
-            skills=_skills_block(project),
-            connected_providers=await _connected_providers_block(),
-        )
+        connected_providers = await _connected_providers_block()
+
+        # The block builders below scan child directories on disk (runs, specs,
+        # evals, fine-tunes, ...) — far too slow for the event loop on large
+        # datasets, so the whole assembly runs in a worker thread.
+        def build_overview() -> AgentOverview:
+            task = task_from_id(project_id, task_id)
+            project = task.parent_project()
+            if project is None:
+                raise HTTPException(status_code=404, detail="Project not found")
+
+            task_run_configs = get_all_run_configs(project_id, task_id)
+
+            instruction_text = truncate_to_words_with_agent_sentinel(
+                task.instruction, 70
+            )
+
+            return AgentOverview(
+                project=AgentOverviewProject(
+                    id=project.id,
+                    name=project.name,
+                    description=project.description,
+                ),
+                task=AgentOverviewTask(
+                    id=task.id,
+                    name=task.name,
+                    description=task.description,
+                    instruction=instruction_text or "",
+                    input_json_schema=task.input_json_schema,
+                    output_json_schema=task.output_json_schema,
+                    default_run_config_id=task.default_run_config_id,
+                ),
+                dataset=_dataset_stats(task),
+                docs=_docs_stats(project),
+                search_tools=_search_tools_block(project),
+                prompts=_prompts_block(task, project, task_run_configs),
+                specs=_specs_block(task),
+                evals=_evals_block(task),
+                tool_servers=_tool_servers_block(project),
+                run_configs=_run_configs_block(task, task_run_configs),
+                fine_tunes=_fine_tunes_block(task),
+                prompt_optimization_jobs=_prompt_optimization_jobs_block(task),
+                skills=_skills_block(project),
+                connected_providers=connected_providers,
+            )
+
+        return await asyncio.to_thread(build_overview)

app/desktop/studio_server/chat/__init__.py

@@ -1,7 +1,7 @@
-from app.desktop.studio_server.chat.routes import ExecuteToolsRequest, connect_chat_api
+from app.desktop.studio_server.chat.routes import connect_chat_api
+from app.desktop.studio_server.chat.runtime.api import connect_conversations_api
 from app.desktop.studio_server.chat.sse_parser import EventParser
 from app.desktop.studio_server.chat.stream_session import (
-    ChatStreamSession,
     RoundState,
     ToolCallInfo,
     execute_tool,
@@ -14,13 +14,12 @@
 )
 
 __all__ = [
-    "ChatStreamSession",
     "EventParser",
-    "ExecuteToolsRequest",
     "KilnToolInputMetadata",
     "RoundState",
     "ToolCallInfo",
     "connect_chat_api",
+    "connect_conversations_api",
     "execute_tool",
     "execute_tool_batch",
     "tool_input_executor_is_server",

app/desktop/studio_server/chat/constants.py

@@ -6,11 +6,22 @@
 from pydantic import TypeAdapter
 
 CHAT_TIMEOUT = httpx.Timeout(timeout=300.0, connect=30.0)
-MAX_TOOL_ROUNDS = 100
+MAX_TOOL_ROUNDS = 1000
 
 SSE_TYPE_TOOL_CALLS_PENDING = "tool-calls-pending"
 SSE_TYPE_TOOL_EXEC_START = "kiln-tool-execution-start"
 SSE_TYPE_TOOL_EXEC_END = "kiln-tool-execution-end"
+# Emitted by the OLD interactive loop when the model calls enable_auto_mode
+# (survives until phase 4 ports the interactive path; the runtime interceptor
+# emits the same event via stream_session's formatter). The rest of the old
+# auto lifecycle vocabulary (auto-mode-on/off/idle/state) died in phase 3,
+# replaced by the unified conversation-state event (runtime/sse.py).
+SSE_TYPE_AUTO_MODE_CONSENT_REQUIRED = "auto-mode-consent-required"
+# Emitted between retry attempts after a transient upstream failure (by BOTH the
+# interactive chat stream and the runtime engine — they share the retry helper),
+# so the UI can show "retrying N/M…" instead of a hard error. Carries
+# {attempt, max_attempts, status_code?, run_id?}.
+SSE_TYPE_CHAT_RETRY = "kiln-chat-retry"
 
 DENIED_TOOL_OUTPUT = json.dumps(
     {"error": "The user did not accept the toolcall"}, ensure_ascii=False

app/desktop/studio_server/chat/debug_log.py

@@ -0,0 +1,71 @@
+"""Forensic debug logging for the assistant runtime (desktop side).
+
+Enabled with the ``KILN_CHAT_DEBUG_LOG`` environment variable: ``1``/``true``
+logs to ``<settings>/logs/kiln_chat_debug.jsonl``; any other non-empty value
+is used as the target file path. Each event is one JSON line carrying ids and
+timings — never message content — keyed by the conversation ``session_id``
+(``cv_…``). The desktop sends that same id upstream on every request as
+``X-Kiln-Conversation-Id``, and kiln_server's ``chat_debug_log_enabled`` flag
+logs it on its side, so the two timelines can be joined per conversation when
+investigating a stuck or misbehaving assistant conversation.
+
+This is a local debugging tool: writes are synchronous appends under a
+process-wide lock and the file grows unbounded while the flag is set.
+"""
+
+import json
+import os
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from kiln_ai.utils.config import Config
+
+ENV_VAR = "KILN_CHAT_DEBUG_LOG"
+_FALSY = frozenset(["", "0", "false", "no", "off"])
+_TRUTHY_DEFAULT_PATH = frozenset(["1", "true", "yes", "on"])
+
+_write_lock = threading.Lock()
+_process_started = time.monotonic()
+
+
+def chat_debug_enabled() -> bool:
+    return os.getenv(ENV_VAR, "").strip().lower() not in _FALSY
+
+
+def _target_path() -> Path:
+    value = os.getenv(ENV_VAR, "").strip()
+    if value.lower() in _TRUTHY_DEFAULT_PATH:
+        return Path(Config.settings_dir()) / "logs" / "kiln_chat_debug.jsonl"
+    return Path(value)
+
+
+def chat_debug_log(
+    event: str, conversation_id: str | None = None, **fields: Any
+) -> None:
+    """Append one JSONL debug event; a no-op unless the env flag is set.
+
+    ``elapsed_ms`` is process-relative (a shared monotonic clock) so gaps
+    between events are readable without timestamp arithmetic.
+    """
+    if not chat_debug_enabled():
+        return
+    record: dict[str, Any] = {
+        "ts": datetime.now(timezone.utc).isoformat(timespec="milliseconds"),
+        "elapsed_ms": round((time.monotonic() - _process_started) * 1000, 1),
+        "event": event,
+        "conversation_id": conversation_id,
+        **fields,
+    }
+    line = json.dumps(record, default=str, ensure_ascii=False)
+    # Debug-only tool: a failed write must never affect the runtime.
+    try:
+        path = _target_path()
+        with _write_lock:
+            path.parent.mkdir(parents=True, exist_ok=True)
+            with open(path, "a", encoding="utf-8") as f:
+                f.write(line + "\n")
+    except OSError:
+        pass

app/desktop/studio_server/chat/helpers.py

@@ -1,8 +1,13 @@
-import json
-from unittest.mock import AsyncMock, MagicMock
+"""Tiny shared test fixtures for the chat package.
+
+Phase 4 note: this module used to also carry the httpx mock builders
+(``make_httpx_mock`` / ``make_n_round_mock_client`` and the PATCH_* targets)
+that drove the deleted ``POST /api/chat`` route tests; the surviving suites
+script upstream rounds through ``chat/test_fakes.py`` instead, so only the
+SSE payload builder remains.
+"""
 
-PATCH_ASYNC_CLIENT = "app.desktop.studio_server.chat.stream_session.httpx.AsyncClient"
-PATCH_EXECUTE_TOOL = "app.desktop.studio_server.chat.stream_session.execute_tool"
+import json
 
 
 def sse_text_delta(delta: str, text_id: str = "text-test") -> bytes:
@@ -12,62 +17,3 @@ def sse_text_delta(delta: str, text_id: str = "text-test") -> bytes:
         "delta": delta,
     }
     return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode()
-
-
-def make_httpx_mock(status_code: int = 200, chunks: list[bytes] | None = None):
-    if chunks is None:
-        chunks = [sse_text_delta("hello")]
-
-    async def mock_aiter_bytes():
-        for chunk in chunks:
-            yield chunk
-
-    mock_upstream = MagicMock()
-    mock_upstream.status_code = status_code
-    mock_upstream.aiter_bytes.return_value = mock_aiter_bytes()
-    mock_upstream.aread = AsyncMock(
-        return_value=b'{"message":"upstream error"}' if status_code != 200 else b""
-    )
-    mock_upstream.__aenter__ = AsyncMock(return_value=mock_upstream)
-    mock_upstream.__aexit__ = AsyncMock(return_value=None)
-
-    mock_client = MagicMock()
-    mock_client.stream.return_value = mock_upstream
-    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
-    mock_client.__aexit__ = AsyncMock(return_value=None)
-
-    mock_async_client_class = MagicMock(return_value=mock_client)
-    return mock_async_client_class, mock_client, mock_upstream
-
-
-def make_stream_mock(chunks: list[bytes]):
-    """Create a mock upstream response that yields the given SSE chunks."""
-
-    async def mock_aiter_bytes():
-        for chunk in chunks:
-            yield chunk
-
-    mock_upstream = MagicMock()
-    mock_upstream.status_code = 200
-    mock_upstream.aiter_bytes.return_value = mock_aiter_bytes()
-    mock_upstream.__aenter__ = AsyncMock(return_value=mock_upstream)
-    mock_upstream.__aexit__ = AsyncMock(return_value=None)
-    return mock_upstream
-
-
-def make_n_round_mock_client(*chunk_rounds: list[bytes]):
-    """Create a mock httpx client that serves multiple streaming rounds in sequence."""
-    mocks = [make_stream_mock(chunks) for chunks in chunk_rounds]
-    call_count = 0
-
-    def side_effect(*args, **kwargs):
-        nonlocal call_count
-        idx = min(call_count, len(mocks) - 1)
-        call_count += 1
-        return mocks[idx]
-
-    mock_client = MagicMock()
-    mock_client.stream.side_effect = side_effect
-    mock_client.__aenter__ = AsyncMock(return_value=mock_client)
-    mock_client.__aexit__ = AsyncMock(return_value=None)
-    return mock_client, lambda: call_count