refactor floating_domain to structured object-only payload

app/core/deep_agent.py

@@ -23,7 +23,8 @@ from app.db import async_session
 
 logger = logging.getLogger(__name__)
 
-FloatingDomain = Literal["tasks", "projects", "notes", "timelines"]
+FloatingDomainType = Literal["task", "timeline", "project", "node"]
+FloatingDomainSection = Literal["task", "timeline", "note"]
 
 _HOME_SINGLE_AGENT_SYSTEM = (
     "You are the home assistant with direct access to all tools: tasks, projects, notes, timelines, and memory tools. "
@@ -44,8 +45,18 @@ _FLOATING_SINGLE_AGENT_SYSTEM = (
     "Always use tools for factual data retrieval before answering. "
     "When the user asks to remember, forget, or update what you know about them, use memory tools. "
     "If context.context.resolved_project_id exists, use it as project_id for scoped list calls. "
-    "Return markdown and embed inline tags when relevant: <project>[ids]</project>, <task>[ids]</task>, "
-    "<note>[ids]</note>, <timeline>[ids]</timeline>, <chart>{json}</chart>."
+)
+
+_FLOATING_DOMAIN_CLASSIFIER_SYSTEM = (
+    "You are a strict domain classifier for websocket floating requests. "
+    "Return ONLY a JSON object with keys: type, id, section. "
+    "Allowed type values: task, timeline, project, node. "
+    "Allowed section values: task, timeline, note, or null. "
+    "Rules: infer from user message intent first; do not blindly trust scope.type. "
+    "If user asks tasks/timeline/notes for a project, set type=project and section accordingly. "
+    "If project id is unknown but context.resolved_project_id exists, use it as id. "
+    "If id is unknown, use null. "
+    "No markdown, no prose, JSON only."
 )
 
 
@@ -347,27 +358,145 @@ def _all_tools_for_user(user_id: str, trace_id: str | None) -> list[Any]:
     return [*_all_tools(), *_memory_tools(user_id, trace_id)]
 
 
-def _infer_floating_domain(message: str, context: dict[str, Any]) -> FloatingDomain:
-    scope = context.get("scope") if isinstance(context, dict) else None
-    if isinstance(scope, dict):
-        scope_type = str(scope.get("type") or "").strip().lower()
-        if scope_type in {"task", "tasks"}:
-            return "tasks"
-        if scope_type in {"project", "projects"}:
-            return "projects"
-        if scope_type in {"note", "notes"}:
-            return "notes"
-        if scope_type in {"timeline", "timelines"}:
-            return "timelines"
-
+def _detect_domain_section(message: str) -> FloatingDomainSection | None:
     lowered = message.lower()
     if any(keyword in lowered for keyword in ["timeline", "milestone", "release", "schedule"]):
-        return "timelines"
+        return "timeline"
+    if any(keyword in lowered for keyword in ["task", "tasks", "todo", "attivit", "azione"]):
+        return "task"
     if any(keyword in lowered for keyword in ["note", "notes", "memo", "document"]):
-        return "notes"
-    if any(keyword in lowered for keyword in ["project", "progetto", "client"]):
-        return "projects"
-    return "tasks"
+        return "note"
+    return None
+
+
+def _normalize_domain_payload(payload: dict[str, Any], fallback_id: str | None) -> dict[str, str | None]:
+    type_raw = str(payload.get("type") or "").strip().lower()
+    domain_type: FloatingDomainType = "task"
+    if type_raw in {"task", "timeline", "project", "node"}:
+        domain_type = type_raw
+
+    id_value = payload.get("id")
+    domain_id = id_value if isinstance(id_value, str) and id_value.strip() else None
+    if domain_type == "project" and not domain_id:
+        domain_id = fallback_id
+
+    section_raw = payload.get("section")
+    section: FloatingDomainSection | None = None
+    if isinstance(section_raw, str):
+        section_candidate = section_raw.strip().lower()
+        if section_candidate in {"task", "timeline", "note"}:
+            section = section_candidate
+
+    if domain_type != "project":
+        section = None
+
+    return {
+        "type": domain_type,
+        "id": domain_id,
+        "section": section,
+    }
+
+
+def _parse_json_object(text: str) -> dict[str, Any] | None:
+    raw = text.strip()
+    if not raw:
+        return None
+    try:
+        parsed = json.loads(raw)
+        return parsed if isinstance(parsed, dict) else None
+    except json.JSONDecodeError:
+        pass
+
+    match = re.search(r"\{.*\}", raw, re.DOTALL)
+    if not match:
+        return None
+    try:
+        parsed = json.loads(match.group(0))
+    except json.JSONDecodeError:
+        return None
+    return parsed if isinstance(parsed, dict) else None
+
+
+def _infer_floating_domain_rule_based(message: str, context: dict[str, Any]) -> dict[str, str | None]:
+    section = _detect_domain_section(message)
+    scope = context.get("scope") if isinstance(context, dict) else None
+    resolved_project_id = context.get("resolved_project_id") if isinstance(context, dict) else None
+    project_id = resolved_project_id if isinstance(resolved_project_id, str) and resolved_project_id else None
+
+    if isinstance(scope, dict):
+        scope_type = str(scope.get("type") or "").strip().lower()
+        scope_id = scope.get("id")
+        scope_id_value = scope_id if isinstance(scope_id, str) and scope_id else None
+
+        if scope_type in {"task", "tasks"}:
+            return {"type": "task", "id": scope_id_value, "section": None}
+        if scope_type in {"project", "projects"}:
+            project_scope_id = scope_id_value or project_id
+            return {
+                "type": "project",
+                "id": project_scope_id,
+                "section": section,
+            }
+        if scope_type in {"note", "notes"}:
+            return {
+                "type": "node",
+                "id": scope_id_value,
+                "section": None,
+            }
+        if scope_type in {"timeline", "timelines"}:
+            return {"type": "timeline", "id": scope_id_value, "section": None}
+
+    lowered = message.lower()
+    if any(keyword in lowered for keyword in ["project", "progetto", "client"]) or project_id:
+        return {
+            "type": "project",
+            "id": project_id,
+            "section": section,
+        }
+    if section == "timeline":
+        return {"type": "timeline", "id": None, "section": None}
+    if section == "note":
+        return {"type": "node", "id": None, "section": None}
+    return {"type": "task", "id": None, "section": None}
+
+
+async def _infer_floating_domain(message: str, context: dict[str, Any]) -> dict[str, str | None]:
+    resolved_project_id = context.get("resolved_project_id") if isinstance(context, dict) else None
+    project_id = resolved_project_id if isinstance(resolved_project_id, str) and resolved_project_id else None
+
+    classifier_context = {
+        "scope": context.get("scope") if isinstance(context.get("scope"), dict) else None,
+        "resolved_project_id": project_id,
+    }
+
+    try:
+        llm = get_llm()
+        response = await llm.ainvoke(
+            [
+                SystemMessage(content=_FLOATING_DOMAIN_CLASSIFIER_SYSTEM),
+                HumanMessage(
+                    content=(
+                        f"Message:\n{message}\n\n"
+                        f"Context:\n{json.dumps(classifier_context, ensure_ascii=True)}"
+                    )
+                ),
+            ]
+        )
+        parsed = _parse_json_object(_as_text(response.content))
+        if parsed is not None:
+            domain = _normalize_domain_payload(parsed, project_id)
+            logger.info(
+                "deep_agent: floating_domain_classified type=%s id=%s section=%s",
+                domain.get("type"),
+                domain.get("id"),
+                domain.get("section"),
+            )
+            return domain
+        logger.warning("deep_agent: floating_domain classifier returned non-json output")
+    except Exception as exc:
+        logger.warning("deep_agent: floating_domain classifier failed: %s", exc)
+
+    return _infer_floating_domain_rule_based(message, context)
 
 
 async def _run_single_agent(
@@ -558,9 +687,9 @@ async def run_home(user_id: str, message: str, context: dict[str, Any]) -> str:
     return _normalize_tagged_list_lines(response, message)
 
 
-async def run_floating(user_id: str, message: str, context: dict[str, Any]) -> tuple[str, str]:
-    domain = _infer_floating_domain(message, context)
+async def run_floating(user_id: str, message: str, context: dict[str, Any]) -> tuple[str, dict[str, str | None]]:
     prepared_context = await _prepare_context(message, context)
+    domain = await _infer_floating_domain(message, prepared_context)
     response = await _run_single_agent(
         user_id=user_id,
         system_prompt=_FLOATING_SINGLE_AGENT_SYSTEM,
@@ -599,10 +728,10 @@ async def run_floating_stream(
     message: str,
     context: dict[str, Any],
 ) -> AsyncGenerator[tuple[str, Any], None]:
-    domain = _infer_floating_domain(message, context)
+    prepared_context = await _prepare_context(message, context)
+    domain = await _infer_floating_domain(message, prepared_context)
     yield "floating_domain", domain
 
-    prepared_context = await _prepare_context(message, context)
     async for event in _run_single_agent_stream(
         user_id=user_id,
         system_prompt=_FLOATING_SINGLE_AGENT_SYSTEM,

app/core/output_formatter.py

@@ -24,7 +24,11 @@ class StreamFormatter:
 
         async for event_type, data in event_stream:
             if event_type == "floating_domain":
-                yield WsFloatingDomain(request_id=self.request_id, domain=str(data))
+                if isinstance(data, dict):
+                    yield WsFloatingDomain(
+                        request_id=self.request_id,
+                        domain=data,
+                    )
                 continue
 
             if event_type != "token":