2670 lines
122 KiB
Python
2670 lines
122 KiB
Python
# [DEF:AssistantApi:Module]
|
||
# @COMPLEXITY: 5
|
||
# @SEMANTICS: api, assistant, chat, command, confirmation
|
||
# @PURPOSE: API routes for LLM assistant command parsing and safe execution orchestration.
|
||
# @LAYER: API
|
||
# @RELATION: DEPENDS_ON -> [TaskManager]
|
||
# @RELATION: DEPENDS_ON -> [AssistantMessageRecord]
|
||
# @RELATION: DEPENDS_ON -> [AssistantConfirmationRecord]
|
||
# @RELATION: DEPENDS_ON -> [AssistantAuditRecord]
|
||
# @INVARIANT: Risky operations are never executed without valid confirmation token.
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import uuid
|
||
from datetime import datetime, timedelta
|
||
from typing import Any, Dict, List, Optional, Tuple, cast
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, Query, status
|
||
from pydantic import BaseModel, Field
|
||
from sqlalchemy.orm import Session
|
||
from sqlalchemy import desc
|
||
|
||
from ...core.logger import belief_scope, logger
|
||
from ...core.task_manager import TaskManager
|
||
from ...dependencies import (
|
||
get_current_user,
|
||
get_task_manager,
|
||
get_config_manager,
|
||
has_permission,
|
||
)
|
||
from ...core.config_manager import ConfigManager
|
||
from ...core.database import get_db
|
||
from ...services.git_service import GitService
|
||
from ...services.llm_provider import LLMProviderService
|
||
from ...services.llm_prompt_templates import (
|
||
is_multimodal_model,
|
||
normalize_llm_settings,
|
||
resolve_bound_provider_id,
|
||
)
|
||
from ...core.superset_client import SupersetClient
|
||
from ...core.utils.superset_context_extractor import (
|
||
sanitize_imported_filter_for_assistant,
|
||
)
|
||
from ...plugins.llm_analysis.service import LLMClient
|
||
from ...plugins.llm_analysis.models import LLMProviderType
|
||
from ...schemas.auth import User
|
||
from ...models.assistant import (
|
||
AssistantAuditRecord,
|
||
AssistantConfirmationRecord,
|
||
AssistantMessageRecord,
|
||
)
|
||
from ...models.dataset_review import (
|
||
ApprovalState,
|
||
DatasetReviewSession,
|
||
ReadinessState,
|
||
RecommendedAction,
|
||
)
|
||
from ...services.dataset_review.orchestrator import (
|
||
DatasetReviewOrchestrator,
|
||
PreparePreviewCommand,
|
||
)
|
||
from ...services.dataset_review.repositories.session_repository import (
|
||
DatasetReviewSessionRepository,
|
||
DatasetReviewSessionVersionConflictError,
|
||
)
|
||
from .dataset_review import FieldSemanticUpdateRequest, _update_semantic_field_state
|
||
|
||
router = APIRouter(tags=["Assistant"])
|
||
git_service = GitService()
|
||
logger = cast(Any, logger)
|
||
|
||
|
||
# [DEF:AssistantMessageRequest:Class]
|
||
# @COMPLEXITY: 1
|
||
# @PURPOSE: Input payload for assistant message endpoint.
|
||
# @DATA_CONTRACT: Input[conversation_id?:str, message:str(1..4000)] -> Output[AssistantMessageRequest]
|
||
# @RELATION: USED_BY -> [send_message]
|
||
# @SIDE_EFFECT: None (schema declaration only).
|
||
# @PRE: message length is within accepted bounds.
|
||
# @POST: Request object provides message text and optional conversation binding.
|
||
# @INVARIANT: message is always non-empty and no longer than 4000 characters.
|
||
class AssistantMessageRequest(BaseModel):
|
||
conversation_id: Optional[str] = None
|
||
message: str = Field(..., min_length=1, max_length=4000)
|
||
dataset_review_session_id: Optional[str] = None
|
||
|
||
|
||
# [/DEF:AssistantMessageRequest:Class]
|
||
|
||
|
||
# [DEF:AssistantAction:Class]
|
||
# @COMPLEXITY: 1
|
||
# @PURPOSE: UI action descriptor returned with assistant responses.
|
||
# @DATA_CONTRACT: Input[type:str, label:str, target?:str] -> Output[AssistantAction]
|
||
# @RELATION: USED_BY -> [AssistantMessageResponse]
|
||
# @SIDE_EFFECT: None (schema declaration only).
|
||
# @PRE: type and label are provided by orchestration logic.
|
||
# @POST: Action can be rendered as button on frontend.
|
||
# @INVARIANT: type and label are required for every UI action.
|
||
class AssistantAction(BaseModel):
|
||
type: str
|
||
label: str
|
||
target: Optional[str] = None
|
||
|
||
|
||
# [/DEF:AssistantAction:Class]
|
||
|
||
|
||
# [DEF:AssistantMessageResponse:Class]
|
||
# @COMPLEXITY: 1
|
||
# @PURPOSE: Output payload contract for assistant interaction endpoints.
|
||
# @DATA_CONTRACT: Input[conversation_id,response_id,state,text,intent?,confirmation_id?,task_id?,actions[],created_at] -> Output[AssistantMessageResponse]
|
||
# @RELATION: RETURNED_BY -> [send_message]
|
||
# @RELATION: RETURNED_BY -> [confirm_operation]
|
||
# @RELATION: RETURNED_BY -> [cancel_operation]
|
||
# @SIDE_EFFECT: None (schema declaration only).
|
||
# @PRE: Response includes deterministic state and text.
|
||
# @POST: Payload may include task_id/confirmation_id/actions for UI follow-up.
|
||
# @INVARIANT: created_at and state are always present in endpoint responses.
|
||
class AssistantMessageResponse(BaseModel):
|
||
conversation_id: str
|
||
response_id: str
|
||
state: str
|
||
text: str
|
||
intent: Optional[Dict[str, Any]] = None
|
||
confirmation_id: Optional[str] = None
|
||
task_id: Optional[str] = None
|
||
actions: List[AssistantAction] = Field(default_factory=list)
|
||
created_at: datetime
|
||
|
||
|
||
# [/DEF:AssistantMessageResponse:Class]
|
||
|
||
|
||
# [DEF:ConfirmationRecord:Class]
|
||
# @COMPLEXITY: 1
|
||
# @PURPOSE: In-memory confirmation token model for risky operation dispatch.
|
||
# @DATA_CONTRACT: Input[id,user_id,conversation_id,intent,dispatch,expires_at,state?,created_at] -> Output[ConfirmationRecord]
|
||
# @RELATION: USED_BY -> [send_message]
|
||
# @RELATION: USED_BY -> [confirm_operation]
|
||
# @RELATION: USED_BY -> [cancel_operation]
|
||
# @SIDE_EFFECT: None (schema declaration only).
|
||
# @PRE: intent/dispatch/user_id are populated at confirmation request time.
|
||
# @POST: Record tracks lifecycle state and expiry timestamp.
|
||
# @INVARIANT: state defaults to "pending" and expires_at bounds confirmation validity.
|
||
class ConfirmationRecord(BaseModel):
|
||
id: str
|
||
user_id: str
|
||
conversation_id: str
|
||
intent: Dict[str, Any]
|
||
dispatch: Dict[str, Any]
|
||
expires_at: datetime
|
||
state: str = "pending"
|
||
created_at: datetime
|
||
|
||
|
||
# [/DEF:ConfirmationRecord:Class]
|
||
|
||
|
||
CONVERSATIONS: Dict[Tuple[str, str], List[Dict[str, Any]]] = {}
|
||
USER_ACTIVE_CONVERSATION: Dict[str, str] = {}
|
||
CONFIRMATIONS: Dict[str, ConfirmationRecord] = {}
|
||
ASSISTANT_AUDIT: Dict[str, List[Dict[str, Any]]] = {}
|
||
ASSISTANT_ARCHIVE_AFTER_DAYS = 14
|
||
ASSISTANT_MESSAGE_TTL_DAYS = 90
|
||
|
||
INTENT_PERMISSION_CHECKS: Dict[str, List[Tuple[str, str]]] = {
|
||
"get_task_status": [("tasks", "READ")],
|
||
"create_branch": [("plugin:git", "EXECUTE")],
|
||
"commit_changes": [("plugin:git", "EXECUTE")],
|
||
"deploy_dashboard": [("plugin:git", "EXECUTE")],
|
||
"execute_migration": [
|
||
("plugin:migration", "EXECUTE"),
|
||
("plugin:superset-migration", "EXECUTE"),
|
||
],
|
||
"run_backup": [("plugin:superset-backup", "EXECUTE"), ("plugin:backup", "EXECUTE")],
|
||
"run_llm_validation": [("plugin:llm_dashboard_validation", "EXECUTE")],
|
||
"run_llm_documentation": [("plugin:llm_documentation", "EXECUTE")],
|
||
"get_health_summary": [("plugin:migration", "READ")],
|
||
"dataset_review_answer_context": [("dataset:session", "READ")],
|
||
"dataset_review_approve_mappings": [("dataset:session", "MANAGE")],
|
||
"dataset_review_set_field_semantics": [("dataset:session", "MANAGE")],
|
||
"dataset_review_generate_sql_preview": [("dataset:session", "MANAGE")],
|
||
}
|
||
|
||
|
||
# [DEF:_append_history:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Append conversation message to in-memory history buffer.
|
||
# @DATA_CONTRACT: Input[user_id,conversation_id,role,text,state?,task_id?,confirmation_id?] -> Output[None]
|
||
# @RELATION: UPDATES -> [CONVERSATIONS]
|
||
# @SIDE_EFFECT: Mutates in-memory CONVERSATIONS store for user conversation history.
|
||
# @PRE: user_id and conversation_id identify target conversation bucket.
|
||
# @POST: Message entry is appended to CONVERSATIONS key list.
|
||
# @INVARIANT: every appended entry includes generated message_id and created_at timestamp.
|
||
def _append_history(
|
||
user_id: str,
|
||
conversation_id: str,
|
||
role: str,
|
||
text: str,
|
||
state: Optional[str] = None,
|
||
task_id: Optional[str] = None,
|
||
confirmation_id: Optional[str] = None,
|
||
):
|
||
key = (user_id, conversation_id)
|
||
if key not in CONVERSATIONS:
|
||
CONVERSATIONS[key] = []
|
||
CONVERSATIONS[key].append(
|
||
{
|
||
"message_id": str(uuid.uuid4()),
|
||
"conversation_id": conversation_id,
|
||
"role": role,
|
||
"text": text,
|
||
"state": state,
|
||
"task_id": task_id,
|
||
"confirmation_id": confirmation_id,
|
||
"created_at": datetime.utcnow(),
|
||
}
|
||
)
|
||
|
||
|
||
# [/DEF:_append_history:Function]
|
||
|
||
|
||
# [DEF:_persist_message:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Persist assistant/user message record to database.
|
||
# @DATA_CONTRACT: Input[Session,user_id,conversation_id,role,text,state?,task_id?,confirmation_id?,metadata?] -> Output[None]
|
||
# @RELATION: DEPENDS_ON -> [AssistantMessageRecord]
|
||
# @SIDE_EFFECT: Writes AssistantMessageRecord rows and commits or rollbacks the DB session.
|
||
# @PRE: db session is writable and message payload is serializable.
|
||
# @POST: Message row is committed or persistence failure is logged.
|
||
# @INVARIANT: failed persistence attempts always rollback before returning.
|
||
def _persist_message(
|
||
db: Session,
|
||
user_id: str,
|
||
conversation_id: str,
|
||
role: str,
|
||
text: str,
|
||
state: Optional[str] = None,
|
||
task_id: Optional[str] = None,
|
||
confirmation_id: Optional[str] = None,
|
||
metadata: Optional[Dict[str, Any]] = None,
|
||
):
|
||
try:
|
||
row = AssistantMessageRecord(
|
||
id=str(uuid.uuid4()),
|
||
user_id=user_id,
|
||
conversation_id=conversation_id,
|
||
role=role,
|
||
text=text,
|
||
state=state,
|
||
task_id=task_id,
|
||
confirmation_id=confirmation_id,
|
||
payload=metadata,
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
except Exception as exc:
|
||
db.rollback()
|
||
logger.warning(f"[assistant.message][persist_failed] {exc}")
|
||
|
||
|
||
# [/DEF:_persist_message:Function]
|
||
|
||
|
||
# [DEF:_audit:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Append in-memory audit record for assistant decision trace.
|
||
# @DATA_CONTRACT: Input[user_id,payload:Dict[str,Any]] -> Output[None]
|
||
# @RELATION: UPDATES -> [ASSISTANT_AUDIT]
|
||
# @SIDE_EFFECT: Mutates in-memory ASSISTANT_AUDIT store and emits structured log event.
|
||
# @PRE: payload describes decision/outcome fields.
|
||
# @POST: ASSISTANT_AUDIT list for user contains new timestamped entry.
|
||
# @INVARIANT: persisted in-memory audit entry always contains created_at in ISO format.
|
||
def _audit(user_id: str, payload: Dict[str, Any]):
|
||
if user_id not in ASSISTANT_AUDIT:
|
||
ASSISTANT_AUDIT[user_id] = []
|
||
ASSISTANT_AUDIT[user_id].append(
|
||
{**payload, "created_at": datetime.utcnow().isoformat()}
|
||
)
|
||
logger.info(f"[assistant.audit] {payload}")
|
||
|
||
|
||
# [/DEF:_audit:Function]
|
||
|
||
|
||
# [DEF:_persist_audit:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Persist structured assistant audit payload in database.
|
||
# @PRE: db session is writable and payload is JSON-serializable.
|
||
# @POST: Audit row is committed or failure is logged with rollback.
|
||
def _persist_audit(
|
||
db: Session, user_id: str, payload: Dict[str, Any], conversation_id: Optional[str]
|
||
):
|
||
try:
|
||
row = AssistantAuditRecord(
|
||
id=str(uuid.uuid4()),
|
||
user_id=user_id,
|
||
conversation_id=conversation_id,
|
||
decision=payload.get("decision"),
|
||
task_id=payload.get("task_id"),
|
||
message=payload.get("message"),
|
||
payload=payload,
|
||
)
|
||
db.add(row)
|
||
db.commit()
|
||
except Exception as exc:
|
||
db.rollback()
|
||
logger.warning(f"[assistant.audit][persist_failed] {exc}")
|
||
|
||
|
||
# [/DEF:_persist_audit:Function]
|
||
|
||
|
||
# [DEF:_persist_confirmation:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Persist confirmation token record to database.
|
||
# @PRE: record contains id/user/intent/dispatch/expiry fields.
|
||
# @POST: Confirmation row exists in persistent storage.
|
||
def _persist_confirmation(db: Session, record: ConfirmationRecord):
|
||
try:
|
||
row = AssistantConfirmationRecord(
|
||
id=record.id,
|
||
user_id=record.user_id,
|
||
conversation_id=record.conversation_id,
|
||
state=record.state,
|
||
intent=record.intent,
|
||
dispatch=record.dispatch,
|
||
expires_at=record.expires_at,
|
||
created_at=record.created_at,
|
||
consumed_at=None,
|
||
)
|
||
db.merge(row)
|
||
db.commit()
|
||
except Exception as exc:
|
||
db.rollback()
|
||
logger.warning(f"[assistant.confirmation][persist_failed] {exc}")
|
||
|
||
|
||
# [/DEF:_persist_confirmation:Function]
|
||
|
||
|
||
# [DEF:_update_confirmation_state:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Update persistent confirmation token lifecycle state.
|
||
# @PRE: confirmation_id references existing row.
|
||
# @POST: State and consumed_at fields are updated when applicable.
|
||
def _update_confirmation_state(db: Session, confirmation_id: str, state: str):
|
||
try:
|
||
row = (
|
||
db.query(AssistantConfirmationRecord)
|
||
.filter(AssistantConfirmationRecord.id == confirmation_id)
|
||
.first()
|
||
)
|
||
if not row:
|
||
return
|
||
row.state = state
|
||
if state in {"consumed", "expired", "cancelled"}:
|
||
row.consumed_at = datetime.utcnow()
|
||
db.commit()
|
||
except Exception as exc:
|
||
db.rollback()
|
||
logger.warning(f"[assistant.confirmation][update_failed] {exc}")
|
||
|
||
|
||
# [/DEF:_update_confirmation_state:Function]
|
||
|
||
|
||
# [DEF:_load_confirmation_from_db:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Load confirmation token from database into in-memory model.
|
||
# @PRE: confirmation_id may or may not exist in storage.
|
||
# @POST: Returns ConfirmationRecord when found, otherwise None.
|
||
def _load_confirmation_from_db(
|
||
db: Session, confirmation_id: str
|
||
) -> Optional[ConfirmationRecord]:
|
||
row = (
|
||
db.query(AssistantConfirmationRecord)
|
||
.filter(AssistantConfirmationRecord.id == confirmation_id)
|
||
.first()
|
||
)
|
||
if not row:
|
||
return None
|
||
return ConfirmationRecord(
|
||
id=row.id,
|
||
user_id=row.user_id,
|
||
conversation_id=row.conversation_id,
|
||
intent=row.intent or {},
|
||
dispatch=row.dispatch or {},
|
||
expires_at=row.expires_at,
|
||
state=row.state,
|
||
created_at=row.created_at,
|
||
)
|
||
|
||
|
||
# [/DEF:_load_confirmation_from_db:Function]
|
||
|
||
|
||
# [DEF:_ensure_conversation:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve active conversation id in memory or create a new one.
|
||
# @PRE: user_id identifies current actor.
|
||
# @POST: Returns stable conversation id and updates USER_ACTIVE_CONVERSATION.
|
||
def _ensure_conversation(user_id: str, conversation_id: Optional[str]) -> str:
|
||
if conversation_id:
|
||
USER_ACTIVE_CONVERSATION[user_id] = conversation_id
|
||
return conversation_id
|
||
|
||
active = USER_ACTIVE_CONVERSATION.get(user_id)
|
||
if active:
|
||
return active
|
||
|
||
new_id = str(uuid.uuid4())
|
||
USER_ACTIVE_CONVERSATION[user_id] = new_id
|
||
return new_id
|
||
|
||
|
||
# [/DEF:_ensure_conversation:Function]
|
||
|
||
|
||
# [DEF:_resolve_or_create_conversation:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve active conversation using explicit id, memory cache, or persisted history.
|
||
# @PRE: user_id and db session are available.
|
||
# @POST: Returns conversation id and updates USER_ACTIVE_CONVERSATION cache.
|
||
def _resolve_or_create_conversation(
|
||
user_id: str, conversation_id: Optional[str], db: Session
|
||
) -> str:
|
||
if conversation_id:
|
||
USER_ACTIVE_CONVERSATION[user_id] = conversation_id
|
||
return conversation_id
|
||
|
||
active = USER_ACTIVE_CONVERSATION.get(user_id)
|
||
if active:
|
||
return active
|
||
|
||
last_message = (
|
||
db.query(AssistantMessageRecord)
|
||
.filter(AssistantMessageRecord.user_id == user_id)
|
||
.order_by(desc(AssistantMessageRecord.created_at))
|
||
.first()
|
||
)
|
||
if last_message:
|
||
USER_ACTIVE_CONVERSATION[user_id] = last_message.conversation_id
|
||
return last_message.conversation_id
|
||
|
||
new_id = str(uuid.uuid4())
|
||
USER_ACTIVE_CONVERSATION[user_id] = new_id
|
||
return new_id
|
||
|
||
|
||
# [/DEF:_resolve_or_create_conversation:Function]
|
||
|
||
|
||
# [DEF:_cleanup_history_ttl:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Enforce assistant message retention window by deleting expired rows and in-memory records.
|
||
# @PRE: db session is available and user_id references current actor scope.
|
||
# @POST: Messages older than ASSISTANT_MESSAGE_TTL_DAYS are removed from persistence and memory mirrors.
|
||
def _cleanup_history_ttl(db: Session, user_id: str):
|
||
cutoff = datetime.utcnow() - timedelta(days=ASSISTANT_MESSAGE_TTL_DAYS)
|
||
try:
|
||
query = db.query(AssistantMessageRecord).filter(
|
||
AssistantMessageRecord.user_id == user_id,
|
||
AssistantMessageRecord.created_at < cutoff,
|
||
)
|
||
if hasattr(query, "delete"):
|
||
query.delete(synchronize_session=False)
|
||
db.commit()
|
||
except Exception as exc:
|
||
db.rollback()
|
||
logger.warning(
|
||
f"[assistant.history][ttl_cleanup_failed] user={user_id} error={exc}"
|
||
)
|
||
|
||
stale_keys: List[Tuple[str, str]] = []
|
||
for key, items in CONVERSATIONS.items():
|
||
if key[0] != user_id:
|
||
continue
|
||
kept = []
|
||
for item in items:
|
||
created_at = item.get("created_at")
|
||
if isinstance(created_at, datetime) and created_at < cutoff:
|
||
continue
|
||
kept.append(item)
|
||
if kept:
|
||
CONVERSATIONS[key] = kept
|
||
else:
|
||
stale_keys.append(key)
|
||
for key in stale_keys:
|
||
CONVERSATIONS.pop(key, None)
|
||
|
||
|
||
# [/DEF:_cleanup_history_ttl:Function]
|
||
|
||
|
||
# [DEF:_is_conversation_archived:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Determine archived state for a conversation based on last update timestamp.
|
||
# @PRE: updated_at can be null for empty conversations.
|
||
# @POST: Returns True when conversation inactivity exceeds archive threshold.
|
||
def _is_conversation_archived(updated_at: Optional[datetime]) -> bool:
|
||
if not updated_at:
|
||
return False
|
||
cutoff = datetime.utcnow() - timedelta(days=ASSISTANT_ARCHIVE_AFTER_DAYS)
|
||
return updated_at < cutoff
|
||
|
||
|
||
# [/DEF:_is_conversation_archived:Function]
|
||
|
||
|
||
# [DEF:_coerce_query_bool:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Normalize bool-like query values for compatibility in direct handler invocations/tests.
|
||
# @PRE: value may be bool, string, or FastAPI Query metadata object.
|
||
# @POST: Returns deterministic boolean flag.
|
||
def _coerce_query_bool(value: Any) -> bool:
|
||
if isinstance(value, bool):
|
||
return value
|
||
if isinstance(value, str):
|
||
return value.strip().lower() in {"1", "true", "yes", "on"}
|
||
return False
|
||
|
||
|
||
# [/DEF:_coerce_query_bool:Function]
|
||
|
||
|
||
# [DEF:_extract_id:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Extract first regex match group from text by ordered pattern list.
|
||
# @PRE: patterns contain at least one capture group.
|
||
# @POST: Returns first matched token or None.
|
||
def _extract_id(text: str, patterns: List[str]) -> Optional[str]:
|
||
for p in patterns:
|
||
m = re.search(p, text, flags=re.IGNORECASE)
|
||
if m:
|
||
return m.group(1)
|
||
return None
|
||
|
||
|
||
# [/DEF:_extract_id:Function]
|
||
|
||
|
||
# [DEF:_resolve_env_id:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve environment identifier/name token to canonical environment id.
|
||
# @PRE: config_manager provides environment list.
|
||
# @POST: Returns matched environment id or None.
|
||
def _resolve_env_id(
|
||
token: Optional[str], config_manager: ConfigManager
|
||
) -> Optional[str]:
|
||
if not token:
|
||
return None
|
||
|
||
normalized = token.strip().lower()
|
||
envs = config_manager.get_environments()
|
||
for env in envs:
|
||
if env.id.lower() == normalized or env.name.lower() == normalized:
|
||
return env.id
|
||
return None
|
||
|
||
|
||
# [/DEF:_resolve_env_id:Function]
|
||
|
||
|
||
# [DEF:_is_production_env:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Determine whether environment token resolves to production-like target.
|
||
# @PRE: config_manager provides environments or token text is provided.
|
||
# @POST: Returns True for production/prod synonyms, else False.
|
||
def _is_production_env(token: Optional[str], config_manager: ConfigManager) -> bool:
|
||
env_id = _resolve_env_id(token, config_manager)
|
||
if not env_id:
|
||
return (token or "").strip().lower() in {"prod", "production", "прод"}
|
||
|
||
env = next((e for e in config_manager.get_environments() if e.id == env_id), None)
|
||
if not env:
|
||
return False
|
||
target = f"{env.id} {env.name}".lower()
|
||
return "prod" in target or "production" in target or "прод" in target
|
||
|
||
|
||
# [/DEF:_is_production_env:Function]
|
||
|
||
|
||
# [DEF:_resolve_provider_id:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve provider token to provider id with active/default fallback.
|
||
# @PRE: db session can load provider list through LLMProviderService.
|
||
# @POST: Returns provider id or None when no providers configured.
|
||
def _resolve_provider_id(
|
||
provider_token: Optional[str],
|
||
db: Session,
|
||
config_manager: Optional[ConfigManager] = None,
|
||
task_key: Optional[str] = None,
|
||
) -> Optional[str]:
|
||
service = LLMProviderService(db)
|
||
providers = service.get_all_providers()
|
||
if not providers:
|
||
return None
|
||
|
||
if provider_token:
|
||
needle = provider_token.strip().lower()
|
||
for p in providers:
|
||
if p.id.lower() == needle or p.name.lower() == needle:
|
||
return p.id
|
||
|
||
if config_manager and task_key:
|
||
try:
|
||
llm_settings = config_manager.get_config().settings.llm
|
||
bound_provider_id = resolve_bound_provider_id(llm_settings, task_key)
|
||
if bound_provider_id and any(p.id == bound_provider_id for p in providers):
|
||
return bound_provider_id
|
||
except Exception:
|
||
pass
|
||
|
||
active = next((p for p in providers if p.is_active), None)
|
||
return active.id if active else providers[0].id
|
||
|
||
|
||
# [/DEF:_resolve_provider_id:Function]
|
||
|
||
|
||
# [DEF:_get_default_environment_id:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve default environment id from settings or first configured environment.
|
||
# @PRE: config_manager returns environments list.
|
||
# @POST: Returns default environment id or None when environment list is empty.
|
||
def _get_default_environment_id(config_manager: ConfigManager) -> Optional[str]:
|
||
configured = config_manager.get_environments()
|
||
if not configured:
|
||
return None
|
||
preferred = None
|
||
if hasattr(config_manager, "get_config"):
|
||
try:
|
||
preferred = config_manager.get_config().settings.default_environment_id
|
||
except Exception:
|
||
preferred = None
|
||
if preferred and any(env.id == preferred for env in configured):
|
||
return preferred
|
||
explicit_default = next(
|
||
(env.id for env in configured if getattr(env, "is_default", False)), None
|
||
)
|
||
return explicit_default or configured[0].id
|
||
|
||
|
||
# [/DEF:_get_default_environment_id:Function]
|
||
|
||
|
||
# [DEF:_resolve_dashboard_id_by_ref:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve dashboard id by title or slug reference in selected environment.
|
||
# @PRE: dashboard_ref is a non-empty string-like token.
|
||
# @POST: Returns dashboard id when uniquely matched, otherwise None.
|
||
def _resolve_dashboard_id_by_ref(
|
||
dashboard_ref: Optional[str],
|
||
env_id: Optional[str],
|
||
config_manager: ConfigManager,
|
||
) -> Optional[int]:
|
||
if not dashboard_ref or not env_id:
|
||
return None
|
||
env = next(
|
||
(item for item in config_manager.get_environments() if item.id == env_id), None
|
||
)
|
||
if not env:
|
||
return None
|
||
|
||
needle = dashboard_ref.strip().lower()
|
||
try:
|
||
client = SupersetClient(env)
|
||
_, dashboards = client.get_dashboards(query={"page_size": 200})
|
||
except Exception as exc:
|
||
logger.warning(
|
||
f"[assistant.dashboard_resolve][failed] ref={dashboard_ref} env={env_id} error={exc}"
|
||
)
|
||
return None
|
||
|
||
exact = next(
|
||
(
|
||
d
|
||
for d in dashboards
|
||
if str(d.get("slug", "")).lower() == needle
|
||
or str(d.get("dashboard_title", "")).lower() == needle
|
||
or str(d.get("title", "")).lower() == needle
|
||
),
|
||
None,
|
||
)
|
||
if exact:
|
||
return int(exact.get("id"))
|
||
|
||
partial = [
|
||
d
|
||
for d in dashboards
|
||
if needle in str(d.get("dashboard_title", d.get("title", ""))).lower()
|
||
]
|
||
if len(partial) == 1 and partial[0].get("id") is not None:
|
||
return int(partial[0]["id"])
|
||
return None
|
||
|
||
|
||
# [/DEF:_resolve_dashboard_id_by_ref:Function]
|
||
|
||
|
||
# [DEF:_resolve_dashboard_id_entity:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve dashboard id from intent entities using numeric id or dashboard_ref fallback.
|
||
# @PRE: entities may contain dashboard_id as int/str and optional dashboard_ref.
|
||
# @POST: Returns resolved dashboard id or None when ambiguous/unresolvable.
|
||
def _resolve_dashboard_id_entity(
|
||
entities: Dict[str, Any],
|
||
config_manager: ConfigManager,
|
||
env_hint: Optional[str] = None,
|
||
) -> Optional[int]:
|
||
raw_dashboard_id = entities.get("dashboard_id")
|
||
dashboard_ref = entities.get("dashboard_ref")
|
||
|
||
if isinstance(raw_dashboard_id, int):
|
||
return raw_dashboard_id
|
||
|
||
if isinstance(raw_dashboard_id, str):
|
||
token = raw_dashboard_id.strip()
|
||
if token.isdigit():
|
||
return int(token)
|
||
if token and not dashboard_ref:
|
||
dashboard_ref = token
|
||
|
||
if not dashboard_ref:
|
||
return None
|
||
|
||
env_token = (
|
||
env_hint
|
||
or entities.get("environment")
|
||
or entities.get("source_env")
|
||
or entities.get("target_env")
|
||
)
|
||
env_id = (
|
||
_resolve_env_id(env_token, config_manager)
|
||
if env_token
|
||
else _get_default_environment_id(config_manager)
|
||
)
|
||
return _resolve_dashboard_id_by_ref(str(dashboard_ref), env_id, config_manager)
|
||
|
||
|
||
# [/DEF:_resolve_dashboard_id_entity:Function]
|
||
|
||
|
||
# [DEF:_get_environment_name_by_id:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve human-readable environment name by id.
|
||
# @PRE: environment id may be None.
|
||
# @POST: Returns matching environment name or fallback id.
|
||
def _get_environment_name_by_id(
|
||
env_id: Optional[str], config_manager: ConfigManager
|
||
) -> str:
|
||
if not env_id:
|
||
return "unknown"
|
||
env = next(
|
||
(item for item in config_manager.get_environments() if item.id == env_id), None
|
||
)
|
||
return env.name if env else env_id
|
||
|
||
|
||
# [/DEF:_get_environment_name_by_id:Function]
|
||
|
||
|
||
# [DEF:_extract_result_deep_links:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Build deep-link actions to verify task result from assistant chat.
|
||
# @PRE: task object is available.
|
||
# @POST: Returns zero or more assistant actions for dashboard open/diff.
|
||
def _extract_result_deep_links(
|
||
task: Any, config_manager: ConfigManager
|
||
) -> List[AssistantAction]:
|
||
plugin_id = getattr(task, "plugin_id", None)
|
||
params = getattr(task, "params", {}) or {}
|
||
result = getattr(task, "result", {}) or {}
|
||
actions: List[AssistantAction] = []
|
||
dashboard_id: Optional[int] = None
|
||
env_id: Optional[str] = None
|
||
|
||
if plugin_id == "superset-migration":
|
||
migrated = (
|
||
result.get("migrated_dashboards") if isinstance(result, dict) else None
|
||
)
|
||
if isinstance(migrated, list) and migrated:
|
||
first = migrated[0]
|
||
if isinstance(first, dict) and first.get("id") is not None:
|
||
dashboard_id = int(first.get("id"))
|
||
if (
|
||
dashboard_id is None
|
||
and isinstance(params.get("selected_ids"), list)
|
||
and params["selected_ids"]
|
||
):
|
||
dashboard_id = int(params["selected_ids"][0])
|
||
env_id = params.get("target_env_id")
|
||
elif plugin_id == "superset-backup":
|
||
dashboards = result.get("dashboards") if isinstance(result, dict) else None
|
||
if isinstance(dashboards, list) and dashboards:
|
||
first = dashboards[0]
|
||
if isinstance(first, dict) and first.get("id") is not None:
|
||
dashboard_id = int(first.get("id"))
|
||
if (
|
||
dashboard_id is None
|
||
and isinstance(params.get("dashboard_ids"), list)
|
||
and params["dashboard_ids"]
|
||
):
|
||
dashboard_id = int(params["dashboard_ids"][0])
|
||
env_id = params.get("environment_id") or _resolve_env_id(
|
||
result.get("environment"), config_manager
|
||
)
|
||
elif plugin_id == "llm_dashboard_validation":
|
||
if params.get("dashboard_id") is not None:
|
||
dashboard_id = int(params["dashboard_id"])
|
||
env_id = params.get("environment_id")
|
||
|
||
if dashboard_id is not None and env_id:
|
||
env_name = _get_environment_name_by_id(env_id, config_manager)
|
||
actions.append(
|
||
AssistantAction(
|
||
type="open_route",
|
||
label=f"Открыть дашборд в {env_name}",
|
||
target=f"/dashboards/{dashboard_id}?env_id={env_id}",
|
||
)
|
||
)
|
||
if dashboard_id is not None:
|
||
actions.append(
|
||
AssistantAction(
|
||
type="open_diff",
|
||
label="Показать Diff",
|
||
target=str(dashboard_id),
|
||
)
|
||
)
|
||
return actions
|
||
|
||
|
||
# [/DEF:_extract_result_deep_links:Function]
|
||
|
||
|
||
# [DEF:_build_task_observability_summary:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Build compact textual summary for completed tasks to reduce "black box" effect.
|
||
# @PRE: task may contain plugin-specific result payload.
|
||
# @POST: Returns non-empty summary line for known task types or empty string fallback.
|
||
def _build_task_observability_summary(task: Any, config_manager: ConfigManager) -> str:
|
||
plugin_id = getattr(task, "plugin_id", None)
|
||
status = str(getattr(task, "status", "")).upper()
|
||
params = getattr(task, "params", {}) or {}
|
||
result = getattr(task, "result", {}) or {}
|
||
|
||
if plugin_id == "superset-migration" and isinstance(result, dict):
|
||
migrated = len(result.get("migrated_dashboards") or [])
|
||
failed_rows = result.get("failed_dashboards") or []
|
||
failed = len(failed_rows)
|
||
selected = result.get("selected_dashboards", migrated + failed)
|
||
mappings = result.get("mapping_count", 0)
|
||
target_env_id = params.get("target_env_id")
|
||
target_env_name = _get_environment_name_by_id(target_env_id, config_manager)
|
||
warning = ""
|
||
if failed_rows:
|
||
first = failed_rows[0]
|
||
warning = (
|
||
f" Внимание: {first.get('title') or first.get('id')}: "
|
||
f"{first.get('error') or 'ошибка'}."
|
||
)
|
||
return (
|
||
f"Сводка миграции: выбрано {selected}, перенесено {migrated}, "
|
||
f"с ошибками {failed}, маппингов {mappings}, целевая среда {target_env_name}."
|
||
f"{warning}"
|
||
)
|
||
|
||
if plugin_id == "superset-backup" and isinstance(result, dict):
|
||
total = int(result.get("total_dashboards", 0) or 0)
|
||
ok = int(result.get("backed_up_dashboards", 0) or 0)
|
||
failed = int(result.get("failed_dashboards", 0) or 0)
|
||
env_id = params.get("environment_id") or _resolve_env_id(
|
||
result.get("environment"), config_manager
|
||
)
|
||
env_name = _get_environment_name_by_id(env_id, config_manager)
|
||
failures = result.get("failures") or []
|
||
warning = ""
|
||
if failures:
|
||
first = failures[0]
|
||
warning = (
|
||
f" Внимание: {first.get('title') or first.get('id')}: "
|
||
f"{first.get('error') or 'ошибка'}."
|
||
)
|
||
return (
|
||
f"Сводка бэкапа: среда {env_name}, всего {total}, успешно {ok}, "
|
||
f"с ошибками {failed}. {status}.{warning}"
|
||
)
|
||
|
||
if plugin_id == "llm_dashboard_validation" and isinstance(result, dict):
|
||
report_status = result.get("status") or status
|
||
report_summary = result.get("summary") or "Итог недоступен."
|
||
issues = result.get("issues") or []
|
||
return f"Сводка валидации: статус {report_status}, проблем {len(issues)}. {report_summary}"
|
||
|
||
# Fallback for unknown task payloads.
|
||
if status in {"SUCCESS", "FAILED"}:
|
||
return f"Задача завершена со статусом {status}."
|
||
return ""
|
||
|
||
|
||
# [/DEF:_build_task_observability_summary:Function]
|
||
|
||
|
||
from src.logger import belief_scope, logger
|
||
|
||
# [DEF:_parse_command:Function]
|
||
# @COMPLEXITY: 4
|
||
# @PURPOSE: Deterministically parse RU/EN command text into intent payload.
|
||
# @DATA_CONTRACT: Input[message:str, config_manager:ConfigManager] -> Output[Dict[str,Any]{domain,operation,entities,confidence,risk_level,requires_confirmation}]
|
||
# @RELATION: DEPENDS_ON -> [_extract_id]
|
||
# @RELATION: DEPENDS_ON -> [_is_production_env]
|
||
# @SIDE_EFFECT: None (pure parsing logic).
|
||
# @PRE: message contains raw user text and config manager resolves environments.
|
||
# @POST: Returns intent dict with domain/operation/entities/confidence/risk fields.
|
||
# @INVARIANT: every return path includes domain, operation, entities, confidence, risk_level, requires_confirmation.
|
||
def _parse_command(message: str, config_manager: ConfigManager) -> Dict[str, Any]:
|
||
with belief_scope('_parse_command'):
|
||
logger.reason('Belief protocol reasoning checkpoint for _parse_command')
|
||
text = message.strip()
|
||
lower = text.lower()
|
||
if any((phrase in lower for phrase in ['что ты умеешь', 'что умеешь', 'что ты можешь', 'help', 'помощь', 'доступные команды', 'какие команды'])):
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'assistant', 'operation': 'show_capabilities', 'entities': {}, 'confidence': 0.98, 'risk_level': 'safe', 'requires_confirmation': False}
|
||
dashboard_id = _extract_id(lower, ['(?:дашборд\\w*|dashboard)\\s*(?:id\\s*)?(\\d+)'])
|
||
dashboard_ref = _extract_id(lower, ['(?:дашборд\\w*|dashboard)\\s*(?:id\\s*)?([a-zа-я0-9._-]+)'])
|
||
dataset_id = _extract_id(lower, ['(?:датасет\\w*|dataset)\\s*(?:id\\s*)?(\\d+)'])
|
||
task_id = _extract_id(lower, ['(task[-_a-z0-9]{1,}|[0-9a-f]{8}-[0-9a-f-]{27,})'])
|
||
if any((k in lower for k in ['статус', 'status', 'state', 'проверь задачу'])):
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'status', 'operation': 'get_task_status', 'entities': {'task_id': task_id}, 'confidence': 0.92 if task_id else 0.66, 'risk_level': 'safe', 'requires_confirmation': False}
|
||
if any((k in lower for k in ['ветк', 'branch'])) and any((k in lower for k in ['созд', 'сделай', 'create'])):
|
||
branch = _extract_id(lower, ['(?:ветк\\w*|branch)\\s+([a-z0-9._/-]+)'])
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'git', 'operation': 'create_branch', 'entities': {'dashboard_id': int(dashboard_id) if dashboard_id else None, 'branch_name': branch}, 'confidence': 0.95 if branch and dashboard_id else 0.7, 'risk_level': 'guarded', 'requires_confirmation': False}
|
||
if any((k in lower for k in ['коммит', 'commit'])):
|
||
quoted = re.search('"([^"]{3,120})"', text)
|
||
message_text = quoted.group(1) if quoted else 'assistant: update dashboard changes'
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'git', 'operation': 'commit_changes', 'entities': {'dashboard_id': int(dashboard_id) if dashboard_id else None, 'message': message_text}, 'confidence': 0.9 if dashboard_id else 0.7, 'risk_level': 'guarded', 'requires_confirmation': False}
|
||
if any((k in lower for k in ['деплой', 'deploy', 'разверн'])):
|
||
env_match = _extract_id(lower, ['(?:в|to)\\s+([a-z0-9_-]+)'])
|
||
is_dangerous = _is_production_env(env_match, config_manager)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'git', 'operation': 'deploy_dashboard', 'entities': {'dashboard_id': int(dashboard_id) if dashboard_id else None, 'environment': env_match}, 'confidence': 0.92 if dashboard_id and env_match else 0.7, 'risk_level': 'dangerous' if is_dangerous else 'guarded', 'requires_confirmation': is_dangerous}
|
||
if any((k in lower for k in ['миграц', 'migration', 'migrate'])):
|
||
src = _extract_id(lower, ['(?:с|from)\\s+([a-z0-9_-]+)'])
|
||
tgt = _extract_id(lower, ['(?:на|to)\\s+([a-z0-9_-]+)'])
|
||
dry_run = '--dry-run' in lower or 'dry run' in lower
|
||
replace_db_config = '--replace-db-config' in lower
|
||
fix_cross_filters = '--no-fix-cross-filters' not in lower
|
||
is_dangerous = _is_production_env(tgt, config_manager)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'migration', 'operation': 'execute_migration', 'entities': {'dashboard_id': int(dashboard_id) if dashboard_id else None, 'source_env': src, 'target_env': tgt, 'dry_run': dry_run, 'replace_db_config': replace_db_config, 'fix_cross_filters': fix_cross_filters}, 'confidence': 0.95 if dashboard_id and src and tgt else 0.72, 'risk_level': 'dangerous' if is_dangerous else 'guarded', 'requires_confirmation': is_dangerous or dry_run}
|
||
if any((k in lower for k in ['бэкап', 'backup', 'резерв'])):
|
||
env_match = _extract_id(lower, ['(?:в|for|из|from)\\s+([a-z0-9_-]+)'])
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'backup', 'operation': 'run_backup', 'entities': {'dashboard_id': int(dashboard_id) if dashboard_id else None, 'environment': env_match}, 'confidence': 0.9 if env_match else 0.7, 'risk_level': 'guarded', 'requires_confirmation': False}
|
||
if any((k in lower for k in ['здоровье', 'health', 'ошибки', 'failing', 'проблемы'])):
|
||
env_match = _extract_id(lower, ['(?:в|for|env|окружени[ея])\\s+([a-z0-9_-]+)'])
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'health', 'operation': 'get_health_summary', 'entities': {'environment': env_match}, 'confidence': 0.9, 'risk_level': 'safe', 'requires_confirmation': False}
|
||
if any((k in lower for k in ['валидац', 'validate', 'провер'])):
|
||
env_match = _extract_id(lower, ['(?:в|for|env|окружени[ея])\\s+([a-z0-9_-]+)'])
|
||
provider_match = _extract_id(lower, ['(?:provider|провайдер)\\s+([a-z0-9_-]+)'])
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'llm', 'operation': 'run_llm_validation', 'entities': {'dashboard_id': int(dashboard_id) if dashboard_id else None, 'dashboard_ref': dashboard_ref if dashboard_ref and (not dashboard_ref.isdigit()) else None, 'environment': env_match, 'provider': provider_match}, 'confidence': 0.88 if dashboard_id else 0.64, 'risk_level': 'guarded', 'requires_confirmation': False}
|
||
if any((k in lower for k in ['документац', 'documentation', 'generate docs', 'сгенерируй док'])):
|
||
env_match = _extract_id(lower, ['(?:в|for|env|окружени[ея])\\s+([a-z0-9_-]+)'])
|
||
provider_match = _extract_id(lower, ['(?:provider|провайдер)\\s+([a-z0-9_-]+)'])
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'llm', 'operation': 'run_llm_documentation', 'entities': {'dataset_id': int(dataset_id) if dataset_id else None, 'environment': env_match, 'provider': provider_match}, 'confidence': 0.88 if dataset_id else 0.64, 'risk_level': 'guarded', 'requires_confirmation': False}
|
||
logger.reflect('Belief protocol postcondition checkpoint for _parse_command')
|
||
return {'domain': 'unknown', 'operation': 'clarify', 'entities': {}, 'confidence': 0.3, 'risk_level': 'safe', 'requires_confirmation': False}
|
||
# [/DEF:_parse_command:Function]
|
||
|
||
|
||
# [DEF:_check_any_permission:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Validate user against alternative permission checks (logical OR).
|
||
# @PRE: checks list contains resource-action tuples.
|
||
# @POST: Returns on first successful permission; raises 403-like HTTPException otherwise.
|
||
def _check_any_permission(current_user: User, checks: List[Tuple[str, str]]):
|
||
errors: List[HTTPException] = []
|
||
for resource, action in checks:
|
||
try:
|
||
has_permission(resource, action)(current_user)
|
||
return
|
||
except HTTPException as exc:
|
||
errors.append(exc)
|
||
|
||
raise (
|
||
errors[-1]
|
||
if errors
|
||
else HTTPException(status_code=403, detail="Permission denied")
|
||
)
|
||
|
||
|
||
# [/DEF:_check_any_permission:Function]
|
||
|
||
|
||
# [DEF:_has_any_permission:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Check whether user has at least one permission tuple from the provided list.
|
||
# @PRE: current_user and checks list are valid.
|
||
# @POST: Returns True when at least one permission check passes.
|
||
def _has_any_permission(current_user: User, checks: List[Tuple[str, str]]) -> bool:
|
||
try:
|
||
_check_any_permission(current_user, checks)
|
||
return True
|
||
except HTTPException:
|
||
return False
|
||
|
||
|
||
# [/DEF:_has_any_permission:Function]
|
||
|
||
|
||
# [DEF:_build_tool_catalog:Function]
|
||
# @COMPLEXITY: 3
|
||
# @PURPOSE: Build current-user tool catalog for LLM planner with operation contracts and defaults.
|
||
# @PRE: current_user is authenticated; config/db are available.
|
||
# @POST: Returns list of executable tools filtered by permission and runtime availability.
|
||
def _build_tool_catalog(
|
||
current_user: User,
|
||
config_manager: ConfigManager,
|
||
db: Session,
|
||
dataset_review_context: Optional[Dict[str, Any]] = None,
|
||
) -> List[Dict[str, Any]]:
|
||
envs = config_manager.get_environments()
|
||
default_env_id = _get_default_environment_id(config_manager)
|
||
providers = LLMProviderService(db).get_all_providers()
|
||
llm_settings = {}
|
||
try:
|
||
llm_settings = config_manager.get_config().settings.llm
|
||
except Exception:
|
||
llm_settings = {}
|
||
active_provider = next((p.id for p in providers if p.is_active), None)
|
||
fallback_provider = active_provider or (providers[0].id if providers else None)
|
||
validation_provider = (
|
||
resolve_bound_provider_id(llm_settings, "dashboard_validation")
|
||
or fallback_provider
|
||
)
|
||
documentation_provider = (
|
||
resolve_bound_provider_id(llm_settings, "documentation") or fallback_provider
|
||
)
|
||
|
||
candidates: List[Dict[str, Any]] = [
|
||
{
|
||
"operation": "show_capabilities",
|
||
"domain": "assistant",
|
||
"description": "Show available assistant commands and examples",
|
||
"required_entities": [],
|
||
"optional_entities": [],
|
||
"risk_level": "safe",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "get_task_status",
|
||
"domain": "status",
|
||
"description": "Get task status by task_id or latest user task",
|
||
"required_entities": [],
|
||
"optional_entities": ["task_id"],
|
||
"risk_level": "safe",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "create_branch",
|
||
"domain": "git",
|
||
"description": "Create git branch for dashboard by id/slug/title",
|
||
"required_entities": ["branch_name"],
|
||
"optional_entities": ["dashboard_id", "dashboard_ref"],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "commit_changes",
|
||
"domain": "git",
|
||
"description": "Commit dashboard repository changes by dashboard id/slug/title",
|
||
"required_entities": [],
|
||
"optional_entities": ["dashboard_id", "dashboard_ref", "message"],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "deploy_dashboard",
|
||
"domain": "git",
|
||
"description": "Deploy dashboard (id/slug/title) to target environment",
|
||
"required_entities": ["environment"],
|
||
"optional_entities": ["dashboard_id", "dashboard_ref"],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "execute_migration",
|
||
"domain": "migration",
|
||
"description": "Run dashboard migration (id/slug/title) between environments. Optional boolean flags: replace_db_config, fix_cross_filters",
|
||
"required_entities": ["source_env", "target_env"],
|
||
"optional_entities": [
|
||
"dashboard_id",
|
||
"dashboard_ref",
|
||
"replace_db_config",
|
||
"fix_cross_filters",
|
||
],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "run_backup",
|
||
"domain": "backup",
|
||
"description": "Run backup for environment or specific dashboard by id/slug/title",
|
||
"required_entities": ["environment"],
|
||
"optional_entities": ["dashboard_id", "dashboard_ref"],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "run_llm_validation",
|
||
"domain": "llm",
|
||
"description": "Run LLM dashboard validation by dashboard id/slug/title",
|
||
"required_entities": [],
|
||
"optional_entities": ["dashboard_ref", "environment", "provider"],
|
||
"defaults": {
|
||
"environment": default_env_id,
|
||
"provider": validation_provider,
|
||
},
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "run_llm_documentation",
|
||
"domain": "llm",
|
||
"description": "Generate dataset documentation via LLM",
|
||
"required_entities": ["dataset_id"],
|
||
"optional_entities": ["environment", "provider"],
|
||
"defaults": {
|
||
"environment": default_env_id,
|
||
"provider": documentation_provider,
|
||
},
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "get_health_summary",
|
||
"domain": "health",
|
||
"description": "Get summary of dashboard health and failing validations",
|
||
"required_entities": [],
|
||
"optional_entities": ["environment"],
|
||
"risk_level": "safe",
|
||
"requires_confirmation": False,
|
||
},
|
||
]
|
||
|
||
available: List[Dict[str, Any]] = []
|
||
for tool in candidates:
|
||
checks = INTENT_PERMISSION_CHECKS.get(tool["operation"], [])
|
||
if checks and not _has_any_permission(current_user, checks):
|
||
continue
|
||
available.append(tool)
|
||
|
||
if dataset_review_context is not None:
|
||
dataset_tools: List[Dict[str, Any]] = [
|
||
{
|
||
"operation": "dataset_review_answer_context",
|
||
"domain": "dataset_review",
|
||
"description": "Answer questions using the currently bound dataset review session context",
|
||
"required_entities": ["dataset_review_session_id"],
|
||
"optional_entities": [],
|
||
"risk_level": "safe",
|
||
"requires_confirmation": False,
|
||
},
|
||
{
|
||
"operation": "dataset_review_approve_mappings",
|
||
"domain": "dataset_review",
|
||
"description": "Approve warning-sensitive execution mappings in the current dataset review session",
|
||
"required_entities": ["dataset_review_session_id", "session_version"],
|
||
"optional_entities": ["mapping_ids"],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": True,
|
||
},
|
||
{
|
||
"operation": "dataset_review_set_field_semantics",
|
||
"domain": "dataset_review",
|
||
"description": "Apply explicit semantic field override or candidate selection in the current dataset review session",
|
||
"required_entities": [
|
||
"dataset_review_session_id",
|
||
"session_version",
|
||
"field_id",
|
||
],
|
||
"optional_entities": [
|
||
"candidate_id",
|
||
"verbose_name",
|
||
"description",
|
||
"display_format",
|
||
"lock_field",
|
||
],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": True,
|
||
},
|
||
{
|
||
"operation": "dataset_review_generate_sql_preview",
|
||
"domain": "dataset_review",
|
||
"description": "Generate a Superset-compiled SQL preview for the current dataset review session",
|
||
"required_entities": ["dataset_review_session_id", "session_version"],
|
||
"optional_entities": [],
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": True,
|
||
},
|
||
]
|
||
for tool in dataset_tools:
|
||
checks = INTENT_PERMISSION_CHECKS.get(tool["operation"], [])
|
||
if checks and not _has_any_permission(current_user, checks):
|
||
continue
|
||
available.append(tool)
|
||
return available
|
||
|
||
|
||
# [/DEF:_build_tool_catalog:Function]
|
||
|
||
|
||
# [DEF:_coerce_intent_entities:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Normalize intent entity value types from LLM output to route-compatible values.
|
||
# @PRE: intent contains entities dict or missing entities.
|
||
# @POST: Returned intent has numeric ids coerced where possible and string values stripped.
|
||
def _coerce_intent_entities(intent: Dict[str, Any]) -> Dict[str, Any]:
|
||
entities = intent.get("entities")
|
||
if not isinstance(entities, dict):
|
||
intent["entities"] = {}
|
||
entities = intent["entities"]
|
||
for key in ("dashboard_id", "dataset_id"):
|
||
value = entities.get(key)
|
||
if isinstance(value, str) and value.strip().isdigit():
|
||
entities[key] = int(value.strip())
|
||
for key, value in list(entities.items()):
|
||
if isinstance(value, str):
|
||
entities[key] = value.strip()
|
||
return intent
|
||
|
||
|
||
# [/DEF:_coerce_intent_entities:Function]
|
||
|
||
|
||
# Operations that are read-only and do not require confirmation.
|
||
_SAFE_OPS = {
|
||
"show_capabilities",
|
||
"get_task_status",
|
||
"get_health_summary",
|
||
"dataset_review_answer_context",
|
||
}
|
||
|
||
_DATASET_REVIEW_OPS = {
|
||
"dataset_review_approve_mappings",
|
||
"dataset_review_set_field_semantics",
|
||
"dataset_review_generate_sql_preview",
|
||
}
|
||
|
||
|
||
# [DEF:_serialize_dataset_review_context:Function]
|
||
# @COMPLEXITY: 4
|
||
# @PURPOSE: Build assistant-safe dataset-review context snapshot with masked imported-filter payloads for session-scoped assistant routing.
|
||
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSession]
|
||
def _serialize_dataset_review_context(session: DatasetReviewSession) -> Dict[str, Any]:
|
||
with belief_scope('_serialize_dataset_review_context'):
|
||
logger.reason('Belief protocol reasoning checkpoint for _serialize_dataset_review_context')
|
||
latest_preview = None
|
||
previews = getattr(session, 'previews', []) or []
|
||
if previews:
|
||
latest_preview = previews[-1]
|
||
logger.reflect('Belief protocol postcondition checkpoint for _serialize_dataset_review_context')
|
||
return {'session_id': session.session_id, 'version': int(getattr(session, 'version', 0) or 0), 'dataset_ref': session.dataset_ref, 'dataset_id': session.dataset_id, 'environment_id': session.environment_id, 'readiness_state': session.readiness_state.value, 'recommended_action': session.recommended_action.value, 'status': session.status.value, 'current_phase': session.current_phase.value, 'findings': [{'finding_id': item.finding_id, 'code': item.code, 'severity': item.severity.value, 'message': item.message, 'resolution_state': item.resolution_state.value} for item in getattr(session, 'findings', [])], 'imported_filters': [sanitize_imported_filter_for_assistant({'filter_id': item.filter_id, 'filter_name': item.filter_name, 'display_name': item.display_name, 'raw_value': item.raw_value, 'raw_value_masked': bool(getattr(item, 'raw_value_masked', False)), 'normalized_value': item.normalized_value, 'source': getattr(item.source, 'value', item.source), 'confidence_state': getattr(item.confidence_state, 'value', item.confidence_state), 'requires_confirmation': bool(item.requires_confirmation), 'recovery_status': getattr(item.recovery_status, 'value', item.recovery_status), 'notes': item.notes}) for item in getattr(session, 'imported_filters', [])], 'mappings': [{'mapping_id': item.mapping_id, 'filter_id': item.filter_id, 'variable_id': item.variable_id, 'mapping_method': getattr(item.mapping_method, 'value', item.mapping_method), 'effective_value': item.effective_value, 'approval_state': getattr(item.approval_state, 'value', item.approval_state), 'requires_explicit_approval': bool(item.requires_explicit_approval)} for item in getattr(session, 'execution_mappings', [])], 'semantic_fields': [{'field_id': item.field_id, 'field_name': item.field_name, 'verbose_name': item.verbose_name, 'description': item.description, 'display_format': item.display_format, 'provenance': getattr(item.provenance, 'value', item.provenance), 'is_locked': bool(item.is_locked), 'needs_review': bool(item.needs_review), 'candidates': [{'candidate_id': candidate.candidate_id, 'proposed_verbose_name': candidate.proposed_verbose_name, 'proposed_description': candidate.proposed_description, 'proposed_display_format': candidate.proposed_display_format, 'status': getattr(candidate.status, 'value', candidate.status)} for candidate in getattr(item, 'candidates', [])]} for item in getattr(session, 'semantic_fields', [])], 'preview': {'preview_id': latest_preview.preview_id, 'preview_status': getattr(latest_preview.preview_status, 'value', latest_preview.preview_status), 'compiled_sql': latest_preview.compiled_sql} if latest_preview is not None else None}
|
||
# [/DEF:_serialize_dataset_review_context:Function]
|
||
|
||
|
||
# [DEF:_load_dataset_review_context:Function]
|
||
# @COMPLEXITY: 4
|
||
# @PURPOSE: Load owner-scoped dataset-review context for assistant planning and grounded response generation.
|
||
# @RELATION: [DEPENDS_ON] ->[DatasetReviewSessionRepository]
|
||
def _load_dataset_review_context(dataset_review_session_id: Optional[str], current_user: User, db: Session) -> Optional[Dict[str, Any]]:
|
||
with belief_scope('_load_dataset_review_context'):
|
||
if not dataset_review_session_id:
|
||
return None
|
||
logger.reason('Belief protocol reasoning checkpoint for _load_dataset_review_context')
|
||
repository = DatasetReviewSessionRepository(db)
|
||
session = repository.load_session_detail(dataset_review_session_id, current_user.id)
|
||
if session is None or session.user_id != current_user.id:
|
||
raise HTTPException(status_code=404, detail='Dataset review session not found')
|
||
logger.reflect('Belief protocol postcondition checkpoint for _load_dataset_review_context')
|
||
return _serialize_dataset_review_context(session)
|
||
# [/DEF:_load_dataset_review_context:Function]
|
||
|
||
|
||
# [DEF:_extract_dataset_review_target:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Extract structured dataset-review focus target hints embedded in assistant prompts.
|
||
def _extract_dataset_review_target(message: str) -> Tuple[Optional[str], Optional[str]]:
|
||
match = re.search(
|
||
r"(?:target|focus)\s*[:=]\s*(field|mapping|finding|filter)[:=]([A-Za-z0-9._-]+)",
|
||
str(message or ""),
|
||
re.IGNORECASE,
|
||
)
|
||
if not match:
|
||
return None, None
|
||
return match.group(1).lower(), match.group(2)
|
||
|
||
|
||
# [/DEF:_extract_dataset_review_target:Function]
|
||
|
||
|
||
# [DEF:_match_dataset_review_field:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Resolve one semantic field from assistant-visible context by id or user-visible label.
|
||
def _match_dataset_review_field(
|
||
dataset_context: Dict[str, Any],
|
||
message: str,
|
||
) -> Optional[Dict[str, Any]]:
|
||
target_kind, target_id = _extract_dataset_review_target(message)
|
||
fields = dataset_context.get("semantic_fields", []) or []
|
||
if target_kind == "field" and target_id:
|
||
return next(
|
||
(item for item in fields if str(item.get("field_id")) == str(target_id)),
|
||
None,
|
||
)
|
||
|
||
normalized_message = str(message or "").lower()
|
||
for field in fields:
|
||
if str(field.get("field_id", "")).lower() in normalized_message:
|
||
return field
|
||
field_name = str(field.get("field_name", "")).lower()
|
||
if field_name and field_name in normalized_message:
|
||
return field
|
||
verbose_name = str(field.get("verbose_name", "")).lower()
|
||
if verbose_name and verbose_name in normalized_message:
|
||
return field
|
||
return None
|
||
|
||
|
||
# [/DEF:_match_dataset_review_field:Function]
|
||
|
||
|
||
# [DEF:_extract_quoted_segment:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Extract one quoted assistant command segment after a label token.
|
||
def _extract_quoted_segment(message: str, label: str) -> Optional[str]:
|
||
pattern = rf"{label}\s*[=:]?\s*[\"']([^\"']+)[\"']"
|
||
match = re.search(pattern, str(message or ""), re.IGNORECASE)
|
||
return match.group(1).strip() if match else None
|
||
|
||
|
||
# [/DEF:_extract_quoted_segment:Function]
|
||
|
||
|
||
# [DEF:_dataset_review_conflict_http_exception:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Convert dataset-review optimistic-lock conflicts into shared 409 assistant semantics.
|
||
def _dataset_review_conflict_http_exception(
|
||
exc: DatasetReviewSessionVersionConflictError,
|
||
) -> HTTPException:
|
||
return HTTPException(
|
||
status_code=status.HTTP_409_CONFLICT,
|
||
detail={
|
||
"error_code": "session_version_conflict",
|
||
"message": str(exc),
|
||
"session_id": exc.session_id,
|
||
"expected_version": exc.expected_version,
|
||
"actual_version": exc.actual_version,
|
||
},
|
||
)
|
||
|
||
|
||
# [/DEF:_dataset_review_conflict_http_exception:Function]
|
||
|
||
|
||
# [DEF:_plan_dataset_review_intent:Function]
|
||
# @COMPLEXITY: 3
|
||
# @PURPOSE: Parse session-scoped dataset-review assistant commands before falling back to generic assistant tool routing.
|
||
def _plan_dataset_review_intent(
|
||
message: str,
|
||
dataset_context: Dict[str, Any],
|
||
) -> Optional[Dict[str, Any]]:
|
||
lower = message.strip().lower()
|
||
session_id = dataset_context["session_id"]
|
||
session_version = int(dataset_context.get("version", 0) or 0)
|
||
target_kind, target_id = _extract_dataset_review_target(message)
|
||
|
||
if any(
|
||
token in lower
|
||
for token in [
|
||
"approve mappings",
|
||
"approve mapping",
|
||
"подтверди мапп",
|
||
"одобри мапп",
|
||
]
|
||
):
|
||
pending_mapping_ids = [
|
||
item["mapping_id"]
|
||
for item in dataset_context.get("mappings", [])
|
||
if item.get("requires_explicit_approval")
|
||
and item.get("approval_state") != ApprovalState.APPROVED.value
|
||
]
|
||
return {
|
||
"domain": "dataset_review",
|
||
"operation": "dataset_review_approve_mappings",
|
||
"entities": {
|
||
"dataset_review_session_id": session_id,
|
||
"session_version": session_version,
|
||
"mapping_ids": pending_mapping_ids,
|
||
},
|
||
"confidence": 0.95,
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": True,
|
||
}
|
||
|
||
if any(
|
||
token in lower
|
||
for token in [
|
||
"generate sql preview",
|
||
"generate preview",
|
||
"сгенерируй превью",
|
||
"собери превью",
|
||
]
|
||
):
|
||
return {
|
||
"domain": "dataset_review",
|
||
"operation": "dataset_review_generate_sql_preview",
|
||
"entities": {
|
||
"dataset_review_session_id": session_id,
|
||
"session_version": session_version,
|
||
},
|
||
"confidence": 0.94,
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": True,
|
||
}
|
||
|
||
if any(
|
||
token in lower
|
||
for token in [
|
||
"set field semantics",
|
||
"apply field semantics",
|
||
"semantic override",
|
||
"update semantic field",
|
||
"установи семантик",
|
||
"обнови семантик",
|
||
]
|
||
):
|
||
field = _match_dataset_review_field(dataset_context, message)
|
||
if field is None:
|
||
return None
|
||
candidate_id = None
|
||
if any(
|
||
token in lower
|
||
for token in ["accept candidate", "apply candidate", "прими кандидат"]
|
||
):
|
||
candidates = field.get("candidates") or []
|
||
if candidates:
|
||
candidate_id = candidates[0].get("candidate_id")
|
||
verbose_name = _extract_quoted_segment(
|
||
message, "verbose_name|verbose name|label"
|
||
)
|
||
description = _extract_quoted_segment(message, "description|desc")
|
||
display_format = _extract_quoted_segment(
|
||
message, "display_format|display format|format"
|
||
)
|
||
lock_field = any(
|
||
token in lower for token in [" lock", "lock it", "зафикс", "закреп"]
|
||
)
|
||
if not any([candidate_id, verbose_name, description, display_format]):
|
||
return None
|
||
return {
|
||
"domain": "dataset_review",
|
||
"operation": "dataset_review_set_field_semantics",
|
||
"entities": {
|
||
"dataset_review_session_id": session_id,
|
||
"session_version": session_version,
|
||
"field_id": field.get("field_id") or target_id,
|
||
"candidate_id": candidate_id,
|
||
"verbose_name": verbose_name,
|
||
"description": description,
|
||
"display_format": display_format,
|
||
"lock_field": lock_field,
|
||
},
|
||
"confidence": 0.9,
|
||
"risk_level": "guarded",
|
||
"requires_confirmation": True,
|
||
}
|
||
|
||
if any(
|
||
token in lower
|
||
for token in [
|
||
"filters",
|
||
"фильтр",
|
||
"mapping",
|
||
"маппинг",
|
||
"preview",
|
||
"превью",
|
||
"finding",
|
||
"ошиб",
|
||
]
|
||
):
|
||
findings_count = len(dataset_context.get("findings", []))
|
||
mappings_count = len(dataset_context.get("mappings", []))
|
||
filters_count = len(dataset_context.get("imported_filters", []))
|
||
preview = dataset_context.get("preview") or {}
|
||
preview_status = preview.get("preview_status") or "missing"
|
||
masked_filters = dataset_context.get("imported_filters", [])
|
||
return {
|
||
"domain": "dataset_review",
|
||
"operation": "dataset_review_answer_context",
|
||
"entities": {
|
||
"dataset_review_session_id": session_id,
|
||
"summary": (
|
||
f"Session {session_id}: readiness={dataset_context['readiness_state']}, "
|
||
f"recommended_action={dataset_context['recommended_action']}, "
|
||
f"filters={filters_count}, mappings={mappings_count}, findings={findings_count}, "
|
||
f"preview_status={preview_status}, imported_filters={masked_filters}"
|
||
),
|
||
},
|
||
"confidence": 0.8,
|
||
"risk_level": "safe",
|
||
"requires_confirmation": False,
|
||
}
|
||
return None
|
||
|
||
|
||
# [/DEF:_plan_dataset_review_intent:Function]
|
||
|
||
|
||
# [DEF:_dispatch_dataset_review_intent:Function]
|
||
# @COMPLEXITY: 4
|
||
# @PURPOSE: Route confirmed dataset-review assistant intents through existing backend dataset-review APIs and orchestration boundaries.
|
||
async def _dispatch_dataset_review_intent(
|
||
intent: Dict[str, Any],
|
||
current_user: User,
|
||
config_manager: ConfigManager,
|
||
db: Session,
|
||
) -> Tuple[str, Optional[str], List[AssistantAction]]:
|
||
with belief_scope("_dispatch_dataset_review_intent"):
|
||
logger.reason(
|
||
"Dispatching assistant dataset-review intent",
|
||
extra={"operation": intent.get("operation")},
|
||
)
|
||
entities = intent.get("entities", {})
|
||
session_id = entities.get("dataset_review_session_id")
|
||
session_version = entities.get("session_version")
|
||
if not session_id or session_version is None:
|
||
raise HTTPException(
|
||
status_code=422,
|
||
detail="Missing dataset_review_session_id/session_version",
|
||
)
|
||
|
||
operation = str(intent.get("operation") or "")
|
||
repository = DatasetReviewSessionRepository(db)
|
||
if operation == "dataset_review_answer_context":
|
||
summary = str(entities.get("summary") or "")
|
||
logger.reflect(
|
||
"Returned assistant-safe dataset review context summary",
|
||
extra={"session_id": session_id, "operation": operation},
|
||
)
|
||
return summary, None, []
|
||
|
||
session = repository.load_session_detail(session_id, current_user.id)
|
||
if session is None or session.user_id != current_user.id:
|
||
logger.explore(
|
||
"Assistant dataset-review intent rejected because session was not found",
|
||
extra={"session_id": session_id, "user_id": current_user.id},
|
||
)
|
||
raise HTTPException(
|
||
status_code=404, detail="Dataset review session not found"
|
||
)
|
||
|
||
try:
|
||
repository.require_session_version(session, int(session_version))
|
||
except DatasetReviewSessionVersionConflictError as exc:
|
||
logger.explore(
|
||
"Assistant dataset-review intent rejected due to stale session version",
|
||
extra={
|
||
"session_id": exc.session_id,
|
||
"expected_version": exc.expected_version,
|
||
"actual_version": exc.actual_version,
|
||
"operation": operation,
|
||
},
|
||
)
|
||
raise _dataset_review_conflict_http_exception(exc) from exc
|
||
|
||
logger.reason(
|
||
"Dispatching confirmed assistant dataset-review intent",
|
||
extra={
|
||
"session_id": session_id,
|
||
"session_version": session_version,
|
||
"operation": operation,
|
||
},
|
||
)
|
||
|
||
if operation == "dataset_review_approve_mappings":
|
||
mapping_ids = list(dict.fromkeys(entities.get("mapping_ids") or []))
|
||
if not mapping_ids:
|
||
raise HTTPException(
|
||
status_code=409, detail="No pending mappings to approve"
|
||
)
|
||
updated_count = 0
|
||
for mapping in session.execution_mappings:
|
||
if mapping.mapping_id not in mapping_ids:
|
||
continue
|
||
mapping.approval_state = ApprovalState.APPROVED
|
||
mapping.approved_by_user_id = current_user.id
|
||
mapping.approved_at = datetime.utcnow()
|
||
updated_count += 1
|
||
if updated_count == 0:
|
||
raise HTTPException(
|
||
status_code=409, detail="No matching mappings available to approve"
|
||
)
|
||
session.last_activity_at = datetime.utcnow()
|
||
if session.readiness_state == ReadinessState.MAPPING_REVIEW_NEEDED:
|
||
session.recommended_action = RecommendedAction.GENERATE_SQL_PREVIEW
|
||
repository.bump_session_version(session)
|
||
repository.db.commit()
|
||
repository.db.refresh(session)
|
||
repository.event_logger.log_for_session(
|
||
session,
|
||
actor_user_id=current_user.id,
|
||
event_type="assistant_mapping_approval",
|
||
event_summary="Assistant-approved warning-sensitive mappings persisted",
|
||
event_details={
|
||
"mapping_ids": mapping_ids,
|
||
"count": updated_count,
|
||
"version": int(getattr(session, "version", 0) or 0),
|
||
},
|
||
)
|
||
logger.reflect(
|
||
"Assistant mapping approval persisted within optimistic-lock boundary",
|
||
extra={
|
||
"session_id": session_id,
|
||
"updated_count": updated_count,
|
||
"version": int(getattr(session, "version", 0) or 0),
|
||
},
|
||
)
|
||
return (
|
||
f"Approved {updated_count} mapping(s) for dataset review session {session_id}.",
|
||
None,
|
||
[
|
||
AssistantAction(
|
||
type="focus_target",
|
||
label="Open mapping review",
|
||
target="mapping",
|
||
)
|
||
],
|
||
)
|
||
|
||
if operation == "dataset_review_set_field_semantics":
|
||
field_id = str(entities.get("field_id") or "").strip()
|
||
if not field_id:
|
||
raise HTTPException(status_code=422, detail="Missing field_id")
|
||
field = next(
|
||
(item for item in session.semantic_fields if item.field_id == field_id),
|
||
None,
|
||
)
|
||
if field is None:
|
||
raise HTTPException(status_code=404, detail="Semantic field not found")
|
||
update_request = FieldSemanticUpdateRequest(
|
||
candidate_id=entities.get("candidate_id"),
|
||
verbose_name=entities.get("verbose_name"),
|
||
description=entities.get("description"),
|
||
display_format=entities.get("display_format"),
|
||
lock_field=bool(entities.get("lock_field", False)),
|
||
)
|
||
try:
|
||
_update_semantic_field_state(
|
||
field, update_request, changed_by="assistant"
|
||
)
|
||
except HTTPException:
|
||
raise
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=400, detail=str(exc)) from exc
|
||
session.last_activity_at = datetime.utcnow()
|
||
repository.bump_session_version(session)
|
||
repository.db.commit()
|
||
repository.db.refresh(session)
|
||
repository.db.refresh(field)
|
||
repository.event_logger.log_for_session(
|
||
session,
|
||
actor_user_id=current_user.id,
|
||
event_type="assistant_field_semantics_updated",
|
||
event_summary="Assistant semantic field update persisted",
|
||
event_details={
|
||
"field_id": field.field_id,
|
||
"candidate_id": entities.get("candidate_id"),
|
||
"lock_field": bool(entities.get("lock_field", False)),
|
||
"version": int(getattr(session, "version", 0) or 0),
|
||
},
|
||
)
|
||
logger.reflect(
|
||
"Assistant semantic field update committed safely",
|
||
extra={
|
||
"session_id": session_id,
|
||
"field_id": field_id,
|
||
"version": int(getattr(session, "version", 0) or 0),
|
||
},
|
||
)
|
||
return (
|
||
f"Updated semantic field {field.field_name} for dataset review session {session_id}.",
|
||
None,
|
||
[
|
||
AssistantAction(
|
||
type="focus_target",
|
||
label="Open semantic review",
|
||
target=f"field:{field.field_id}",
|
||
)
|
||
],
|
||
)
|
||
|
||
if operation == "dataset_review_generate_sql_preview":
|
||
orchestrator = DatasetReviewOrchestrator(
|
||
repository=repository,
|
||
config_manager=config_manager,
|
||
)
|
||
result = orchestrator.prepare_launch_preview(
|
||
PreparePreviewCommand(
|
||
user=current_user,
|
||
session_id=session_id,
|
||
expected_version=int(session_version),
|
||
)
|
||
)
|
||
preview_status = getattr(
|
||
result.preview.preview_status, "value", result.preview.preview_status
|
||
)
|
||
logger.reflect(
|
||
"Assistant-triggered Superset preview generation completed",
|
||
extra={
|
||
"session_id": session_id,
|
||
"preview_status": preview_status,
|
||
},
|
||
)
|
||
return (
|
||
f"SQL preview {preview_status} for dataset review session {session_id}.",
|
||
None,
|
||
[
|
||
AssistantAction(
|
||
type="focus_target",
|
||
label="Open SQL preview",
|
||
target="sql-preview",
|
||
)
|
||
],
|
||
)
|
||
|
||
raise HTTPException(
|
||
status_code=400, detail="Unsupported dataset review operation"
|
||
)
|
||
|
||
|
||
# [/DEF:_dispatch_dataset_review_intent:Function]
|
||
|
||
|
||
# [DEF:_confirmation_summary:Function]
|
||
# @COMPLEXITY: 4
|
||
# @PURPOSE: Build human-readable confirmation prompt for an intent before execution.
|
||
# @PRE: intent contains operation and entities fields.
|
||
# @POST: Returns descriptive Russian-language text ending with confirmation prompt.
|
||
async def _async_confirmation_summary(intent: Dict[str, Any], config_manager: ConfigManager, db: Session) -> str:
|
||
with belief_scope('_confirmation_summary'):
|
||
logger.reason('Belief protocol reasoning checkpoint for _confirmation_summary')
|
||
operation = intent.get('operation', '')
|
||
entities = intent.get('entities', {})
|
||
descriptions: Dict[str, str] = {'create_branch': 'создание ветки{branch} для дашборда{dashboard}', 'commit_changes': 'коммит изменений для дашборда{dashboard}', 'deploy_dashboard': 'деплой дашборда{dashboard} в окружение{env}', 'execute_migration': 'миграция дашборда{dashboard} с{src} на{tgt}', 'run_backup': 'бэкап окружения{env}{dashboard}', 'run_llm_validation': 'LLM-валидация дашборда{dashboard}{env}', 'run_llm_documentation': 'генерация документации для датасета{dataset}{env}'}
|
||
template = descriptions.get(operation)
|
||
if not template:
|
||
logger.reflect('Belief protocol postcondition checkpoint for _confirmation_summary')
|
||
return 'Подтвердите выполнение операции или отмените.'
|
||
|
||
def _label(value: Any, prefix: str=' ') -> str:
|
||
logger.reflect('Belief protocol postcondition checkpoint for _confirmation_summary')
|
||
return f'{prefix}{value}' if value else ''
|
||
dashboard = entities.get('dashboard_id') or entities.get('dashboard_ref')
|
||
text = template.format(branch=_label(entities.get('branch_name')), dashboard=_label(dashboard), env=_label(entities.get('environment') or entities.get('target_env')), src=_label(entities.get('source_env')), tgt=_label(entities.get('target_env')), dataset=_label(entities.get('dataset_id')))
|
||
if operation == 'execute_migration':
|
||
flags = []
|
||
flags.append('маппинг БД: ' + ('ВКЛ' if _coerce_query_bool(entities.get('replace_db_config', False)) else 'ВЫКЛ'))
|
||
flags.append('исправление кроссфильтров: ' + ('ВКЛ' if _coerce_query_bool(entities.get('fix_cross_filters', True)) else 'ВЫКЛ'))
|
||
dry_run_enabled = _coerce_query_bool(entities.get('dry_run', False))
|
||
flags.append('отчет dry-run: ' + ('ВКЛ' if dry_run_enabled else 'ВЫКЛ'))
|
||
text += f" ({', '.join(flags)})"
|
||
if dry_run_enabled:
|
||
try:
|
||
from ...core.migration.dry_run_orchestrator import MigrationDryRunService
|
||
from ...models.dashboard import DashboardSelection
|
||
from ...core.superset_client import SupersetClient
|
||
src_token = entities.get('source_env')
|
||
tgt_token = entities.get('target_env')
|
||
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=src_token)
|
||
if dashboard_id and src_token and tgt_token:
|
||
src_env_id = _resolve_env_id(src_token, config_manager)
|
||
tgt_env_id = _resolve_env_id(tgt_token, config_manager)
|
||
if src_env_id and tgt_env_id:
|
||
env_map = {env.id: env for env in config_manager.get_environments()}
|
||
source_env = env_map.get(src_env_id)
|
||
target_env = env_map.get(tgt_env_id)
|
||
if source_env and target_env and (source_env.id != target_env.id):
|
||
selection = DashboardSelection(source_env_id=source_env.id, target_env_id=target_env.id, selected_ids=[dashboard_id], replace_db_config=_coerce_query_bool(entities.get('replace_db_config', False)), fix_cross_filters=_coerce_query_bool(entities.get('fix_cross_filters', True)))
|
||
service = MigrationDryRunService()
|
||
source_client = SupersetClient(source_env)
|
||
target_client = SupersetClient(target_env)
|
||
report = service.run(selection, source_client, target_client, db)
|
||
s = report.get('summary', {})
|
||
dash_s = s.get('dashboards', {})
|
||
charts_s = s.get('charts', {})
|
||
ds_s = s.get('datasets', {})
|
||
creates = dash_s.get('create', 0) + charts_s.get('create', 0) + ds_s.get('create', 0)
|
||
updates = dash_s.get('update', 0) + charts_s.get('update', 0) + ds_s.get('update', 0)
|
||
deletes = dash_s.get('delete', 0) + charts_s.get('delete', 0) + ds_s.get('delete', 0)
|
||
text += f'\n\nОтчет dry-run:\n- Будет создано новых объектов: {creates}\n- Будет обновлено: {updates}\n- Будет удалено: {deletes}'
|
||
else:
|
||
text += '\n\n(Не удалось загрузить отчет dry-run: неверные окружения).'
|
||
except Exception as e:
|
||
import traceback
|
||
logger.warning('[assistant.dry_run_summary][failed] Exception: %s\n%s', e, traceback.format_exc())
|
||
text += f'\n\n(Не удалось загрузить отчет dry-run: {e}).'
|
||
logger.reflect('Belief protocol postcondition checkpoint for _confirmation_summary')
|
||
return f'Выполнить: {text}. Подтвердите или отмените.'
|
||
# [/DEF:_confirmation_summary:Function]
|
||
|
||
|
||
# [DEF:_clarification_text_for_intent:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Convert technical missing-parameter errors into user-facing clarification prompts.
|
||
# @PRE: state was classified as needs_clarification for current intent/error combination.
|
||
# @POST: Returned text is human-readable and actionable for target operation.
|
||
def _clarification_text_for_intent(
|
||
intent: Optional[Dict[str, Any]], detail_text: str
|
||
) -> str:
|
||
operation = (intent or {}).get("operation")
|
||
guidance_by_operation: Dict[str, str] = {
|
||
"run_llm_validation": (
|
||
"Нужно уточнение для запуска LLM-валидации: Укажите дашборд (id или slug), окружение и провайдер LLM."
|
||
),
|
||
"run_llm_documentation": (
|
||
"Нужно уточнение для генерации документации: Укажите dataset_id, окружение и провайдер LLM."
|
||
),
|
||
"create_branch": "Нужно уточнение: укажите дашборд (id/slug/title) и имя ветки.",
|
||
"commit_changes": "Нужно уточнение: укажите дашборд (id/slug/title) для коммита.",
|
||
"deploy_dashboard": "Нужно уточнение: укажите дашборд (id/slug/title) и целевое окружение.",
|
||
"execute_migration": "Нужно уточнение: укажите дашборд (id/slug/title), source_env и target_env.",
|
||
"run_backup": "Нужно уточнение: укажите окружение и при необходимости дашборд (id/slug/title).",
|
||
}
|
||
return guidance_by_operation.get(operation, detail_text)
|
||
|
||
|
||
# [/DEF:_clarification_text_for_intent:Function]
|
||
|
||
|
||
# [DEF:_plan_intent_with_llm:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Use active LLM provider to select best tool/operation from dynamic catalog.
|
||
# @PRE: tools list contains allowed operations for current user.
|
||
# @POST: Returns normalized intent dict when planning succeeds; otherwise None.
|
||
async def _plan_intent_with_llm(
|
||
message: str,
|
||
tools: List[Dict[str, Any]],
|
||
db: Session,
|
||
config_manager: ConfigManager,
|
||
) -> Optional[Dict[str, Any]]:
|
||
if not tools:
|
||
return None
|
||
|
||
llm_settings = normalize_llm_settings(config_manager.get_config().settings.llm)
|
||
planner_provider_token = llm_settings.get("assistant_planner_provider")
|
||
planner_model_override = llm_settings.get("assistant_planner_model")
|
||
llm_service = LLMProviderService(db)
|
||
providers = llm_service.get_all_providers()
|
||
provider_id = _resolve_provider_id(planner_provider_token, db)
|
||
provider = next((p for p in providers if p.id == provider_id), None)
|
||
if not provider:
|
||
return None
|
||
api_key = llm_service.get_decrypted_api_key(provider.id)
|
||
if not api_key:
|
||
return None
|
||
|
||
planner = LLMClient(
|
||
provider_type=LLMProviderType(provider.provider_type),
|
||
api_key=api_key,
|
||
base_url=provider.base_url,
|
||
default_model=planner_model_override or provider.default_model,
|
||
)
|
||
|
||
system_instruction = (
|
||
"You are a deterministic intent planner for backend tools.\n"
|
||
"Choose exactly one operation from available_tools or return clarify.\n"
|
||
"Output strict JSON object:\n"
|
||
"{"
|
||
'"domain": string, '
|
||
'"operation": string, '
|
||
'"entities": object, '
|
||
'"confidence": number, '
|
||
'"risk_level": "safe"|"guarded"|"dangerous", '
|
||
'"requires_confirmation": boolean'
|
||
"}\n"
|
||
"Rules:\n"
|
||
"- Use only operation names from available_tools.\n"
|
||
'- If input is ambiguous, operation must be "clarify" with low confidence.\n'
|
||
"- If dashboard is provided as name/slug (e.g., COVID), put it into entities.dashboard_ref.\n"
|
||
"- Keep entities minimal and factual.\n"
|
||
)
|
||
payload = {
|
||
"available_tools": tools,
|
||
"user_message": message,
|
||
"known_environments": [
|
||
{"id": e.id, "name": e.name} for e in config_manager.get_environments()
|
||
],
|
||
}
|
||
try:
|
||
response = await planner.get_json_completion(
|
||
[
|
||
{"role": "system", "content": system_instruction},
|
||
{"role": "user", "content": json.dumps(payload, ensure_ascii=False)},
|
||
]
|
||
)
|
||
except Exception as exc:
|
||
import traceback
|
||
|
||
logger.warning(
|
||
f"[assistant.planner][fallback] LLM planner unavailable: {exc}\n{traceback.format_exc()}"
|
||
)
|
||
return None
|
||
if not isinstance(response, dict):
|
||
return None
|
||
|
||
operation = response.get("operation")
|
||
valid_ops = {tool["operation"] for tool in tools}
|
||
if operation == "clarify":
|
||
return {
|
||
"domain": "unknown",
|
||
"operation": "clarify",
|
||
"entities": {},
|
||
"confidence": float(response.get("confidence", 0.3)),
|
||
"risk_level": "safe",
|
||
"requires_confirmation": False,
|
||
}
|
||
if operation not in valid_ops:
|
||
return None
|
||
|
||
by_operation = {tool["operation"]: tool for tool in tools}
|
||
selected = by_operation[operation]
|
||
intent = {
|
||
"domain": response.get("domain") or selected["domain"],
|
||
"operation": operation,
|
||
"entities": response.get("entities", {}),
|
||
"confidence": float(response.get("confidence", 0.75)),
|
||
"risk_level": response.get("risk_level") or selected["risk_level"],
|
||
"requires_confirmation": bool(
|
||
response.get("requires_confirmation", selected["requires_confirmation"])
|
||
),
|
||
}
|
||
intent = _coerce_intent_entities(intent)
|
||
|
||
defaults = selected.get("defaults") or {}
|
||
for key, value in defaults.items():
|
||
if value and not intent["entities"].get(key):
|
||
intent["entities"][key] = value
|
||
|
||
if operation in {"deploy_dashboard", "execute_migration"}:
|
||
env_token = intent["entities"].get("environment") or intent["entities"].get(
|
||
"target_env"
|
||
)
|
||
if _is_production_env(env_token, config_manager):
|
||
intent["risk_level"] = "dangerous"
|
||
intent["requires_confirmation"] = True
|
||
return intent
|
||
|
||
|
||
# [/DEF:_plan_intent_with_llm:Function]
|
||
|
||
|
||
# [DEF:_authorize_intent:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Validate user permissions for parsed intent before confirmation/dispatch.
|
||
# @PRE: intent.operation is present for known assistant command domains.
|
||
# @POST: Returns if authorized; raises HTTPException(403) when denied.
|
||
def _authorize_intent(intent: Dict[str, Any], current_user: User):
|
||
operation = intent.get("operation")
|
||
if operation in INTENT_PERMISSION_CHECKS:
|
||
_check_any_permission(current_user, INTENT_PERMISSION_CHECKS[operation])
|
||
|
||
|
||
# [/DEF:_authorize_intent:Function]
|
||
|
||
|
||
# [DEF:_dispatch_intent:Function]
|
||
# @COMPLEXITY: 5
|
||
# @PURPOSE: Execute parsed assistant intent via existing task/plugin/git services.
|
||
# @DATA_CONTRACT: Input[intent,current_user,task_manager,config_manager,db] -> Output[Tuple[text:str,task_id:Optional[str],actions:List[AssistantAction]]]
|
||
# @RELATION: DEPENDS_ON -> [_check_any_permission]
|
||
# @RELATION: DEPENDS_ON -> [_resolve_dashboard_id_entity]
|
||
# @RELATION: DEPENDS_ON -> [TaskManager]
|
||
# @RELATION: DEPENDS_ON -> [GitService]
|
||
# @SIDE_EFFECT: May enqueue tasks, invoke git operations, and query/update external service state.
|
||
# @PRE: intent operation is known and actor permissions are validated per operation.
|
||
# @POST: Returns response text, optional task id, and UI actions for follow-up.
|
||
# @INVARIANT: unsupported operations are rejected via HTTPException(400).
|
||
async def _dispatch_intent(intent: Dict[str, Any], current_user: User, task_manager: TaskManager, config_manager: ConfigManager, db: Session) -> Tuple[str, Optional[str], List[AssistantAction]]:
|
||
with belief_scope('_dispatch_intent'):
|
||
logger.reason('Belief protocol reasoning checkpoint for _dispatch_intent')
|
||
operation = intent.get('operation')
|
||
entities = intent.get('entities', {})
|
||
if operation in _DATASET_REVIEW_OPS or operation == 'dataset_review_answer_context':
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return await _dispatch_dataset_review_intent(intent, current_user, config_manager, db)
|
||
if operation == 'show_capabilities':
|
||
tools_catalog = _build_tool_catalog(current_user, config_manager, db)
|
||
labels = {'create_branch': 'Git: создание ветки', 'commit_changes': 'Git: коммит', 'deploy_dashboard': 'Git: деплой дашборда', 'execute_migration': 'Миграции: запуск переноса', 'run_backup': 'Бэкапы: запуск резервного копирования', 'run_llm_validation': 'LLM: валидация дашборда', 'run_llm_documentation': 'LLM: генерация документации', 'get_task_status': 'Статус: проверка задачи', 'get_health_summary': 'Здоровье: сводка по дашбордам'}
|
||
available = [labels[t['operation']] for t in tools_catalog if t['operation'] in labels]
|
||
if not available:
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return ('Сейчас нет доступных для вас операций ассистента.', None, [])
|
||
commands = '\n'.join((f'- {item}' for item in available))
|
||
text = f'Вот что я могу сделать для вас:\n{commands}\n\nПример: `запусти миграцию с dev на prod для дашборда 42`.'
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (text, None, [])
|
||
if operation == 'get_health_summary':
|
||
from ...services.health_service import HealthService
|
||
env_token = entities.get('environment')
|
||
env_id = _resolve_env_id(env_token, config_manager)
|
||
service = HealthService(db)
|
||
summary = await service.get_health_summary(environment_id=env_id)
|
||
env_name = _get_environment_name_by_id(env_id, config_manager) if env_id else 'всех окружений'
|
||
text = f'Сводка здоровья дашбордов для {env_name}:\n- ✅ Прошли проверку: {summary.pass_count}\n- ⚠️ С предупреждениями: {summary.warn_count}\n- ❌ Ошибки валидации: {summary.fail_count}\n- ❓ Неизвестно: {summary.unknown_count}'
|
||
actions = [AssistantAction(type='open_route', label='Открыть Health Center', target='/dashboards/health')]
|
||
if summary.fail_count > 0:
|
||
text += '\n\nОбнаружены ошибки в следующих дашбордах:'
|
||
for item in summary.items:
|
||
if item.status == 'FAIL':
|
||
text += f"\n- {item.dashboard_id} ({item.environment_id}): {item.summary or 'Нет деталей'}"
|
||
actions.append(AssistantAction(type='open_route', label=f'Отчет {item.dashboard_id}', target=f'/reports/llm/{item.task_id}'))
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (text, None, actions[:5])
|
||
if operation == 'get_task_status':
|
||
_check_any_permission(current_user, [('tasks', 'READ')])
|
||
task_id = entities.get('task_id')
|
||
if not task_id:
|
||
recent = [t for t in task_manager.get_tasks(limit=20, offset=0) if t.user_id == current_user.id]
|
||
if not recent:
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return ('У вас пока нет задач в истории.', None, [])
|
||
task = recent[0]
|
||
actions = [AssistantAction(type='open_task', label='Open Task', target=task.id)]
|
||
if str(task.status).upper() in {'SUCCESS', 'FAILED'}:
|
||
actions.extend(_extract_result_deep_links(task, config_manager))
|
||
summary_line = _build_task_observability_summary(task, config_manager)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'Последняя задача: {task.id}, статус: {task.status}.' + (f'\n{summary_line}' if summary_line else ''), task.id, actions)
|
||
task = task_manager.get_task(task_id)
|
||
if not task:
|
||
raise HTTPException(status_code=404, detail=f'Task {task_id} not found')
|
||
actions = [AssistantAction(type='open_task', label='Open Task', target=task.id)]
|
||
if str(task.status).upper() in {'SUCCESS', 'FAILED'}:
|
||
actions.extend(_extract_result_deep_links(task, config_manager))
|
||
summary_line = _build_task_observability_summary(task, config_manager)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'Статус задачи {task.id}: {task.status}.' + (f'\n{summary_line}' if summary_line else ''), task.id, actions)
|
||
if operation == 'create_branch':
|
||
_check_any_permission(current_user, [('plugin:git', 'EXECUTE')])
|
||
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager)
|
||
branch_name = entities.get('branch_name')
|
||
if not dashboard_id or not branch_name:
|
||
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref or branch_name')
|
||
git_service.create_branch(dashboard_id, branch_name, 'main')
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'Ветка `{branch_name}` создана для дашборда {dashboard_id}.', None, [])
|
||
if operation == 'commit_changes':
|
||
_check_any_permission(current_user, [('plugin:git', 'EXECUTE')])
|
||
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager)
|
||
commit_message = entities.get('message')
|
||
if not dashboard_id:
|
||
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref')
|
||
git_service.commit_changes(dashboard_id, commit_message, None)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return ('Коммит выполнен успешно.', None, [])
|
||
if operation == 'deploy_dashboard':
|
||
_check_any_permission(current_user, [('plugin:git', 'EXECUTE')])
|
||
env_token = entities.get('environment')
|
||
env_id = _resolve_env_id(env_token, config_manager)
|
||
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=env_token)
|
||
if not dashboard_id or not env_id:
|
||
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref or environment')
|
||
task = await task_manager.create_task(plugin_id='git-integration', params={'operation': 'deploy', 'dashboard_id': dashboard_id, 'environment_id': env_id}, user_id=current_user.id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'Деплой запущен. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports')])
|
||
if operation == 'execute_migration':
|
||
_check_any_permission(current_user, [('plugin:migration', 'EXECUTE'), ('plugin:superset-migration', 'EXECUTE')])
|
||
src_token = entities.get('source_env')
|
||
dashboard_ref = entities.get('dashboard_ref')
|
||
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=src_token)
|
||
src = _resolve_env_id(src_token, config_manager)
|
||
tgt = _resolve_env_id(entities.get('target_env'), config_manager)
|
||
if not src or not tgt:
|
||
raise HTTPException(status_code=422, detail='Missing source_env/target_env')
|
||
if not dashboard_id and (not dashboard_ref):
|
||
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref')
|
||
migration_params: Dict[str, Any] = {'source_env_id': src, 'target_env_id': tgt, 'replace_db_config': _coerce_query_bool(entities.get('replace_db_config', False)), 'fix_cross_filters': _coerce_query_bool(entities.get('fix_cross_filters', True))}
|
||
if dashboard_id:
|
||
migration_params['selected_ids'] = [dashboard_id]
|
||
else:
|
||
migration_params['dashboard_regex'] = str(dashboard_ref)
|
||
task = await task_manager.create_task(plugin_id='superset-migration', params=migration_params, user_id=current_user.id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'Миграция запущена. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports'), *([AssistantAction(type='open_route', label=f'Открыть дашборд в {_get_environment_name_by_id(tgt, config_manager)}', target=f'/dashboards/{dashboard_id}?env_id={tgt}'), AssistantAction(type='open_diff', label='Показать Diff', target=str(dashboard_id))] if dashboard_id else [])])
|
||
if operation == 'run_backup':
|
||
_check_any_permission(current_user, [('plugin:superset-backup', 'EXECUTE'), ('plugin:backup', 'EXECUTE')])
|
||
env_token = entities.get('environment')
|
||
env_id = _resolve_env_id(env_token, config_manager)
|
||
if not env_id:
|
||
raise HTTPException(status_code=400, detail='Missing or unknown environment')
|
||
params: Dict[str, Any] = {'environment_id': env_id}
|
||
if entities.get('dashboard_id') or entities.get('dashboard_ref'):
|
||
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=env_token)
|
||
if not dashboard_id:
|
||
raise HTTPException(status_code=422, detail='Missing dashboard_id/dashboard_ref')
|
||
params['dashboard_ids'] = [dashboard_id]
|
||
task = await task_manager.create_task(plugin_id='superset-backup', params=params, user_id=current_user.id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'Бэкап запущен. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports'), *([AssistantAction(type='open_route', label=f'Открыть дашборд в {_get_environment_name_by_id(env_id, config_manager)}', target=f'/dashboards/{dashboard_id}?env_id={env_id}'), AssistantAction(type='open_diff', label='Показать Diff', target=str(dashboard_id))] if entities.get('dashboard_id') or entities.get('dashboard_ref') else [])])
|
||
if operation == 'run_llm_validation':
|
||
_check_any_permission(current_user, [('plugin:llm_dashboard_validation', 'EXECUTE')])
|
||
env_token = entities.get('environment')
|
||
env_id = _resolve_env_id(env_token, config_manager) or _get_default_environment_id(config_manager)
|
||
dashboard_id = _resolve_dashboard_id_entity(entities, config_manager, env_hint=env_token)
|
||
provider_id = _resolve_provider_id(entities.get('provider'), db, config_manager=config_manager, task_key='dashboard_validation')
|
||
if not dashboard_id or not env_id or (not provider_id):
|
||
raise HTTPException(status_code=422, detail='Missing dashboard_id/environment/provider. Укажите ID/slug дашборда или окружение.')
|
||
provider = LLMProviderService(db).get_provider(provider_id)
|
||
provider_model = provider.default_model if provider else ''
|
||
if not is_multimodal_model(provider_model, provider.provider_type if provider else None):
|
||
raise HTTPException(status_code=422, detail='Selected provider model is not multimodal for dashboard validation. Выберите мультимодальную модель (например, gpt-4o).')
|
||
task = await task_manager.create_task(plugin_id='llm_dashboard_validation', params={'dashboard_id': str(dashboard_id), 'environment_id': env_id, 'provider_id': provider_id}, user_id=current_user.id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'LLM-валидация запущена. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports')])
|
||
if operation == 'run_llm_documentation':
|
||
_check_any_permission(current_user, [('plugin:llm_documentation', 'EXECUTE')])
|
||
dataset_id = entities.get('dataset_id')
|
||
env_id = _resolve_env_id(entities.get('environment'), config_manager)
|
||
provider_id = _resolve_provider_id(entities.get('provider'), db, config_manager=config_manager, task_key='documentation')
|
||
if not dataset_id or not env_id or (not provider_id):
|
||
raise HTTPException(status_code=400, detail='Missing dataset_id/environment/provider')
|
||
task = await task_manager.create_task(plugin_id='llm_documentation', params={'dataset_id': str(dataset_id), 'environment_id': env_id, 'provider_id': provider_id}, user_id=current_user.id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for _dispatch_intent')
|
||
return (f'Генерация документации запущена. task_id={task.id}', task.id, [AssistantAction(type='open_task', label='Open Task', target=task.id), AssistantAction(type='open_reports', label='Open Reports', target='/reports')])
|
||
raise HTTPException(status_code=400, detail='Unsupported operation')
|
||
# [/DEF:_dispatch_intent:Function]
|
||
|
||
|
||
@router.post("/messages", response_model=AssistantMessageResponse)
|
||
# [DEF:send_message:Function]
|
||
# @COMPLEXITY: 5
|
||
# @PURPOSE: Parse assistant command, enforce safety gates, and dispatch executable intent.
|
||
# @DATA_CONTRACT: Input[AssistantMessageRequest,User,TaskManager,ConfigManager,Session] -> Output[AssistantMessageResponse]
|
||
# @RELATION: DEPENDS_ON -> [_plan_intent_with_llm]
|
||
# @RELATION: DEPENDS_ON -> [_parse_command]
|
||
# @RELATION: DEPENDS_ON -> [_dispatch_intent]
|
||
# @RELATION: DEPENDS_ON -> [_append_history]
|
||
# @RELATION: DEPENDS_ON -> [_persist_message]
|
||
# @RELATION: DEPENDS_ON -> [_audit]
|
||
# @SIDE_EFFECT: Persists chat/audit state, mutates in-memory conversation and confirmation stores, and may create confirmation records.
|
||
# @PRE: Authenticated user is available and message text is non-empty.
|
||
# @POST: Response state is one of clarification/confirmation/started/success/denied/failed.
|
||
# @RETURN: AssistantMessageResponse with operation feedback and optional actions.
|
||
# @INVARIANT: non-safe operations are gated with confirmation before execution from this endpoint.
|
||
async def send_message(request: AssistantMessageRequest, current_user: User=Depends(get_current_user), task_manager: TaskManager=Depends(get_task_manager), config_manager: ConfigManager=Depends(get_config_manager), db: Session=Depends(get_db)):
|
||
with belief_scope('send_message'):
|
||
logger.reason('Belief protocol reasoning checkpoint for send_message')
|
||
user_id = current_user.id
|
||
dataset_review_context = _load_dataset_review_context(request.dataset_review_session_id, current_user, db)
|
||
conversation_id = _resolve_or_create_conversation(user_id, request.conversation_id, db)
|
||
_append_history(user_id, conversation_id, 'user', request.message)
|
||
_persist_message(db, user_id, conversation_id, 'user', request.message)
|
||
tools_catalog = _build_tool_catalog(current_user, config_manager, db)
|
||
intent = None
|
||
try:
|
||
intent = await _plan_intent_with_llm(request.message, tools_catalog, db, config_manager)
|
||
except Exception as exc:
|
||
logger.warning(f'[assistant.planner][fallback] Planner error: {exc}')
|
||
if not intent:
|
||
intent = _parse_command(request.message, config_manager)
|
||
if dataset_review_context:
|
||
dataset_review_intent = _plan_dataset_review_intent(request.message, dataset_review_context)
|
||
if dataset_review_intent is not None:
|
||
intent = dataset_review_intent
|
||
confidence = float(intent.get('confidence', 0.0))
|
||
if intent.get('domain') == 'unknown' or confidence < 0.6:
|
||
text = 'Команда неоднозначна. Уточните действие: git / migration / backup / llm / status.'
|
||
_append_history(user_id, conversation_id, 'assistant', text, state='needs_clarification')
|
||
_persist_message(db, user_id, conversation_id, 'assistant', text, state='needs_clarification', metadata={'intent': intent})
|
||
audit_payload = {'decision': 'needs_clarification', 'message': request.message, 'intent': intent, 'dataset_review_session_id': request.dataset_review_session_id}
|
||
_audit(user_id, audit_payload)
|
||
_persist_audit(db, user_id, audit_payload, conversation_id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for send_message')
|
||
return AssistantMessageResponse(conversation_id=conversation_id, response_id=str(uuid.uuid4()), state='needs_clarification', text=text, intent=intent, actions=[AssistantAction(type='rephrase', label='Rephrase command')], created_at=datetime.utcnow())
|
||
try:
|
||
_authorize_intent(intent, current_user)
|
||
operation = intent.get('operation')
|
||
if operation not in _SAFE_OPS:
|
||
confirmation_id = str(uuid.uuid4())
|
||
confirm = ConfirmationRecord(id=confirmation_id, user_id=user_id, conversation_id=conversation_id, intent=intent, dispatch={'intent': intent}, expires_at=datetime.utcnow() + timedelta(minutes=5), created_at=datetime.utcnow())
|
||
CONFIRMATIONS[confirmation_id] = confirm
|
||
_persist_confirmation(db, confirm)
|
||
text = await _async_confirmation_summary(intent, config_manager, db)
|
||
_append_history(user_id, conversation_id, 'assistant', text, state='needs_confirmation', confirmation_id=confirmation_id)
|
||
_persist_message(db, user_id, conversation_id, 'assistant', text, state='needs_confirmation', confirmation_id=confirmation_id, metadata={'intent': intent, 'dataset_review_context': dataset_review_context, 'actions': [{'type': 'confirm', 'label': '✅ Подтвердить', 'target': confirmation_id}, {'type': 'cancel', 'label': '❌ Отменить', 'target': confirmation_id}]})
|
||
audit_payload = {'decision': 'needs_confirmation', 'message': request.message, 'intent': intent, 'confirmation_id': confirmation_id, 'dataset_review_session_id': request.dataset_review_session_id}
|
||
_audit(user_id, audit_payload)
|
||
_persist_audit(db, user_id, audit_payload, conversation_id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for send_message')
|
||
return AssistantMessageResponse(conversation_id=conversation_id, response_id=str(uuid.uuid4()), state='needs_confirmation', text=text, intent=intent, confirmation_id=confirmation_id, actions=[AssistantAction(type='confirm', label='✅ Подтвердить', target=confirmation_id), AssistantAction(type='cancel', label='❌ Отменить', target=confirmation_id)], created_at=datetime.utcnow())
|
||
text, task_id, actions = await _dispatch_intent(intent, current_user, task_manager, config_manager, db)
|
||
state = 'started' if task_id else 'success'
|
||
_append_history(user_id, conversation_id, 'assistant', text, state=state, task_id=task_id)
|
||
_persist_message(db, user_id, conversation_id, 'assistant', text, state=state, task_id=task_id, metadata={'intent': intent, 'dataset_review_context': dataset_review_context, 'actions': [a.model_dump() for a in actions]})
|
||
audit_payload = {'decision': 'executed', 'message': request.message, 'intent': intent, 'task_id': task_id, 'dataset_review_session_id': request.dataset_review_session_id}
|
||
_audit(user_id, audit_payload)
|
||
_persist_audit(db, user_id, audit_payload, conversation_id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for send_message')
|
||
return AssistantMessageResponse(conversation_id=conversation_id, response_id=str(uuid.uuid4()), state=state, text=text, intent=intent, task_id=task_id, actions=actions, created_at=datetime.utcnow())
|
||
except HTTPException as exc:
|
||
detail_text = str(exc.detail)
|
||
is_clarification_error = exc.status_code in (400, 422) and (detail_text.lower().startswith('missing') or 'укажите' in detail_text.lower() or 'выберите' in detail_text.lower())
|
||
if exc.status_code == status.HTTP_403_FORBIDDEN:
|
||
state = 'denied'
|
||
elif is_clarification_error:
|
||
state = 'needs_clarification'
|
||
else:
|
||
state = 'failed'
|
||
text = _clarification_text_for_intent(intent, detail_text) if state == 'needs_clarification' else detail_text
|
||
_append_history(user_id, conversation_id, 'assistant', text, state=state)
|
||
_persist_message(db, user_id, conversation_id, 'assistant', text, state=state, metadata={'intent': intent})
|
||
audit_payload = {'decision': state, 'message': request.message, 'intent': intent, 'error': text, 'dataset_review_session_id': request.dataset_review_session_id}
|
||
_audit(user_id, audit_payload)
|
||
_persist_audit(db, user_id, audit_payload, conversation_id)
|
||
logger.reflect('Belief protocol postcondition checkpoint for send_message')
|
||
return AssistantMessageResponse(conversation_id=conversation_id, response_id=str(uuid.uuid4()), state=state, text=text, intent=intent, actions=[AssistantAction(type='rephrase', label='Rephrase command')] if state == 'needs_clarification' else [], created_at=datetime.utcnow())
|
||
# [/DEF:send_message:Function]
|
||
|
||
|
||
@router.post(
|
||
"/confirmations/{confirmation_id}/confirm", response_model=AssistantMessageResponse
|
||
)
|
||
# [DEF:confirm_operation:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Execute previously requested risky operation after explicit user confirmation.
|
||
# @PRE: confirmation_id exists, belongs to current user, is pending, and not expired.
|
||
# @POST: Confirmation state becomes consumed and operation result is persisted in history.
|
||
# @RETURN: AssistantMessageResponse with task details when async execution starts.
|
||
async def confirm_operation(
|
||
confirmation_id: str,
|
||
current_user: User = Depends(get_current_user),
|
||
task_manager: TaskManager = Depends(get_task_manager),
|
||
config_manager: ConfigManager = Depends(get_config_manager),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
with belief_scope("assistant.confirm"):
|
||
record = CONFIRMATIONS.get(confirmation_id)
|
||
if not record:
|
||
record = _load_confirmation_from_db(db, confirmation_id)
|
||
if record:
|
||
CONFIRMATIONS[confirmation_id] = record
|
||
else:
|
||
raise HTTPException(status_code=404, detail="Confirmation not found")
|
||
|
||
if record.user_id != current_user.id:
|
||
raise HTTPException(
|
||
status_code=403, detail="Confirmation does not belong to current user"
|
||
)
|
||
|
||
if record.state != "pending":
|
||
raise HTTPException(
|
||
status_code=400, detail=f"Confirmation already {record.state}"
|
||
)
|
||
|
||
if datetime.utcnow() > record.expires_at:
|
||
record.state = "expired"
|
||
_update_confirmation_state(db, confirmation_id, "expired")
|
||
raise HTTPException(status_code=400, detail="Confirmation expired")
|
||
|
||
intent = record.intent
|
||
text, task_id, actions = await _dispatch_intent(
|
||
intent, current_user, task_manager, config_manager, db
|
||
)
|
||
record.state = "consumed"
|
||
_update_confirmation_state(db, confirmation_id, "consumed")
|
||
|
||
_append_history(
|
||
current_user.id,
|
||
record.conversation_id,
|
||
"assistant",
|
||
text,
|
||
state="started" if task_id else "success",
|
||
task_id=task_id,
|
||
)
|
||
_persist_message(
|
||
db,
|
||
current_user.id,
|
||
record.conversation_id,
|
||
"assistant",
|
||
text,
|
||
state="started" if task_id else "success",
|
||
task_id=task_id,
|
||
metadata={"intent": intent, "confirmation_id": confirmation_id},
|
||
)
|
||
audit_payload = {
|
||
"decision": "confirmed_execute",
|
||
"confirmation_id": confirmation_id,
|
||
"task_id": task_id,
|
||
"intent": intent,
|
||
}
|
||
_audit(current_user.id, audit_payload)
|
||
_persist_audit(db, current_user.id, audit_payload, record.conversation_id)
|
||
|
||
return AssistantMessageResponse(
|
||
conversation_id=record.conversation_id,
|
||
response_id=str(uuid.uuid4()),
|
||
state="started" if task_id else "success",
|
||
text=text,
|
||
intent=intent,
|
||
task_id=task_id,
|
||
actions=actions,
|
||
created_at=datetime.utcnow(),
|
||
)
|
||
|
||
|
||
# [/DEF:confirm_operation:Function]
|
||
|
||
|
||
@router.post(
|
||
"/confirmations/{confirmation_id}/cancel", response_model=AssistantMessageResponse
|
||
)
|
||
# [DEF:cancel_operation:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Cancel pending risky operation and mark confirmation token as cancelled.
|
||
# @PRE: confirmation_id exists, belongs to current user, and is still pending.
|
||
# @POST: Confirmation becomes cancelled and cannot be executed anymore.
|
||
# @RETURN: AssistantMessageResponse confirming cancellation.
|
||
async def cancel_operation(
|
||
confirmation_id: str,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
with belief_scope("assistant.cancel"):
|
||
record = CONFIRMATIONS.get(confirmation_id)
|
||
if not record:
|
||
record = _load_confirmation_from_db(db, confirmation_id)
|
||
if record:
|
||
CONFIRMATIONS[confirmation_id] = record
|
||
else:
|
||
raise HTTPException(status_code=404, detail="Confirmation not found")
|
||
|
||
if record.user_id != current_user.id:
|
||
raise HTTPException(
|
||
status_code=403, detail="Confirmation does not belong to current user"
|
||
)
|
||
|
||
if record.state != "pending":
|
||
raise HTTPException(
|
||
status_code=400, detail=f"Confirmation already {record.state}"
|
||
)
|
||
|
||
record.state = "cancelled"
|
||
_update_confirmation_state(db, confirmation_id, "cancelled")
|
||
text = "Операция отменена. Выполнение не запускалось."
|
||
_append_history(
|
||
current_user.id,
|
||
record.conversation_id,
|
||
"assistant",
|
||
text,
|
||
state="success",
|
||
confirmation_id=confirmation_id,
|
||
)
|
||
_persist_message(
|
||
db,
|
||
current_user.id,
|
||
record.conversation_id,
|
||
"assistant",
|
||
text,
|
||
state="success",
|
||
confirmation_id=confirmation_id,
|
||
metadata={"intent": record.intent},
|
||
)
|
||
audit_payload = {
|
||
"decision": "cancelled",
|
||
"confirmation_id": confirmation_id,
|
||
"intent": record.intent,
|
||
}
|
||
_audit(current_user.id, audit_payload)
|
||
_persist_audit(db, current_user.id, audit_payload, record.conversation_id)
|
||
|
||
return AssistantMessageResponse(
|
||
conversation_id=record.conversation_id,
|
||
response_id=str(uuid.uuid4()),
|
||
state="success",
|
||
text=text,
|
||
intent=record.intent,
|
||
confirmation_id=confirmation_id,
|
||
actions=[],
|
||
created_at=datetime.utcnow(),
|
||
)
|
||
|
||
|
||
# [/DEF:cancel_operation:Function]
|
||
|
||
|
||
# [DEF:list_conversations:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Return paginated conversation list for current user with archived flag and last message preview.
|
||
# @PRE: Authenticated user context and valid pagination params.
|
||
# @POST: Conversations are grouped by conversation_id sorted by latest activity descending.
|
||
# @RETURN: Dict with items, paging metadata, and archive segmentation counts.
|
||
@router.get("/conversations")
|
||
async def list_conversations(
|
||
page: int = Query(1, ge=1),
|
||
page_size: int = Query(20, ge=1, le=100),
|
||
include_archived: bool = Query(False),
|
||
archived_only: bool = Query(False),
|
||
search: Optional[str] = Query(None),
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
with belief_scope("assistant.conversations"):
|
||
user_id = current_user.id
|
||
include_archived = _coerce_query_bool(include_archived)
|
||
archived_only = _coerce_query_bool(archived_only)
|
||
_cleanup_history_ttl(db, user_id)
|
||
|
||
rows = (
|
||
db.query(AssistantMessageRecord)
|
||
.filter(AssistantMessageRecord.user_id == user_id)
|
||
.order_by(desc(AssistantMessageRecord.created_at))
|
||
.all()
|
||
)
|
||
|
||
summary: Dict[str, Dict[str, Any]] = {}
|
||
for row in rows:
|
||
conv_id = row.conversation_id
|
||
if not conv_id:
|
||
continue
|
||
created_at = row.created_at or datetime.utcnow()
|
||
if conv_id not in summary:
|
||
summary[conv_id] = {
|
||
"conversation_id": conv_id,
|
||
"title": "",
|
||
"updated_at": created_at,
|
||
"last_message": row.text,
|
||
"last_role": row.role,
|
||
"last_state": row.state,
|
||
"last_task_id": row.task_id,
|
||
"message_count": 0,
|
||
}
|
||
item = summary[conv_id]
|
||
item["message_count"] += 1
|
||
if row.role == "user" and row.text and not item["title"]:
|
||
item["title"] = row.text.strip()[:80]
|
||
|
||
items = []
|
||
search_term = search.lower().strip() if search else ""
|
||
archived_total = sum(
|
||
1
|
||
for c in summary.values()
|
||
if _is_conversation_archived(c.get("updated_at"))
|
||
)
|
||
active_total = len(summary) - archived_total
|
||
for conv in summary.values():
|
||
conv["archived"] = _is_conversation_archived(conv.get("updated_at"))
|
||
if not conv.get("title"):
|
||
conv["title"] = f"Conversation {conv['conversation_id'][:8]}"
|
||
if search_term:
|
||
haystack = (
|
||
f"{conv.get('title', '')} {conv.get('last_message', '')}".lower()
|
||
)
|
||
if search_term not in haystack:
|
||
continue
|
||
if archived_only and not conv["archived"]:
|
||
continue
|
||
if not archived_only and not include_archived and conv["archived"]:
|
||
continue
|
||
updated = conv.get("updated_at")
|
||
conv["updated_at"] = (
|
||
updated.isoformat() if isinstance(updated, datetime) else None
|
||
)
|
||
items.append(conv)
|
||
|
||
items.sort(key=lambda x: x.get("updated_at") or "", reverse=True)
|
||
total = len(items)
|
||
start = (page - 1) * page_size
|
||
page_items = items[start : start + page_size]
|
||
|
||
return {
|
||
"items": page_items,
|
||
"total": total,
|
||
"page": page,
|
||
"page_size": page_size,
|
||
"has_next": start + page_size < total,
|
||
"active_total": active_total,
|
||
"archived_total": archived_total,
|
||
}
|
||
|
||
|
||
# [/DEF:list_conversations:Function]
|
||
|
||
|
||
# [DEF:delete_conversation:Function]
|
||
# @COMPLEXITY: 2
|
||
# @PURPOSE: Soft-delete or hard-delete a conversation and clear its in-memory trace.
|
||
# @PRE: conversation_id belongs to current_user.
|
||
# @POST: Conversation records are removed from DB and CONVERSATIONS cache.
|
||
@router.delete("/conversations/{conversation_id}")
|
||
async def delete_conversation(
|
||
conversation_id: str,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
with belief_scope("assistant.conversations.delete"):
|
||
user_id = current_user.id
|
||
|
||
# 1. Remove from in-memory cache
|
||
key = (user_id, conversation_id)
|
||
if key in CONVERSATIONS:
|
||
del CONVERSATIONS[key]
|
||
|
||
# 2. Delete from database
|
||
deleted_count = (
|
||
db.query(AssistantMessageRecord)
|
||
.filter(
|
||
AssistantMessageRecord.user_id == user_id,
|
||
AssistantMessageRecord.conversation_id == conversation_id,
|
||
)
|
||
.delete()
|
||
)
|
||
|
||
db.commit()
|
||
|
||
if deleted_count == 0:
|
||
raise HTTPException(
|
||
status_code=404, detail="Conversation not found or already deleted"
|
||
)
|
||
|
||
return {
|
||
"status": "success",
|
||
"deleted": deleted_count,
|
||
"conversation_id": conversation_id,
|
||
}
|
||
|
||
|
||
# [/DEF:delete_conversation:Function]
|
||
|
||
|
||
@router.get("/history")
|
||
# [DEF:get_history:Function]
|
||
# @PURPOSE: Retrieve paginated assistant conversation history for current user.
|
||
# @PRE: Authenticated user is available and page params are valid.
|
||
# @POST: Returns persistent messages and mirrored in-memory snapshot for diagnostics.
|
||
# @RETURN: Dict with items, paging metadata, and resolved conversation_id.
|
||
async def get_history(
|
||
page: int = Query(1, ge=1),
|
||
page_size: int = Query(20, ge=1, le=100),
|
||
conversation_id: Optional[str] = Query(None),
|
||
from_latest: bool = Query(False),
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
):
|
||
with belief_scope("assistant.history"):
|
||
user_id = current_user.id
|
||
_cleanup_history_ttl(db, user_id)
|
||
conv_id = _resolve_or_create_conversation(user_id, conversation_id, db)
|
||
|
||
base_query = db.query(AssistantMessageRecord).filter(
|
||
AssistantMessageRecord.user_id == user_id,
|
||
AssistantMessageRecord.conversation_id == conv_id,
|
||
)
|
||
total = base_query.count()
|
||
start = (page - 1) * page_size
|
||
if from_latest:
|
||
rows = (
|
||
base_query.order_by(desc(AssistantMessageRecord.created_at))
|
||
.offset(start)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
rows = list(reversed(rows))
|
||
else:
|
||
rows = (
|
||
base_query.order_by(AssistantMessageRecord.created_at.asc())
|
||
.offset(start)
|
||
.limit(page_size)
|
||
.all()
|
||
)
|
||
|
||
persistent_items = [
|
||
{
|
||
"message_id": row.id,
|
||
"conversation_id": row.conversation_id,
|
||
"role": row.role,
|
||
"text": row.text,
|
||
"state": row.state,
|
||
"task_id": row.task_id,
|
||
"confirmation_id": row.confirmation_id,
|
||
"created_at": row.created_at.isoformat() if row.created_at else None,
|
||
"metadata": row.payload,
|
||
}
|
||
for row in rows
|
||
]
|
||
|
||
memory_items = CONVERSATIONS.get((user_id, conv_id), [])
|
||
return {
|
||
"items": persistent_items,
|
||
"memory_items": memory_items,
|
||
"total": total,
|
||
"page": page,
|
||
"page_size": page_size,
|
||
"has_next": start + page_size < total,
|
||
"from_latest": from_latest,
|
||
"conversation_id": conv_id,
|
||
}
|
||
|
||
|
||
# [/DEF:get_history:Function]
|
||
|
||
|
||
@router.get("/audit")
|
||
# [DEF:get_assistant_audit:Function]
|
||
# @PURPOSE: Return assistant audit decisions for current user from persistent and in-memory stores.
|
||
# @PRE: User has tasks:READ permission.
|
||
# @POST: Audit payload is returned in reverse chronological order from DB.
|
||
# @RETURN: Dict with persistent and memory audit slices.
|
||
async def get_assistant_audit(
|
||
limit: int = Query(50, ge=1, le=500),
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
_=Depends(has_permission("tasks", "READ")),
|
||
):
|
||
with belief_scope("assistant.audit"):
|
||
memory_rows = ASSISTANT_AUDIT.get(current_user.id, [])
|
||
db_rows = (
|
||
db.query(AssistantAuditRecord)
|
||
.filter(AssistantAuditRecord.user_id == current_user.id)
|
||
.order_by(AssistantAuditRecord.created_at.desc())
|
||
.limit(limit)
|
||
.all()
|
||
)
|
||
persistent = [
|
||
{
|
||
"id": row.id,
|
||
"user_id": row.user_id,
|
||
"conversation_id": row.conversation_id,
|
||
"decision": row.decision,
|
||
"task_id": row.task_id,
|
||
"message": row.message,
|
||
"payload": row.payload,
|
||
"created_at": row.created_at.isoformat() if row.created_at else None,
|
||
}
|
||
for row in db_rows
|
||
]
|
||
return {
|
||
"items": persistent,
|
||
"memory_items": memory_rows[-limit:],
|
||
"total": len(persistent),
|
||
"memory_total": len(memory_rows),
|
||
}
|
||
|
||
|
||
# [/DEF:get_assistant_audit:Function]
|
||
|
||
# [/DEF:AssistantApi:Module]
|