Kiln Assistant: unified conversation runtime#1558
Merged
leonardmq merged 12 commits intoJul 10, 2026
Merged
Conversation
…protocol harness Phase 1 of the assistant unified runtime: a new, self-contained app/desktop/studio_server/chat/runtime/ package that will replace the three parallel chat loops (ChatStreamSession, AutoChatRunner, SubAgentRunner) and their two registries. Nothing is wired into the running app yet — the old code paths remain untouched and authoritative. - models.py: RunState + ConversationRecord (one lifecycle enum + auto_flag axis replacing AutoRunStatus/SubAgentStatus), frozen ConversationPolicy with per-kind factories, InboundMessage/SubAgentSeed/PendingApprovalBatch, byte-identical kickoff/seed-body/report-frame builders. - bus.py: the ONE ByteEventBus + current-turn replay buffer (trace-boundary reset, on-subscribe state marker, terminal EOF, close-on-evict), generalized from the auto and sub-agent buses. - engine.py: ConversationEngine — THE round loop, reusing stream_session.py's round primitives (iter_round_with_retries, execute_tool_batch, continuation builder) with policy-driven approval gating, framing, one-shot semantics, and graceful-stop handling. - interceptors.py: signal-tool chain (enable/disable auto mode, depth guard, child noops) with the old scan-priority order preserved. - supervisor.py: ConversationSupervisor — single session-id registry with one settle path (run-once, cancel-before-first-run backstop), caps, report delivery/queueing, wait/stop/cascades, and GC (terminal TTL + undelivered-report pinning, OFF-auto TTL, interactive LRU). - sse.py: unified conversation-state control event + canonical copies of the generic per-run formatters. Golden-protocol harness: golden_scenarios.py + checked-in fixtures under runtime/golden/ pin the exact upstream request-body sequences the OLD loops produce for seven scripted scenarios (tool round, approval flow, report injection, auto seed + side-note, disable resolve, sub-agent seed + steer, unwrapped report inbox); tests assert old loop == fixture == new engine, so the fixtures remain the behavior contract after later phases delete the old loops. Plus 90 unit tests across bus/interceptors/engine/supervisor (modeled on the existing auto/sub-agent suites), including drift guards pinning the runtime's canonical constant copies to the doomed originals. Also fixes a pre-existing ruff-format drift in subagents/test_runner.py.
Sub-agents now run on the phase-1 ConversationEngine/ConversationSupervisor instead of the dedicated SubAgentRunner/SubAgentRegistry, which are deleted along with the rest of chat/subagents/ (events, sse, models, api + tests). Backend: - runtime/supervisor.py: live conversation_supervisor singleton; registry- level conversation-state firehose (BroadcastBus) fed by one _publish_state dual-publish helper; phase-2-only legacy_report_deliverer hook so a settled child's report still injects into the OLD auto registry's inbox for auto:* parents (the runtime never imports chat/auto/). - chat/orchestration.py: the orchestration tool executor relocated (survives per architecture §1) and retargeted at the supervisor. Spawn caps as tool-result errors, statuses-only wait/status results + note, ownership scoping, stop outcome vocabulary, and consent marking are preserved verbatim. Hosts the phase-2 parent-identity bridge (ParentConversationIndex: trace→parent-key alias chain + consent set; children store the parent_key as parent_session_id) plus the report drains, delete cascade, and the legacy auto-parent report bridge — all annotated for phase-3/4 removal. - Integration points retargeted from the deleted registry: stream_session.py (orchestration ctx, spawn-consent downgrade, note_parent_trace chaining, pending-report drain + echoes), routes.py (sessions join via the supervisor trace index — wire names kept, values are session ids; delete cascade; next-turn report injection), auto/registry.py child-stop cascade, auto/runner.py ctx + aliasing. - New /api/conversations surface (runtime/api.py, registered in desktop_server.py) replacing /api/chat/subagents/*: list?parent=, events firehose, get (+include_report), per-conversation observer SSE emitting the unified conversation-state event instead of kiln-subagent-status, stop, and messages — old statuses (202/404/409) preserved. Frontend: - New src/lib/chat/conversation_store.ts (architecture §7) absorbing subagent_store.ts (deleted): /api/conversations endpoints + conversation-state vocabulary; observe() re-fetches the item for a fresh hydration leaf since state events no longer carry trace ids. - subagent_tabs/transcript/status-dot + chat.svelte re-pointed with zero visual/UX change; api_schema.d.ts regenerated. Tests: - Old executor/API/registry/runner suites ported (chat/test_orchestration.py, runtime/test_api.py, supervisor/engine additions incl. the legacy report bridge, session-deleted cascade, framing matrix, steer drain-before-finish); conversation_store.test.ts ports the store suite + kind-filter and fresh-leaf re-fetch tests. - Golden harness: subagent_seed_and_steer's old-loop side deleted with the runner (test skips with a comment); its fixture remains the durable contract the engine keeps matching; regeneration falls back to the engine for old-less scenarios. - Drift guards for strings whose old home died become byte-pins. uv run ./checks.sh --agent-mode exits 0.
Auto conversations now run on the ConversationSupervisor/Engine and
chat/auto/ (registry, runner, api, events, sse, models + tests) is
deleted. Interactive chat stays on the old loop until phase 4; the
consent accept/decline endpoints bridge into the runtime.
Supervisor: enable_auto (create-or-flip on the same record; ARMED-only
manual enable with no empty upstream POST; R2 no-trace seed; cap 429s;
trace adoption), disable_auto/disable_auto_for_trace (pre-marked
user_disabled, burst cancel, off publish, TTL GC, stop_children — the
phase-1 TODO(phase-3) interactive-disable cascade, wired), set_auto_flag,
resolve_auto_for_trace/auto_record_for_trace. Seed body ported verbatim
as models.build_auto_seed_body and pinned by the golden fixture. The
flag-off GC branch keys on the settled run's POLICY (not record.kind) so
flips can't TTL-GC a live interactive record; send_message refuses
flag-off auto records (old 404 semantics + consent safety). Queued
undelivered reports pin the OFF-auto parent past the terminal TTL so the
resumed interactive turn can still drain them.
Endpoints re-homed under /api/conversations: POST / (enable),
POST /auto/decline (interactive continuation), POST /{sid}/auto (flag
flips), GET /resolve (hard-refresh resync via the whole-chain trace
index); stop/messages/events already unified. The sessions-list join
reads auto state from the supervisor (auto_run_id now carries the
session id). Trace-handle resolution for supervisor parents falls back
to the supervisor trace index in both _resolve_parent_key (auto-parent
children visible via ?parent=<leaf>) and pending_reports_for_trace
(reports from children surviving an in-burst disable_auto_mode drain on
the next interactive turn), kind-guarded so a child's leaf is never a
parent handle.
OrchestrationContext carries parent_session_id (parent_auto_run_id,
auto:* alias keys, and the phase-2 legacy_report_deliverer bridge are
gone — auto parents' reports route natively through the supervisor
inbox). ParentConversationIndex survives only for old-loop interactive
parents (phase-4 removal).
Frontend: auto_run_store.ts folded into conversation_store.ts as
auto_conversation_store; one documented mapping translates the unified
conversation-state event back onto the exact old
auto-mode-on/off/idle/state UX (green dot, waiting-for-you, consent and
stop dialogs) with zero visual change. The attach-time on-subscribe
state marker updates flag/working without signaling the idle sink
(matching the old auto-mode-state marker). chat_session_store,
chat.svelte, chat_history.svelte re-pointed; api_schema.d.ts
regenerated.
Tests: auto endpoint/registry/runner suites ported to
supervisor/engine/api level (consent→burst, armed-only enable, flips,
disable cascade, resolve, off-auto refusal, GC pinning, sessions join,
interactive interception via /api/chat); golden auto scenarios skip
their deleted old-loop side (fixtures remain the durable contract, now
driving the ported seed builder); interceptor drift guards converted to
byte-pins; test_fakes.py relocated to chat/; full auto_run_store test
port in conversation_store.test.ts plus attach-marker and kind-guard
tests.
Interactive conversations — the most-used path — now run as supervised
turn tasks on chat/runtime/ (IDLE → RUNNING → IDLE), and the entire old
interactive surface is deleted: POST /api/chat, /api/chat/execute-tools,
ChatStreamSession (the loop class; the shared round primitives stay), the
/api/conversations/auto/decline bridge, and ParentConversationIndex (the
last old-world identity bridge). routes.py keeps only the history proxies
and the version policy.
Backend:
- POST /api/conversations grows a kind selector: kind="interactive" is
create-or-adopt keyed by the (possibly stale) leaf trace id, idempotent
via the whole-chain trace index (supervisor.adopt_interactive).
- POST /{sid}/messages on IDLE starts a turn (body byte-identical to the
old POST /api/chat, incl. the next-turn sub-agent report injection
re-homed from routes.post_chat) and returns {message_id} so the sending
tab can dedupe its own echo; RUNNING/AWAITING_APPROVAL sends queue.
- Approvals are parked batches: GET /{sid}/approvals +
POST /{sid}/approvals/decisions (batch id validated; first decision set
wins, second tab gets 409). Runless batches (recovery) resume via
engine.run(resume_batch=...) producing exactly the old /execute-tools
continuation body.
- Restart/refresh recovery: pending approvals rehydrate from the persisted
trace tail (unanswered calls only, conservatively gated; signal siblings
resolve as declined via PendingApprovalBatch.preresolved_results so the
trace keeps no dangling call); the pending event stays in the replay
buffer while parked.
- The auto flip is the true "same run, new policy" flip: enable swaps an
interactive record's policy AND kind to auto on the SAME record; every
flag-off settle swaps it back — the OFF-auto TTL GC is replaced by the
idle-interactive LRU pool (with the undelivered-report/live-children
pinning re-homed), and the flag-off send refusal is lifted structurally.
- Consent decline folds into POST /{sid}/auto (enabled=false + decline
ctx), byte-identical continuation, streaming on the observer. The engine
interactive-disable interception wires the full old cascade via
EngineIO.on_auto_flag_cleared. Interactive stop() cancels the turn
without cascading to children (session deletion still cascades,
explicitly). OrchestrationContext shrinks to {parent_session_id, depth}.
Frontend:
- streamChat/resumePendingToolCalls die; the auto store generalizes into
main_conversation_store — ONE observer for both kinds (off transitions
no longer close the stream; new onInteractiveIdle/onAwaitingApproval/
onConsentRequired/onVersionNudge sink hooks; ensure/fetchApprovals/
decide/decline operations; observer errors mid-turn surface like the old
fetch-error path and the not-working signal resets a stuck composer
without flushing the queue).
- chat_session_store.ts shrinks to UI-session state; sessionStorage keeps
only {sessionId, traceId, ui prefs} — transcripts always rebuild from
hydrate+observe. Approval box keys off AWAITING_APPROVAL + the fetched
batch; queued-message UX, consent dialog, retry indicator, context
gauge, version banners and welcome flows unchanged. api_schema.d.ts
regenerated.
Tests: golden interactive scenarios run_old=None (fixtures remain the
durable contract; interactive_approval_flow now also pins a DENIED item);
old route/loop suites ported to the engine/supervisor/API surfaces incl.
rehydration, runless-batch resume, flip, decline and recovery coverage;
paid integration tests re-driven through the supervisor; frontend store
suites fully ported (refresh-convergence and observer-drop regressions
included). checks.sh green.
trace_id leaves the desktop↔browser surface (functional spec §4: browser
code never sees trace_id) — the browser now holds one opaque conversation
key (live cv_ session id / durable upstream root_id / legacy leaf) and the
desktop owns the key→leaf mapping while the upstream API stays trace-keyed
until phase 6.
Desktop:
- routes.py: resolve_conversation_key (live record → root→leaf via a
bounded, snapshot-shape-gated upstream-list scan → legacy leaf); an
INDETERMINATE scan (transport failure / upstream error / bound exhausted)
raises UpstreamResolutionError and fails loudly with 503 — never the
silent root-as-leaf fall-through. Sessions-list rows keyed live-sid →
root_id → leaf; GET/DELETE accept any key and resolve the CURRENT leaf;
the snapshot proxy passes root_id through.
- ConversationRecord.root_id: the durable upstream session id, stamped by
the engine PRE-EMIT on a fresh record's first kiln_chat_trace (so the
browser's item GET triggered by that byte always sees it), by
adopt-by-root, and backfilled from the rehydration fetch.
- runtime/api.py: ConversationItem drops current_trace_id, gains root_id;
POST /api/conversations re-keyed trace_id→session_id (interactive
create/adopt by key incl. terminal-record and dead-cv_ shapes; auto
enable keyed by the live sid, seeding from the record's own leaf —
unknown sid 404s instead of forking); GET /api/conversations/resolve,
supervisor.resolve_auto_for_trace, and _resolve_parent_key deleted; the
consent control event drops trace_id.
- The trace index stays INTERNAL (sessions-list join; TODO(phase-6) notes
where the desktop-side root→leaf resolution collapses).
Frontend:
- sessionStorage persists {sessionId, rootId} — rootId replaces the leaf
traceId as the desktop-restart recovery key, learned from history rows,
hydration snapshots, the resync item GET, and a one-shot item fetch on
the first persisted turn (generation+sid stale-write guards so a
conversation switch racing the fetch can never stamp the old root).
- traceIdForNextChatRequest and hydration trace stamping deleted; ensure/
loadSession/resync/consent/armed/manual-enable all key on session ids;
children sync keys on main_conversation_store.sessionId; child
transcripts hydrate by session id (the desktop resolves the fresh leaf);
the sink's onChatTrace(traceId) became onTurnPersisted().
- Zero visual/UX change; api_schema.d.ts regenerated.
Tests: 241 backend chat tests (+16 skipped paid) and 1314 frontend tests
green; new coverage for row keying, key resolution (incl. indeterminate-
scan 503s and paging bounds), root stamping/backfill/races, sid-keyed
enable and /resolve absence pins. uv run ./checks.sh --agent-mode exit 0.
…an (phase 6)
The backend now resolves either id kind (leaf trace id or durable root/session
id) itself — GET/DELETE /v1/chat/sessions/{id} and session_id continuation on
POST /v1/chat/ — so the desktop's phase-5 key→leaf machinery is deleted and
resume-after-restart needs no leaf bookkeeping anywhere.
- routes.py: delete _leaf_for_root_id, the scan bounds, and
UpstreamResolutionError (+ both 503 branches — the indeterminate-resolution
failure mode lives upstream now and passes through like any other error).
resolve_conversation_key is synchronous and purely local: live records
resolve to their freshest upstream identity (current leaf → root → adopted
resume key); dead cv_ handles 404; anything else forwards VERBATIM. The
sessions-list join gains a root-id fallback so a record adopted by root key
(no leaf until its first persist) still joins its row.
- runtime: ConversationRecord.resume_session_key holds the opaque adopted key
(root or legacy leaf — indistinguishable desktop-side, and deliberately
never stamped into root_id, which must stay the TRUE durable recovery key).
adopt_interactive(session_key) adopts cold keys verbatim, indexes them in
the whole-chain trace index (idempotent re-adopt + list join) and seeds
seen_trace_ids so the engine's root-stamp suppression holds unchanged.
continuation_key_fields gives every idle-start body one precedence:
trace_id for the normal in-process flow (kept deliberately — the engine
holds the fresh leaf; switching would only add a backend resolution per
turn), session_id for resume-by-key first turns and keyless rehydrated
approval batches. Approval rehydration fetches by leaf-or-resume-key and
keeps backfilling the true root_id from the response.
- engine/stream_session: continuation rebuilds drop session_id once a real
trace id exists (the backend 400s the two keys together — same first-POST
lifecycle as the agent block), and _build_openai_tool_continuation treats a
session_id base as trace-only so resumed batch continuations stay
role:tool-results-only.
- orchestration.handle_session_deleted accepts any conversation key (live sid
or indexed trace/root key) since the delete proxy no longer resolves a leaf.
- Generated SDK deliberately NOT regenerated: the only wire change is a
request-body field on an endpoint driven via raw httpx dict bodies, and the
session endpoints take the id as a path param the SDK already passes
verbatim.
Tests: scan-machinery suites replaced with verbatim-forwarding/live-key
resolution tests; adopt-by-key, session_id-first-turn → trace_id-after-persist
(API-level end-to-end and engine-level), rehydrated-batch session_id bases,
and the list-join root fallback. Golden protocol fixtures untouched — the
normal trace_id turn protocol is byte-identical.
Both are intermittent races reproduced live and root-caused: BUG 1 (server inbox stranding): a send_message POST landing in the window between the engine's last empty drain_inbox() and the state flip to IDLE appended to conv.inbox but was never consumed — the turn settled and the message stranded, leaving the browser stuck 'thinking' (echo rendered, no reply) until a refresh. Fix: _finish_run restarts a fresh turn from a non-empty inbox on a NATURAL settle only (ended_naturally gate excludes cancel/exception so stop/disable can't resurrect a run or double-settle); _restart_from_inbox mirrors send_message's idle re-arm exactly (unframed messages, report injection, continuation_key_fields, auto_mode), preserving echo-once and persisted-trace fidelity. BUG 2 (client missed-live-idle flush): conversation-state idle events are live-only (never buffered), so if the observer is momentarily detached at the settle instant the browser gets only the on-subscribe idle MARKER, whose handler suppressed the settle hooks — onInteractiveIdle/maybeFlush never fired and a client-held queuedMessage stranded (chip stayed visible). Fix: a new flush-only onIdleMarker sink hook fires on a flag-off idle marker and calls maybeFlush; guarded so it no-ops in the refresh-brick sequence and can't double-send (dispatchQueued clears before send). The two fixes act on disjoint message sets (client-held vs already-POSTed), so no overlap.
Contributor
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (3)
WalkthroughThis PR replaces separate chat auto and subagent runtimes with a unified conversation runtime, supervisor, API, event model, orchestration layer, and updated tests and protocol fixtures. ChangesUnified Conversation Runtime Migration
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Browser
participant ConversationsAPI
participant ConversationSupervisor
participant ConversationEngine
participant ByteEventBus
Browser->>ConversationsAPI: send conversation message
ConversationsAPI->>ConversationSupervisor: enqueue message
ConversationSupervisor->>ConversationEngine: start run
ConversationEngine->>ByteEventBus: emit SSE payloads
ByteEventBus-->>Browser: replay/live conversation events
ConversationEngine-->>ConversationSupervisor: settle run
ConversationSupervisor->>ByteEventBus: publish conversation state
Possibly related PRs
Suggested reviewers: Poem
🚥 Pre-merge checks | ✅ 3 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
📊 Coverage ReportOverall Coverage: 92% Diff: origin/leonard/assistant-subagents...HEAD
Summary
Line-by-lineView line-by-line diff coverageapp/desktop/studio_server/chat/orchestration.pyLines 105-115 105 "state": record.state.value,
106 "rounds": record.rounds_used,
107 }
108 if record.current_leaf_trace_id is not None:
! 109 payload["session_id"] = record.current_leaf_trace_id
110 if include_report and record.state.is_terminal and record.final_report:
! 111 payload["report"] = record.final_report
112 return payload
113
114
115 async def execute_orchestration_tool(Lines 136-147 136 if tool_name == WAIT_FOR_SUBAGENTS_TOOL_NAME:
137 return await _wait(args, parent_key)
138 if tool_name == STOP_SUBAGENT_TOOL_NAME:
139 return await _stop(args, parent_key)
! 140 except Exception:
! 141 logger.exception("Orchestration tool %s failed", tool_name)
! 142 return _error("Internal error executing the sub-agent operation.")
! 143 return _error(f"Unknown orchestration tool: {tool_name}")
144
145
146 async def _spawn(args: dict[str, Any], parent_key: str) -> str:
147 agent_type = args.get("agent_type")Lines 224-233 224 if subagent_id:
225 record = _owned_record(str(subagent_id), parent_key)
226 if record is None:
227 return _error(f"Unknown sub-agent id: {subagent_id}")
! 228 payload = _record_payload(record, include_report)
! 229 return json.dumps(
230 {"subagents": [payload], "note": _REPORTS_AS_MESSAGES_NOTE},
231 ensure_ascii=False,
232 )Lines 241-249 241
242 async def _wait(args: dict[str, Any], parent_key: str) -> str:
243 raw_ids = args.get("subagent_ids")
244 if not isinstance(raw_ids, list) or not raw_ids:
! 245 return _error("wait_for_subagents requires a non-empty subagent_ids list.")
246 ids = [str(sid) for sid in raw_ids]
247 unknown = [sid for sid in ids if _owned_record(sid, parent_key) is None]
248 if unknown:
249 return _error(f"Unknown sub-agent ids: {', '.join(unknown)}")Lines 250-261 250
251 timeout = args.get("timeout_seconds")
252 try:
253 timeout_seconds = float(timeout) if timeout is not None else None
! 254 except (TypeError, ValueError):
! 255 timeout_seconds = None
256 if timeout_seconds is None:
! 257 timeout_seconds = WAIT_DEFAULT_TIMEOUT_SECONDS
258 timeout_seconds = min(max(timeout_seconds, 1.0), WAIT_MAX_TIMEOUT_SECONDS)
259
260 records, timed_out = await conversation_supervisor.wait(ids, timeout_seconds)
261 return json.dumps(Lines 274-282 274
275 async def _stop(args: dict[str, Any], parent_key: str) -> str:
276 subagent_id = args.get("subagent_id")
277 if not isinstance(subagent_id, str) or not subagent_id:
! 278 return _error("stop_subagent requires a subagent_id.")
279 record = _owned_record(subagent_id, parent_key)
280 if record is None:
281 return json.dumps({"status": "not_found"}, ensure_ascii=False)
282 # The supervisor's stop() is idempotent-void; the outcome vocabulary theLines 303-311 303 left to consume their reports) and stop the conversation itself — a
304 deleted CHILD session stops that child (old behavior), and a deleted
305 parent's in-flight run is cancelled."""
306 if conversation_supervisor.get(conversation_key) is not None:
! 307 session_id: str | None = conversation_key
308 else:
309 session_id = conversation_supervisor.session_for_trace(conversation_key)
310 if session_id is None:
311 returnapp/desktop/studio_server/chat/routes.pyLines 468-476 468 leaf server-side, so the desktop no longer needs the leaf to delete a
469 root-keyed session)."""
470 resolved = resolve_conversation_key(session_id)
471 if resolved.upstream_key is None:
! 472 raise HTTPException(
473 status_code=404, detail=f"Chat session not found: {session_id}"
474 )
475 api_key = get_copilot_api_key()
476 client = get_authenticated_client(api_key)app/desktop/studio_server/chat/runtime/api.pyLines 277-285 277 async for item in iter_with_keepalive(
278 conversation_supervisor.subscribe(session_id), KEEPALIVE_SECONDS
279 ):
280 if isinstance(item, KeepalivePing):
! 281 yield b": ping\n\n"
282 else:
283 yield item
284 Lines 310-318 310 yield payload
311
312 async for item in iter_with_keepalive(_with_snapshot(), KEEPALIVE_SECONDS):
313 if isinstance(item, KeepalivePing):
! 314 yield b": ping\n\n"
315 else:
316 yield item
317 Lines 434-450 434 )
435 except ValueError as exc:
436 # Sub-agent/terminal records can't enable auto mode (mirrors
437 # set_auto_flag's "invalid" outcome).
! 438 raise HTTPException(status_code=409, detail=str(exc))
439 except ConversationCapError as exc:
440 # Same cap surface as today: HTTP 429 with the preserved message.
441 raise HTTPException(status_code=429, detail=str(exc))
! 442 except RuntimeError as exc:
443 # start_run refused because a run is already in flight for this
444 # conversation (the old world silently spawned a duplicate
445 # registry run here — a latent bug, not a contract).
! 446 raise HTTPException(status_code=409, detail=str(exc))
447 if body.reason:
448 # The model's stated reason for requesting auto mode. The old
449 # record.reason field died with /api/chat/auto/sessions; keep the
450 # observability in the log line instead.Lines 469-477 469 )
470 async def stream_conversation_state_events() -> CancellableStreamingResponse:
471 """Registry-level firehose of ``conversation-state`` events (snapshot
472 then live)."""
! 473 return CancellableStreamingResponse(
474 content=_state_firehose_stream(),
475 media_type="text/event-stream",
476 )Lines 573-586 573 raise HTTPException(
574 status_code=404, detail=f"Conversation not found: {session_id}"
575 )
576 if outcome == "invalid":
! 577 raise HTTPException(
578 status_code=409,
579 detail=f"Conversation cannot decline auto mode: {session_id}",
580 )
581 if outcome == "busy":
! 582 raise HTTPException(
583 status_code=409,
584 detail=f"Conversation already has a run in flight: {session_id}",
585 )
586 return Response(status_code=202)Lines 587-596 587 try:
588 outcome = await conversation_supervisor.set_auto_flag(
589 session_id, body.enabled
590 )
! 591 except ConversationCapError as exc:
! 592 raise HTTPException(status_code=429, detail=str(exc))
593 if outcome == "not_found":
594 raise HTTPException(
595 status_code=404, detail=f"Conversation not found: {session_id}"
596 )app/desktop/studio_server/chat/runtime/bus.pyLines 68-87 68 if not line.startswith(b"data: "):
69 continue
70 body = line[6:].strip()
71 if not body or body == b"[DONE]":
! 72 continue
73 try:
74 event = json.loads(body)
! 75 except (ValueError, TypeError):
76 # ValueError covers JSONDecodeError AND UnicodeDecodeError (a
77 # non-UTF-8 byte payload raises the latter before JSON parsing).
! 78 continue
79 if isinstance(event, dict) and event.get("type") == KILN_SSE_CHAT_TRACE:
80 tid = event.get("trace_id")
81 if isinstance(tid, str) and tid:
82 return tid
! 83 return None
84
85
86 class _CloseSentinel:
87 """Pushed onto every subscriber's queue by ``close()`` to end its streamLines 178-186 178 deduping = True
179 while not subscriber.queue.empty():
180 item = subscriber.queue.get_nowait()
181 if isinstance(item, _CloseSentinel):
! 182 return
183 if deduping and item in emitted:
184 continue
185 deduping = False
186 yield itemLines 186-194 186 yield item
187 while True:
188 item = await subscriber.queue.get()
189 if isinstance(item, _CloseSentinel):
! 190 return
191 yield item
192 finally:
193 self._subscribers.discard(subscriber)app/desktop/studio_server/chat/runtime/engine.pyLines 331-340 331 trace_id_for_error = round_state.trace_id_for_error
332 if result.status == "stopped":
333 # Stop pressed mid-retry: the helper surfaced no error;
334 # settle stopped (old auto USER_STOPPED / sub-agent STOPPED).
! 335 self._finish_stopped(record, policy)
! 336 return
337 if result.status != "ok" or round_state is None:
338 # Non-retryable or retry-exhausted upstream error; the
339 # error SSE was already emitted. One-shot runs FAIL (old
340 # sub-agent), others idle with reason "error" — the flagLines 359-367 359 # boundary backstop, idempotent via the None check.
360 # Adopted records (seen_trace_ids seeded at adopt) joined
361 # mid-chain and never stamp from a trace.
362 if record.root_id is None and not record.seen_trace_ids:
! 363 record.root_id = round_state.trace_id
364 record.current_leaf_trace_id = round_state.trace_id
365 if round_state.trace_id not in record.seen_trace_ids:
366 record.seen_trace_ids.append(round_state.trace_id)
367 if io.on_trace is not None:Lines 626-638 626 if not results:
627 if io.stop_requested():
628 # On graceful stop just settle — nothing to surface
629 # for approval (old auto empty-results stop branch).
! 630 self._finish_stopped(record, policy)
! 631 return
632 injected = io.drain_inbox()
633 if injected:
! 634 body = {
635 **body,
636 "messages": [
637 _frame_inbox_message(m, policy) for m in injected
638 ],Lines 636-647 636 "messages": [
637 _frame_inbox_message(m, policy) for m in injected
638 ],
639 }
! 640 continue
641 if policy.one_shot:
! 642 record.state = RunState.COMPLETED
! 643 return
644 # Old auto reason vocabulary: a tool batch that produced
645 # no results settles "done". (The old interactive stream
646 # simply ended its turn here.)
647 self._finish_idle(record, "done")Lines 833-841 833 continue
834 if res.kind == "resolve":
835 assert res.result_json is not None
836 return res.result_json
! 837 return None
838 return None
839
840 async def _resolve_terminal(
841 self,app/desktop/studio_server/chat/runtime/golden_scenarios.pyLines 83-91 83 return list(client.bodies)
84
85
86 async def _consume(stream) -> list[bytes]:
! 87 return [chunk async for chunk in stream]
88
89
90 async def _run_engine(
91 *,Lines 476-501 476 return data["bodies"]
477
478
479 def _write_fixture(name: str, bodies: list[dict[str, Any]]) -> None:
! 480 GOLDEN_DIR.mkdir(exist_ok=True)
! 481 with fixture_path(name).open("w") as f:
! 482 json.dump({"scenario": name, "bodies": bodies}, f, indent=2, ensure_ascii=False)
! 483 f.write("\n")
484
485
486 async def _regenerate_all() -> None:
! 487 for scenario in SCENARIOS:
488 # While an old loop exists it is the reference; once deleted the
489 # engine IS the reference implementation (module docstring), so
490 # deliberate scenario changes re-capture from it.
! 491 capture = scenario.run_old or scenario.run_engine
! 492 bodies = await capture()
! 493 _write_fixture(scenario.name, bodies)
! 494 print(f"wrote {fixture_path(scenario.name)} ({len(bodies)} bodies)")
495
496
497 if __name__ == "__main__":
! 498 import asyncio
499
! 500 asyncio.run(_regenerate_all())app/desktop/studio_server/chat/runtime/models.pyLines 47-55 47
48 if TYPE_CHECKING:
49 # Import cycle avoidance only: interceptors.py imports models for the
50 # context/result types, so the Interceptor callable type is quoted here.
! 51 from app.desktop.studio_server.chat.runtime.interceptors import Interceptor
52
53 # Same id alphabet/length as the old auto (`ar_`) and sub-agent (`sa_`) ids so
54 # ids stay visually consistent across the migration.
55 _ID_ALPHABET = "abcdefghijklmnopqrstuvwxyz234567"Lines 372-380 372 }
373 )
374
375 for tc_id, output in sibling_results.items():
! 376 messages.append(
377 {
378 "role": "tool",
379 "tool_call_id": tc_id,
380 "content": output,Lines 446-459 446
447 def _resolve_positive_int_env(env_var: str, default: int) -> int:
448 raw = os.environ.get(env_var)
449 if raw:
! 450 try:
! 451 value = int(raw)
! 452 if value > 0:
! 453 return value
! 454 except ValueError:
! 455 pass
456 return default
457
458
459 def _resolve_positive_float_env(env_var: str, default: float) -> float:Lines 458-471 458
459 def _resolve_positive_float_env(env_var: str, default: float) -> float:
460 raw = os.environ.get(env_var)
461 if raw:
! 462 try:
! 463 value = float(raw)
! 464 if value > 0:
! 465 return value
! 466 except ValueError:
! 467 pass
468 return default
469
470
471 MessageFraming = Literal["none", "side_note", "steer"]app/desktop/studio_server/chat/runtime/supervisor.pyLines 97-105 97 if TYPE_CHECKING:
98 # Import-cycle avoidance only: chat/orchestration.py imports this module
99 # at module level (it targets the supervisor singleton), so the ctx type
100 # is quoted and resolved lazily where instances are built.
! 101 from app.desktop.studio_server.chat.orchestration import OrchestrationContext
102
103 logger = logging.getLogger(__name__)
104
105 # ── Caps & TTLs: same defaults, env vars, and semantics as the oldLines 136-149 136
137 def _resolve_int_env(env_var: str, default: int) -> int:
138 raw = os.environ.get(env_var)
139 if raw:
! 140 try:
! 141 value = int(raw)
! 142 if value > 0:
! 143 return value
! 144 except ValueError:
! 145 pass
146 return default
147
148
149 # Timeout for the one-off upstream snapshot GET the approval-rehydration pathLines 157-171 157 form some providers persist). Mirrors the web UI's ``extractTextContent``
158 so both ends read the same tail."""
159 if isinstance(content, str):
160 return content
! 161 if isinstance(content, list):
! 162 parts: list[str] = []
! 163 for part in content:
! 164 if isinstance(part, dict) and "text" in part:
! 165 parts.append(str(part["text"]))
! 166 return "".join(parts)
! 167 return ""
168
169
170 def _pending_events_from_trace_tail(
171 trace: list[dict[str, Any]],Lines 201-213 201 last_assistant = trace[idx]
202 last_assistant_idx = idx
203 break
204 if last_assistant is None:
! 205 return [], [], ""
206 assistant_text = _trace_text_content(last_assistant.get("content"))
207 tool_calls = last_assistant.get("tool_calls") or []
208 if not isinstance(tool_calls, list) or not tool_calls:
! 209 return [], [], assistant_text
210 answered = {
211 msg.get("tool_call_id")
212 for msg in trace[last_assistant_idx + 1 :]
213 if msg.get("role") == "tool"Lines 215-228 215 events: list[ToolInputAvailableEvent] = []
216 signal_events: list[ToolInputAvailableEvent] = []
217 for tc in tool_calls:
218 if not isinstance(tc, dict):
! 219 continue
220 tc_id = tc.get("id")
221 function = tc.get("function") or {}
222 name = function.get("name") if isinstance(function, dict) else None
223 if not isinstance(tc_id, str) or not isinstance(name, str) or not name:
! 224 continue
225 if tc_id in answered:
226 continue
227 raw_args = function.get("arguments") if isinstance(function, dict) else None
228 try:Lines 226-235 226 continue
227 raw_args = function.get("arguments") if isinstance(function, dict) else None
228 try:
229 parsed = json.loads(raw_args) if isinstance(raw_args, str) else {}
! 230 except json.JSONDecodeError:
! 231 parsed = {}
232 event = ToolInputAvailableEvent(
233 toolCallId=tc_id,
234 toolName=name,
235 input=parsed if isinstance(parsed, dict) else {},Lines 401-409 401 if session_id is None:
402 return None
403 conv = self._conversations.get(session_id)
404 if conv is None:
! 405 return None
406 record = conv.record
407 if record.kind != "auto" or not record.auto_flag:
408 return None
409 return recordLines 654-662 654 every later subscriber while the batch is parked.
655 """
656 conv = self._conversations.get(session_id)
657 if conv is None or conv.record.state.is_terminal:
! 658 return None
659 if conv.pending_batch is not None and not conv.pending_batch.decided.is_set():
660 # A live (or already-rehydrated) undecided batch is authoritative.
661 return conv.pending_batch
662 if conv.task is not None and not conv.task.done():Lines 660-668 660 # A live (or already-rehydrated) undecided batch is authoritative.
661 return conv.pending_batch
662 if conv.task is not None and not conv.task.done():
663 # A live run owns its own round context — never second-guess it.
! 664 return None
665 # The fetch key mirrors the continuation precedence (phase 6): the
666 # record's own leaf when one is known, else the adopted resume key —
667 # the upstream GET accepts either id kind and returns the session's
668 # CURRENT leaf snapshot either way.Lines 667-675 667 # the upstream GET accepts either id kind and returns the session's
668 # CURRENT leaf snapshot either way.
669 fetch_key = conv.record.current_leaf_trace_id or conv.record.resume_session_key
670 if fetch_key is None:
! 671 return None
672
673 trace = await self._fetch_persisted_trace(conv, fetch_key)
674 if not trace:
675 return NoneLines 740-752 740 timeout=REHYDRATE_FETCH_TIMEOUT_SECONDS
741 ) as client:
742 response = await client.get(url, headers=conv.headers)
743 if response.status_code != 200:
! 744 return None
745 data = response.json()
! 746 except (httpx.HTTPError, json.JSONDecodeError, ValueError):
! 747 logger.debug("Approval rehydration fetch failed for %s", session_key)
! 748 return None
749 if isinstance(data, dict) and conv.record.root_id is None:
750 # Opportunistic backfill of the durable session handle (phase 5):
751 # the upstream snapshot response carries ``session_meta.root_id``
752 # at the top level, and a record adopted from a bare legacy leafLines 756-764 756 conv.record.root_id = root_id
757 task_run = data.get("task_run") if isinstance(data, dict) else None
758 trace = task_run.get("trace") if isinstance(task_run, dict) else None
759 if not isinstance(trace, list):
! 760 return None
761 return [msg for msg in trace if isinstance(msg, dict)]
762
763 # ── Auto mode enable / disable (old chat/auto/ registry + api flows). ──────Lines 865-873 865 # here is the same conversation identity, so spawn lineage/ownership
866 # is unchanged.
867 sibling_results: dict[str, str] = {}
868 if pending_tool_calls:
! 869 sibling_results = await execute_tool_batch(
870 [
871 ToolCallInfo(
872 toolCallId=tc.tool_call_id,
873 toolName=tc.tool_name,Lines 922-930 922 record = conv.record
923 if record.state.is_terminal:
924 # Old parity: disable() on a terminal record reported success
925 # without re-publishing (nothing left to clear).
! 926 return True
927 if not record.auto_flag:
928 return True
929
930 # Pre-mark BEFORE any cancel (CR Moderate 2 — see docstring).Lines 937-946 937 try:
938 await task
939 except asyncio.CancelledError:
940 pass
! 941 except Exception:
! 942 logger.debug(
943 "Conversation %s raised during disable await",
944 session_id,
945 exc_info=True,
946 )Lines 1275-1283 1275 # Terminal bookkeeping (old SubAgentRegistry._settle).
1276 if not record.state.is_terminal:
1277 # Defensive: a one-shot run that exits without a terminal
1278 # state failed in a way we didn't classify.
! 1279 record.state = RunState.FAILED
1280 record.final_report = self._final_report_for(record)
1281 conv.inbox.clear()
1282 conv.settled.set()
1283 self._touch(conv)Lines 1368-1376 1368 """
1369 record = conv.record
1370 drained = conv.drain_inbox()
1371 if not drained:
! 1372 return
1373 messages: list[dict[str, Any]] = [m.as_chat_message() for m in drained]
1374 # Next-turn report injection (parent kinds only — a child never owns a
1375 # report queue), matching send_message's idle branch; reports were not
1376 # echoed at enqueue, so echo them here (once).Lines 1375-1384 1375 # report queue), matching send_message's idle branch; reports were not
1376 # echoed at enqueue, so echo them here (once).
1377 if record.kind != "subagent":
1378 for report in self.drain_reports(record.session_id):
! 1379 messages.append({"role": "user", "content": report})
! 1380 conv.bus.emit(format_user_message(report))
1381 body: dict[str, Any] = {
1382 "messages": messages,
1383 **continuation_key_fields(record),
1384 }Lines 1382-1390 1382 "messages": messages,
1383 **continuation_key_fields(record),
1384 }
1385 if record.auto_flag:
! 1386 body["auto_mode"] = True
1387 self.start_run(record.session_id, body)
1388 logger.info(
1389 "Restarted conversation %s from stranded inbox (%d message(s))",
1390 record.session_id,Lines 1458-1471 1458 # children (their reports have nothing left to consume them).
1459 # If the record had been flipped to the auto policy while this
1460 # interactive turn was still in flight (a manual-enable race),
1461 # swap it back — the model just disabled auto mode.
! 1462 record.idle_reason = "user_disabled"
! 1463 self._touch(conv)
! 1464 self._publish_state(conv)
! 1465 if conv.policy.approvals == "auto" and not conv.policy.one_shot:
! 1466 self._swap_to_interactive(conv)
! 1467 await self.stop_children(record.session_id)
1468
1469 return EngineIO(
1470 emit=conv.bus.emit,
1471 on_trace=on_trace,Lines 1645-1654 1645 try:
1646 await task
1647 except asyncio.CancelledError:
1648 pass
! 1649 except Exception:
! 1650 logger.debug(
1651 "Conversation %s raised during stop await",
1652 session_id,
1653 exc_info=True,
1654 )Lines 1831-1839 1831 reports: list[str] = []
1832 for child_sid in queued:
1833 conv = self._conversations.get(child_sid)
1834 if conv is None or conv.record.report_delivered:
! 1835 continue
1836 reports.append(format_subagent_report(conv.record))
1837 self._mark_report_delivered(conv)
1838 return reportsLines 1842-1850 1842
1843 def _mark_report_delivered(self, conv: _Conversation) -> None:
1844 record = conv.record
1845 if record.report_delivered:
! 1846 return
1847 record.report_delivered = True
1848 if record.parent_session_id is not None:
1849 queued = self._pending_reports.get(record.parent_session_id)
1850 if queued and record.session_id in queued:Lines 1879-1887 1879
1880 def _schedule_gc(self, session_id: str) -> None:
1881 existing = self._gc_tasks.get(session_id)
1882 if existing is not None and not existing.done():
! 1883 return
1884 self._gc_tasks[session_id] = asyncio.create_task(self._gc_after_ttl(session_id))
1885
1886 def _cancel_gc(self, session_id: str) -> None:
1887 gc_task = self._gc_tasks.pop(session_id, None)Lines 1885-1893 1885
1886 def _cancel_gc(self, session_id: str) -> None:
1887 gc_task = self._gc_tasks.pop(session_id, None)
1888 if gc_task is not None:
! 1889 gc_task.cancel()
1890
1891 async def _gc_after_ttl(self, session_id: str) -> None:
1892 try:
1893 await asyncio.sleep(self._terminal_ttl_seconds)Lines 1919-1927 1919
1920 def _evict(self, session_id: str) -> None:
1921 conv = self._conversations.pop(session_id, None)
1922 if conv is None:
! 1923 return
1924 # End any live observer streams: an evicted conversation's bus can
1925 # never emit again, so a subscriber left attached (possible on the
1926 # interactive LRU path in particular — CR n3) would otherwise park
1927 # forever on a dead queue. EOF lets the client re-open from history,app/desktop/studio_server/chat/stream_session.pyLines 31-39 31 from dataclasses import dataclass, field
32 from typing import TYPE_CHECKING, Any, AsyncIterator, Callable, Literal
33
34 if TYPE_CHECKING:
! 35 from app.desktop.studio_server.chat.orchestration import (
36 OrchestrationContext,
37 )
38
39 import httpx
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Code Review
This pull request completes the migration to a unified conversation runtime, replacing the old parallel loops and registries with a single ConversationSupervisor and ConversationEngine. The changes include unifying the lifecycle models, SSE event vocabulary, and orchestration logic. I have identified two performance and robustness improvements: one regarding exception handling in the event bus and another optimizing the eviction process in the supervisor.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/desktop/studio_server/chat/runtime/supervisor.py`:
- Around line 720-759: The URL built in _fetch_persisted_trace is fragile
because it assumes conv.upstream_url always ends with a slash, which can
silently create a bad sessions endpoint. Normalize the join when constructing
the request URL so the sessions path is appended correctly regardless of whether
upstream_url has a trailing slash. Keep the fix localized to
_fetch_persisted_trace and preserve the existing best-effort behavior and error
handling.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: 77471f8b-389e-4e60-880e-4491b7756b9f
📒 Files selected for processing (81)
.gitignoreapp/desktop/desktop_server.pyapp/desktop/studio_server/chat/__init__.pyapp/desktop/studio_server/chat/auto/__init__.pyapp/desktop/studio_server/chat/auto/api.pyapp/desktop/studio_server/chat/auto/events.pyapp/desktop/studio_server/chat/auto/models.pyapp/desktop/studio_server/chat/auto/registry.pyapp/desktop/studio_server/chat/auto/runner.pyapp/desktop/studio_server/chat/auto/sse.pyapp/desktop/studio_server/chat/auto/test_api.pyapp/desktop/studio_server/chat/auto/test_registry.pyapp/desktop/studio_server/chat/auto/test_runner.pyapp/desktop/studio_server/chat/constants.pyapp/desktop/studio_server/chat/helpers.pyapp/desktop/studio_server/chat/orchestration.pyapp/desktop/studio_server/chat/routes.pyapp/desktop/studio_server/chat/runtime/__init__.pyapp/desktop/studio_server/chat/runtime/api.pyapp/desktop/studio_server/chat/runtime/bus.pyapp/desktop/studio_server/chat/runtime/engine.pyapp/desktop/studio_server/chat/runtime/golden/auto_disable_resolve.jsonapp/desktop/studio_server/chat/runtime/golden/auto_report_inbox_unwrapped.jsonapp/desktop/studio_server/chat/runtime/golden/auto_seed_and_tool_round.jsonapp/desktop/studio_server/chat/runtime/golden/interactive_approval_flow.jsonapp/desktop/studio_server/chat/runtime/golden/interactive_report_injection.jsonapp/desktop/studio_server/chat/runtime/golden/interactive_tool_round.jsonapp/desktop/studio_server/chat/runtime/golden/subagent_seed_and_steer.jsonapp/desktop/studio_server/chat/runtime/golden_scenarios.pyapp/desktop/studio_server/chat/runtime/interceptors.pyapp/desktop/studio_server/chat/runtime/models.pyapp/desktop/studio_server/chat/runtime/sse.pyapp/desktop/studio_server/chat/runtime/supervisor.pyapp/desktop/studio_server/chat/runtime/test_api.pyapp/desktop/studio_server/chat/runtime/test_bus.pyapp/desktop/studio_server/chat/runtime/test_engine.pyapp/desktop/studio_server/chat/runtime/test_golden_protocol.pyapp/desktop/studio_server/chat/runtime/test_interceptors.pyapp/desktop/studio_server/chat/runtime/test_supervisor.pyapp/desktop/studio_server/chat/stream_session.pyapp/desktop/studio_server/chat/subagents/__init__.pyapp/desktop/studio_server/chat/subagents/api.pyapp/desktop/studio_server/chat/subagents/events.pyapp/desktop/studio_server/chat/subagents/models.pyapp/desktop/studio_server/chat/subagents/orchestration.pyapp/desktop/studio_server/chat/subagents/registry.pyapp/desktop/studio_server/chat/subagents/runner.pyapp/desktop/studio_server/chat/subagents/sse.pyapp/desktop/studio_server/chat/subagents/test_api.pyapp/desktop/studio_server/chat/subagents/test_orchestration.pyapp/desktop/studio_server/chat/subagents/test_registry.pyapp/desktop/studio_server/chat/subagents/test_runner.pyapp/desktop/studio_server/chat/test_fakes.pyapp/desktop/studio_server/chat/test_integration.pyapp/desktop/studio_server/chat/test_iter_upstream_round.pyapp/desktop/studio_server/chat/test_orchestration.pyapp/desktop/studio_server/chat/test_routes.pyapp/desktop/studio_server/chat/test_stream_session.pyapp/web_ui/src/lib/api_schema.d.tsapp/web_ui/src/lib/chat/auto_run_store.test.tsapp/web_ui/src/lib/chat/auto_run_store.tsapp/web_ui/src/lib/chat/chat_history_apply.tsapp/web_ui/src/lib/chat/chat_session_store.test.tsapp/web_ui/src/lib/chat/chat_session_store.tsapp/web_ui/src/lib/chat/conversation_store.test.tsapp/web_ui/src/lib/chat/conversation_store.tsapp/web_ui/src/lib/chat/session_messages.test.tsapp/web_ui/src/lib/chat/session_messages.tsapp/web_ui/src/lib/chat/streaming_chat.test.tsapp/web_ui/src/lib/chat/streaming_chat.tsapp/web_ui/src/lib/chat/subagent_store.test.tsapp/web_ui/src/lib/chat/subagent_store.tsapp/web_ui/src/routes/(app)/assistant/auto_mode_consent_dialog.test.tsapp/web_ui/src/routes/(app)/assistant/chat.svelteapp/web_ui/src/routes/(app)/assistant/chat_compacting_indicator.test.tsapp/web_ui/src/routes/(app)/assistant/chat_history.svelteapp/web_ui/src/routes/(app)/assistant/chat_queued_message.test.tsapp/web_ui/src/routes/(app)/assistant/subagent_status_dot.svelteapp/web_ui/src/routes/(app)/assistant/subagent_tabs.svelteapp/web_ui/src/routes/(app)/assistant/subagent_transcript.svelteapp/web_ui/src/routes/(app)/assistant/subagent_transcript.test.ts
💤 Files with no reviewable changes (21)
- app/desktop/studio_server/chat/subagents/test_api.py
- app/desktop/studio_server/chat/subagents/events.py
- app/desktop/studio_server/chat/auto/init.py
- app/desktop/studio_server/chat/subagents/api.py
- app/desktop/studio_server/chat/subagents/test_orchestration.py
- app/desktop/studio_server/chat/subagents/sse.py
- app/desktop/studio_server/chat/subagents/models.py
- app/desktop/studio_server/chat/auto/events.py
- app/desktop/studio_server/chat/subagents/runner.py
- app/desktop/studio_server/chat/auto/api.py
- app/desktop/studio_server/chat/auto/test_registry.py
- app/desktop/studio_server/chat/subagents/registry.py
- app/desktop/studio_server/chat/subagents/test_registry.py
- app/desktop/studio_server/chat/auto/models.py
- app/desktop/studio_server/chat/subagents/orchestration.py
- app/desktop/studio_server/chat/auto/sse.py
- app/desktop/studio_server/chat/subagents/test_runner.py
- app/desktop/studio_server/chat/auto/test_api.py
- app/desktop/studio_server/chat/auto/registry.py
- app/desktop/studio_server/chat/auto/runner.py
- app/desktop/studio_server/chat/auto/test_runner.py
Previously, a sub-agent's report woke an AUTO-mode parent immediately (inbox injection) but an INTERACTIVE parent's report merely queued and waited for the user's NEXT message to drain it — so you could spawn helpers, watch them finish, and nothing happened until you poked the parent. That defeated the point of sub-agents (the parent processing their results). _deliver_report now delivers through the parent's inbox for EVERY live parent, auto or interactive: an idle parent is woken to process the report in a normal GATED turn (any mutating tool call still parks for approval; only the already- consented spawns run unprompted), and report panels appear live as children finish. This removes the auto_flag special-case and unifies the two paths. Cascade safety: since delivery now wakes parents, stop_children (parent stopped/disabled/deleted) pre-marks each child report_delivered so a teardown cascade never wakes the parent it is tearing down — while a standalone stop(child) still delivers the child's stop-note so the parent learns the outcome. Queue fallback remains for a gone/off-auto parent (drained on its next turn). Product decision confirmed with the user.
A running sub-agent's tab reached the strip only if the frontend received the child's single spawn-time conversation-state 'running' firehose event AND a resulting fetchChildren survived. Since per-round state publishes were removed, a running child emits that event exactly once — so a firehose micro-drop, the server snapshot/subscribe race, or a fetch-race edge left the running child absent with no recovery until it settled (terminal tabs are dropped by design) or the user refreshed. Two complementary fixes: - Server: BroadcastBus.subscribe gains a snapshot hook so the firehose registers the subscriber BEFORE building the list_records() snapshot — an event published during snapshot construction is now queued, not lost. Safe dedup drops only byte-identical backlog until the first genuinely-new event, so real transitions are never suppressed. - Frontend: a 3.5s reconcile timer (active only while the firehose is connected and a parent is set) re-fetches children, so a missed running-child event converges within seconds instead of needing a manual refresh. Tests: bus snapshot-gap + dedup, firehose end-to-end child-spawned-after-connect, frontend reconcile timer. Live-verified across repeated spawns.
…lake background-sync test - The unified runtime added ten /api/conversations endpoints without their agent_checks annotation files; the schema-bindings job requires them. - test_idle_pause_and_resume raced a fixed 0.3s sleep against a real git fetch on loaded CI runners; poll for the head advance with a deadline instead (same for the fetch-and-fast-forward test).
… robust rehydrate URL join - bus trace scan: catch ValueError (UnicodeDecodeError from non-UTF-8 payloads subclasses it) instead of JSONDecodeError only. - _evict: iterate the record's own seen chain instead of the full trace index; keep the ownership check so an adopted record's eviction never drops another record's entry. - _fetch_persisted_trace: rstrip the upstream URL before joining so a bare base can't silently 404 as 'nothing persisted'.
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@app/desktop/git_sync/test_background_sync.py`:
- Around line 111-115: Ensure background-sync cleanup always runs when polling
times out: in the test containing head_advanced and wait_for, wrap the
post-start assertions and polling in a try/finally block, and call bg.stop() in
the finally clause, matching the cleanup pattern used by the fast-forward test.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: CHILL
Plan: Pro
Run ID: bc02b750-a1c7-42ce-8f94-39e613205ddf
📒 Files selected for processing (11)
app/desktop/git_sync/test_background_sync.pylibs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations.jsonlibs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations_events.jsonlibs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations_session_id.jsonlibs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations_session_id_approvals.jsonlibs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations_session_id_events.jsonlibs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations.jsonlibs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations_session_id_approvals_decisions.jsonlibs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations_session_id_auto.jsonlibs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations_session_id_messages.jsonlibs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations_session_id_stop.json
✅ Files skipped from review due to trivial changes (7)
- libs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations.json
- libs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations_session_id_stop.json
- libs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations_session_id.json
- libs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations_session_id_auto.json
- libs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations_session_id_messages.json
- libs/server/kiln_server/utils/agent_checks/annotations/post_api_conversations.json
- libs/server/kiln_server/utils/agent_checks/annotations/get_api_conversations_session_id_approvals.json
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Kiln Assistant: unified conversation runtime
A deep simplification of the assistant's desktop layer. The assistant had grown three parallel implementations of the same "drive a chat loop against the backend and execute client tools locally" concept — interactive chat (
ChatStreamSession), auto mode (AutoChatRunner), and sub-agents (SubAgentRunner) — each with its own run supervision, event bus, and frontend store. This PR collapses them into one conversation runtime: one loop engine with a pluggable policy, one supervisor, one event bus, and one browser-facing contract (observe / send / decide / stop by a permanent session id). Conversation kinds become policies on the same run.The full design/spec and per-phase plans live in the private kiln_server repo (
specs/projects/assistant_unified_runtime/); this PR is code-only.What changed
chat/runtime/— new package:ConversationEngine(the one loop, reusing the existing round primitives),ConversationSupervisor(records, caps, GC, reports, waits, one settle path), oneByteEventBus+ replay buffer, an interceptor chain for signal tools, unifiedconversation-stateevents, and/api/conversations.chat/auto/,chat/subagents/runtimes,ChatStreamSession's loop,POST /api/chat,/api/chat/execute-tools,/api/chat/auto/*,/api/chat/subagents/*.conversation_store.tsabsorbsauto_run_store.ts+subagent_store.ts;chat_session_store.tsshrinks to UI-session state. Zero visual/UX change.trace_id(which rotates every turn) leaves the desktop↔browser surface entirely; conversations are keyed by a stable session id. The backend gains additive, backward-compatiblesession_idcontinuation (kiln_server side).How it landed
Six reviewed phases (foundation → sub-agents → auto → interactive → session-id surface → backend continuation), each independently shippable with golden-protocol fixtures pinning the deleted loops' exact wire behavior. Plus a follow-up fix for two intermittent queued-message races found during live testing.
Net effect
Desktop chat Python ~11.4k → ~7.1k non-test lines; one place to fix every future loop/lifecycle bug. Full check suite green.
Pairs with
kiln_server PR (backend
session_idcontinuation). Deploy the backend before shipping a desktop release with the final phase.