feat(mobile): correlate voice tasks and recover results on reconnect

tui_gateway/server.py

@@ -2742,6 +2742,7 @@ def _compute_host_turn_frame(
     image_paths: list[str] | None = None,
     queued_prompt_generation: int | None = None,
     display_kind: str | None = None,
+    client_task_id: str | None = None,
 ) -> dict:
     with session["history_lock"]:
         history = list(session.get("history", []))
@@ -2755,6 +2756,7 @@ def _compute_host_turn_frame(
         "type": "turn.start",
         "sid": sid,
         "request_id": rid,
+        **({'client_task_id': client_task_id} if client_task_id else {}),
         "session_key": session.get("session_key") or sid,
         "text": text,
         **({"display_kind": display_kind} if display_kind else {}),
@@ -2930,6 +2932,7 @@ def _submit_prompt_to_compute_host(
     image_paths: list[str] | None = None,
     queued_prompt_generation: int | None = None,
     display_kind: str | None = None,
+    client_task_id: str | None = None,
 ) -> dict:
     cfg = _load_dashboard_process_isolation_config()
     frame = _compute_host_turn_frame(
@@ -2940,6 +2943,7 @@ def _submit_prompt_to_compute_host(
         image_paths=image_paths,
         queued_prompt_generation=queued_prompt_generation,
         display_kind=display_kind,
+        **({'client_task_id': client_task_id} if client_task_id else {}),
     )
 
     def _complete(done: dict) -> None:
@@ -10420,6 +10424,7 @@ def _enqueue_prompt(
     text: Any,
     transport: Any,
     image_paths: list[str] | None = None,
+    client_task_id: str | None = None,
 ) -> None:
     """Stash a message to run as the very next turn once the live one ends.
 
@@ -10438,7 +10443,7 @@ def _enqueue_prompt(
     # Never queue a text-only self-copy of the live inflight user prompt. The
     # live turn already owns that text; draining it after settle would restart
     # the same user turn as a fresh agent invocation.
-    if not image_paths and isinstance(text, str):
+    if not client_task_id and not image_paths and isinstance(text, str):
         turn = session.get("inflight_turn")
         original = (
             str(turn.get("user") or "").strip() if isinstance(turn, dict) else ""
@@ -10446,11 +10451,15 @@ def _enqueue_prompt(
         if original and text.strip() == original:
             return
     queued = {"text": text, "transport": transport}
+    if client_task_id:
+        queued['client_task_id'] = client_task_id
     if image_paths:
         queued["image_paths"] = image_paths
     existing = session.get("queued_prompt")
     if (
         existing
+        and not client_task_id
+        and not existing.get('client_task_id')
         and isinstance(existing.get("text"), str)
         and isinstance(text, str)
         and not existing.get("image_paths")
@@ -10480,7 +10489,7 @@ def _sanitize_queued_entry_vs_inflight_user(
     """
     if not original or not isinstance(entry, dict):
         return entry if isinstance(entry, dict) else None
-    if entry.get("image_paths"):
+    if entry.get("image_paths") or entry.get('client_task_id'):
         return entry
     text = entry.get("text")
     if not isinstance(text, str):
@@ -10578,7 +10587,8 @@ def _interrupt_busy_session(sid: str, session: dict, agent: Any) -> None:
 
 
 def _handle_busy_submit(
-    rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False
+    rid, sid: str, session: dict, text: Any, transport: Any, queued: bool = False,
+    client_task_id: str | None = None,
 ) -> dict | None:
     """Apply the ``display.busy_input_mode`` policy to a prompt that lands while
     a turn is in flight, instead of rejecting it with ``session busy``.
@@ -10656,7 +10666,8 @@ def _handle_busy_submit(
             if image_paths:
                 session["attached_images"] = image_paths + list(session.get("attached_images", []))
             return None
-        _enqueue_prompt(session, text, transport, image_paths=image_paths)
+        _enqueue_prompt(session, text, transport, image_paths=image_paths,
+                        **({'client_task_id': client_task_id} if client_task_id else {}))
         session["last_active"] = time.time()
 
     # Attachments need a separate model invocation. Queue them without
@@ -10717,6 +10728,7 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
             session["running"] = False
             return True
     dispatch_failed = False
+    task_kwargs = {'client_task_id': queued['client_task_id']} if queued.get('client_task_id') else {}
     try:
         if use_compute_host:
             if queued.get("image_paths"):
@@ -10727,10 +10739,11 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
                     queued["text"],
                     image_paths=queued["image_paths"],
                     queued_prompt_generation=queue_generation,
+                    **task_kwargs,
                 )
             else:
                 resp = _submit_prompt_to_compute_host(
-                    rid, sid, session, queued["text"], queued_prompt_generation=queue_generation
+                    rid, sid, session, queued["text"], queued_prompt_generation=queue_generation, **task_kwargs
                 )
             if resp.get("error"):
                 message = str(((resp.get("error") or {}).get("message")) or "queued prompt failed")
@@ -10748,6 +10761,7 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
                     queued["text"],
                     image_paths=queued["image_paths"],
                     queued_prompt_generation=queue_generation,
+                    **task_kwargs,
                 )
             else:
                 _run_prompt_submit(
@@ -10756,6 +10770,7 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
                     session,
                     queued["text"],
                     queued_prompt_generation=queue_generation,
+                    **task_kwargs,
                 )
     except Exception as exc:
         print(
@@ -10767,6 +10782,8 @@ def _drain_queued_prompt(rid, sid: str, session: dict) -> bool:
             session["running"] = False
         dispatch_failed = True
     if dispatch_failed:
+        if queued.get('client_task_id'):
+            _emit('task.complete', sid, {'task_id': queued['client_task_id'], 'status': 'error', 'text': 'Sıradaki görev başlatılamadı.'})
         with session["history_lock"]:
             drain_next = bool(session.get("queued_prompt")) and not session.get(
                 "_turn_cancel_requested"
@@ -13200,6 +13217,7 @@ def _run_prompt_submit(
     image_paths: list[str] | None = None,
     queued_prompt_generation: int | None = None,
     terminal_callback: Callable[[dict[str, Any]], None] | None = None,
+    client_task_id: str | None = None,
 ) -> bool:
     # Ownership admission at the ONE chokepoint every fresh turn source must
     # cross. prompt.submit already claims the slot in its RPC handler (so this
@@ -13263,6 +13281,8 @@ def _run_prompt_submit(
         len(images),
     )
     _emit("message.start", sid)
+    if client_task_id:
+        _emit('task.start', sid, {'task_id': client_task_id})
 
     def run():
         terminal_receipt_attempted = False
@@ -13872,6 +13892,8 @@ def _run_prompt_submit(
             if terminal_receipt_committed:
                 _retire_turn_marker(session, marker_key)
             _emit("message.complete", sid, payload)
+            if client_task_id:
+                _emit('task.complete', sid, {'task_id': client_task_id, 'status': status, 'text': raw if isinstance(raw, str) else str(raw)})
 
             # ── /goal continuation (Ralph-style loop) ─────────────────
             # After every TUI turn, if a /goal is active, ask the judge
@@ -14050,6 +14072,8 @@ def _run_prompt_submit(
             # Keep the partial turn available to the next prompt; the durable
             # inflight record still carries the recoverable error state.
             _restore_agent_history_after_turn_error(session, agent)
+            if client_task_id:
+                _emit('task.complete', sid, {'task_id': client_task_id, 'status': 'error', 'text': 'Görev yürütülürken hata oluştu.'})
             if terminal_callback is not None and not terminal_receipt_attempted:
                 terminal_receipt_attempted = True
                 try:
@@ -17577,7 +17601,7 @@ def _(rid, params: dict) -> dict:
     """
     from hermes_cli.active_sessions import PER_SESSION_EXCLUSIVE_SUBMIT
 
-    return _ok(rid, {"per_session_exclusive_submit": bool(PER_SESSION_EXCLUSIVE_SUBMIT)})
+    return _ok(rid, {"per_session_exclusive_submit": bool(PER_SESSION_EXCLUSIVE_SUBMIT), 'client_task_tracking': True})
 
 
 @method("ping")