diff --git a/.opencode/skills/molecular-cot-logging/SKILL.md b/.opencode/skills/molecular-cot-logging/SKILL.md index 4172090d..f9569672 100644 --- a/.opencode/skills/molecular-cot-logging/SKILL.md +++ b/.opencode/skills/molecular-cot-logging/SKILL.md @@ -9,7 +9,7 @@ description: Structured logging protocol for agent-driven development, based on @RELATION DISPATCHES -> [Std.Semantics.Python] @RELATION DISPATCHES -> [Std.Semantics.Svelte] @RATIONALE Long CoT chains need stabilisation through explicit reasoning bonds. The three-marker system (REASON/REFLECT/EXPLORE) maps directly to the molecular CoT paper and produces machine-readable execution traces that LLM agents can parse, analyse, and use for fine-tuning (MoLE-Syn bond distributions). Without structured markers, agent-generated code exhibits invisible failures: a function returns `None` instead of raising — the agent's attention never sees it because there's no log; a fallback path activates silently — no EXPLORE marker, no trace. JSON-line format ensures every log entry is a self-contained, parseable unit that survives log rotation, aggregation, and agent parsing — unlike plain-text logs that require regex heuristics. -@REJECTED Legacy Entry/Exit/Action/Coherence markers rejected — they are too generic, do not map to reasoning structure, and prevent traceability graph analysis. Plain-text logging rejected — JSON lines are mandatory for agent parsing. Unstructured printf-style logging rejected — agents cannot reliably extract structured fields (trace_id, marker, intent) from free-form text, making automated diagnosis impossible. +@REJECTED Legacy Entry/Exit/Action/Coherence markers rejected — they are too generic, do not map to reasoning structure, and prevent traceability graph analysis. Plain-text logging rejected — JSON lines are mandatory for agent parsing. Unstructured printf-style logging rejected — agents cannot reliably extract structured fields (trace_id, marker, intent) from free-form text, making automated diagnosis impossible. cot_span decorator rejected — replaced by belief_scope context manager + logger.reason/reflect/explore which gives more granular intent control per logical branch. @DATA_CONTRACT LogEntry -> { ts: str, level: str, trace_id: str, span_id?: str, src: str, marker: REASON|REFLECT|EXPLORE, intent: str, payload?: object, error?: str } @INVARIANT Every log line MUST carry exactly one valid marker (REASON | REFLECT | EXPLORE). No markerless log lines in C4/C5 code. @INVARIANT trace_id MUST propagate via ContextVar across async boundaries. Every incoming request or background job seeds a new trace_id. @@ -207,58 +207,7 @@ class TraceMiddleware(BaseHTTPMiddleware): return response ``` -## IV. Python Decorator (Span + Marker) - -For C4/C5 functions, a decorator that auto-emits REASON / REFLECT markers: - -```python -import asyncio -from functools import wraps - -def cot_span(marker: str = "REASON", intent: str | None = None): - """Wrap a function in a CoT span. On enter → REASON, on success → REFLECT, - on exception → EXPLORE.""" - def decorator(func): - @wraps(func) - async def async_wrapper(*args, **kwargs): - src = f"{func.__module__}.{func.__qualname__}" - prev_span = push_span(func.__qualname__) - default_intent = intent or f"Execute {func.__qualname__}" - try: - log(src, marker, default_intent, payload=_summarise_args(args, kwargs)) - result = await func(*args, **kwargs) - log(src, "REFLECT", f"{func.__qualname__} completed", - payload={"result": _summarise_value(result)}) - return result - except Exception as e: - log(src, "EXPLORE", f"{func.__qualname__} failed", - error=str(e), payload={"args": _summarise_args(args, kwargs)}) - raise - finally: - pop_span(prev_span) - - @wraps(func) - def sync_wrapper(*args, **kwargs): - ... # same logic, sync variant - return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper - return decorator - - -def _summarise_value(val, max_len: int = 200) -> str: - s = str(val) - return s[:max_len] + "..." if len(s) > max_len else s - -def _summarise_args(args, kwargs) -> dict: - # Skip 'self', 'cls', 'db', 'request' — too verbose - skip = {"self", "cls", "db", "request", "session"} - result = {} - for k, v in kwargs.items(): - if k not in skip: - result[k] = _summarise_value(v) - return result -``` - -## V. Svelte / Frontend Pattern +## IV. Svelte / Frontend Pattern The frontend implementation lives at `frontend/src/lib/cot-logger.ts` (installed as `$lib/cot-logger`). @@ -321,7 +270,7 @@ const res = await requestApi("/api/endpoint"); if (res.trace_id) setTraceId(res.trace_id); ``` -## VI. CLI / Stdout Reader (for humans) +## V. CLI / Stdout Reader (for humans) To make JSON lines readable in development: @@ -341,7 +290,7 @@ for line in sys.stdin: " ``` -## VII. Anti-patterns +## VI. Anti-patterns | ❌ Don't | ✅ Do | |----------|-------| diff --git a/backend/src/agent/_confirmation.py b/backend/src/agent/_confirmation.py index bf01bb5b..661a6e35 100644 --- a/backend/src/agent/_confirmation.py +++ b/backend/src/agent/_confirmation.py @@ -122,7 +122,7 @@ async def handle_resume( user_jwt: str = "", env_id: str | None = None, ) -> AsyncGenerator[str]: from src.agent.context import set_user_jwt - from src.core.cot_logger import log + from src.core.logger import logger set_user_jwt(user_jwt) pending = _pending_confirmations.pop(conversation_id, None) @@ -134,7 +134,11 @@ async def handle_resume( }) return if action == "confirm": - log("AgentChat.Confirmation", "REASON", "Fast-path confirmation resume", {"tool": pending.get("tool_name"), "conv_id": conversation_id}) + logger.reason( + "Fast-path confirmation resume", + payload={"tool": pending.get("tool_name"), "conv_id": conversation_id}, + extra={"src": "AgentChat.Confirmation"}, + ) tool_name = str(pending.get("tool_name") or "unknown_action") tool_args = normalize_tool_args(pending.get("tool_args")) yield json.dumps({ @@ -148,7 +152,11 @@ async def handle_resume( tool_obj = find_tool(tool_name) if tool_obj is None: error = f"Unknown tool: {tool_name}" - log("AgentChat.Confirmation", "EXPLORE", "Unknown tool in resume", {"tool": tool_name}, error=error) + logger.explore( + "Unknown tool in resume", + payload={"tool": tool_name}, error=error, + extra={"src": "AgentChat.Confirmation"}, + ) yield json.dumps({ "content": f"❌ {tool_name} — {error}", "metadata": {"type": "tool_error", "tool": tool_name, "error": error}, @@ -157,7 +165,11 @@ async def handle_resume( try: output = await tool_obj.ainvoke(tool_args) except Exception as exc: - log("AgentChat.Confirmation", "EXPLORE", "Tool invocation failed in resume", {"tool": tool_name}, error=str(exc)) + logger.explore( + "Tool invocation failed in resume", + payload={"tool": tool_name}, error=str(exc), + extra={"src": "AgentChat.Confirmation"}, + ) yield json.dumps({ "content": f"❌ {tool_name} — {exc}", "metadata": {"type": "tool_error", "tool": tool_name, "error": str(exc)}, @@ -171,10 +183,18 @@ async def handle_resume( "content": str(output), "metadata": {"type": "stream_token", "token": str(output)}, }) - log("AgentChat.Confirmation", "REFLECT", "Fast-path confirmation completed", {"tool": tool_name}) + logger.reflect( + "Fast-path confirmation completed", + payload={"tool": tool_name}, + extra={"src": "AgentChat.Confirmation"}, + ) return - log("AgentChat.Confirmation", "REASON", "LangGraph checkpoint resume", {"conv_id": conversation_id, "action": action}) + logger.reason( + "LangGraph checkpoint resume", + payload={"conv_id": conversation_id, "action": action}, + extra={"src": "AgentChat.Confirmation"}, + ) agent = await create_agent(get_all_tools(), env_id, interrupt_before=[]) if action == "confirm": config = {"configurable": {"thread_id": conversation_id}} @@ -205,7 +225,11 @@ async def handle_resume( "metadata": {"type": "tool_end", "tool": tool_name, "output": {"result": str(output)[:500]}}, }) elif action == "deny": - log("AgentChat.Confirmation", "REFLECT", "Checkpoint resume denied", {"conv_id": conversation_id}) + logger.reflect( + "Checkpoint resume denied", + payload={"conv_id": conversation_id}, + extra={"src": "AgentChat.Confirmation"}, + ) yield json.dumps({ "content": "⏹️ Операция отменена", "metadata": {"type": "confirm_resolved", "result": "denied"}, diff --git a/backend/src/agent/_persistence.py b/backend/src/agent/_persistence.py index ebb80689..4d2cdd05 100644 --- a/backend/src/agent/_persistence.py +++ b/backend/src/agent/_persistence.py @@ -15,7 +15,7 @@ from typing import Any import httpx -from src.core.cot_logger import log +from src.core.logger import logger SAVE_API_URL = os.getenv("FASTAPI_URL", "http://localhost:8000") + "/api/agent/conversations/save" TITLE_MAX_LENGTH = 80 @@ -170,13 +170,13 @@ def _get_llm_config() -> dict[str, Any] | None: async def _call_llm_for_title(user_text: str) -> str | None: """Call LLM to generate a 3-5 word Russian title. Returns title or None on failure.""" - from src.core.cot_logger import log as _log + from src.core.logger import logger as _logger try: import asyncio as _asyncio config = await _asyncio.to_thread(_get_llm_config) if not config or not config.get("configured"): - _log("AgentChat.Persistence", "EXPLORE", "LLM title: no provider configured", {}) + _logger.explore("LLM title: no provider configured", extra={"src": "AgentChat.Persistence"}) return None clean_text = clean_title(user_text)[:200] @@ -211,8 +211,11 @@ async def _call_llm_for_title(user_text: str) -> str | None: async with httpx.AsyncClient(timeout=10) as client: resp = await client.post(api_url, json=payload, headers=headers) if resp.status_code != 200: - _log("AgentChat.Persistence", "EXPLORE", "LLM title: API error", - {"status": resp.status_code}, error=resp.text[:200]) + _logger.explore( + "LLM title: API error", + payload={"status": resp.status_code}, error=resp.text[:200], + extra={"src": "AgentChat.Persistence"}, + ) return None data = resp.json() title = data.get("choices", [{}])[0].get("message", {}).get("content", "") @@ -223,7 +226,7 @@ async def _call_llm_for_title(user_text: str) -> str | None: if title: return title except Exception as e: - _log("AgentChat.Persistence", "EXPLORE", "LLM title generation failed", {}, error=str(e)) + _logger.explore("LLM title generation failed", error=str(e), extra={"src": "AgentChat.Persistence"}) return None @@ -263,11 +266,17 @@ async def generate_llm_title(conv_id: str, user_text: str) -> None: } async with httpx.AsyncClient(timeout=5) as client: await client.post(SAVE_API_URL, json=payload, headers=headers) - log("AgentChat.Persistence", "REFLECT", "LLM title updated", - {"conv_id": conv_id, "title": title[:40]}) + logger.reflect( + "LLM title updated", + payload={"conv_id": conv_id, "title": title[:40]}, + extra={"src": "AgentChat.Persistence"}, + ) except Exception as e: - log("AgentChat.Persistence", "EXPLORE", "LLM title save failed", - {"conv_id": conv_id}, error=str(e)) + logger.explore( + "LLM title save failed", + payload={"conv_id": conv_id}, error=str(e), + extra={"src": "AgentChat.Persistence"}, + ) finally: _title_locks.pop(conv_id, None) # #endregion AgentChat.Persistence.GenerateLlmTitle @@ -314,7 +323,12 @@ async def prefetch_dashboards(env_id: str) -> str: if total > limit: suffix = f"\n... {total - limit} more dashboards omitted. Ask for a narrower search if needed." return f"Available dashboards in environment '{env_id or 'default'}' ({total} total):\n" + "\n".join(lines) + suffix - except Exception: + except Exception as e: + logger.explore( + "Prefetch dashboards failed", + payload={"env_id": env_id}, error=str(e), + extra={"src": "AgentChat.Persistence.PrefetchDashboards"}, + ) return "" # #endregion AgentChat.Persistence.PrefetchDashboards @@ -369,8 +383,16 @@ async def save_conversation(conv_id: str, user_text: str, user_id: str = "admin" async with httpx.AsyncClient(timeout=10) as client: await client.post(SAVE_API_URL, json=payload, headers=headers) + logger.reflect( + "Conversation saved", + payload={"conv_id": conv_id, "msg_count": len(messages), "title": title[:40]}, + extra={"src": "AgentChat.Persistence.SaveConversation"}, + ) except Exception as e: - log("AgentChat.Persistence", "EXPLORE", "Failed to save conversation", - {"conv_id": conv_id}, error=str(e)) + logger.explore( + "Failed to save conversation", + payload={"conv_id": conv_id}, error=str(e), + extra={"src": "AgentChat.Persistence"}, + ) # #endregion AgentChat.Persistence.SaveConversation # #endregion AgentChat.Persistence diff --git a/backend/src/agent/_tool_resolver.py b/backend/src/agent/_tool_resolver.py index c1fe75b0..dea52317 100644 --- a/backend/src/agent/_tool_resolver.py +++ b/backend/src/agent/_tool_resolver.py @@ -9,7 +9,7 @@ from typing import Any from src.agent.tools import get_all_tools -from src.core.cot_logger import log +from src.core.logger import logger # #region AgentChat.ToolResolver.Sets [C:1] [TYPE Constants] [SEMANTICS agent-chat,tools,sets] # @ingroup AgentChat @@ -56,7 +56,11 @@ def known_agent_tool_names() -> set[str]: try: return {str(tool_obj.name) for tool_obj in get_all_tools() if getattr(tool_obj, "name", None)} except Exception as exc: - log("AgentChat.ToolResolver", "EXPLORE", "tool catalog lookup failed", {"error": str(exc)}) + logger.explore( + "tool catalog lookup failed", + payload={"error": str(exc)}, + extra={"src": "AgentChat.ToolResolver"}, + ) return _SAFE_AGENT_TOOLS | _GUARDED_AGENT_TOOLS | _DANGEROUS_AGENT_TOOLS # #endregion AgentChat.ToolResolver.KnownNames @@ -105,39 +109,46 @@ def coerce_tool_call(tool_call: Any) -> tuple[str | None, dict[str, Any]]: # uses keyword matching to guess the user's intent and auto-trigger HITL. def infer_tool_from_text(text: str) -> str | None: lowered = (text or "").lower() + inferred: str | None = None if any(word in lowered for word in ["окруж", "environment", "env"]): - return "list_environments" - if any(word in lowered for word in ["maintenance", "обслуж", "баннер"]): + inferred = "list_environments" + elif any(word in lowered for word in ["maintenance", "обслуж", "баннер"]): if any(word in lowered for word in ["start", "созда", "запусти", "начни"]): - return "start_maintenance" - if any(word in lowered for word in ["end", "закрой", "заверши", "останов"]): - return "end_maintenance" - return "list_maintenance_events" - if any(word in lowered for word in ["дашборд", "dashboard", "dashboards", "дашборды"]): - return "search_dashboards" - if any(word in lowered for word in ["здоров", "health", "статус системы", "system status"]): - return "get_health_summary" - if any(word in lowered for word in ["задач", "task", "таск"]): - return "get_task_status" - if any(word in lowered for word in ["llm", "provider", "провайдер", "модель"]): - return "list_llm_providers" - if any(word in lowered for word in ["branch", "ветк"]): - return "create_branch" - if any(word in lowered for word in ["commit", "коммит"]): - return "commit_changes" - if any(word in lowered for word in ["deploy", "депло", "разверн"]): - return "deploy_dashboard" - if any(word in lowered for word in ["миграц", "migration", "migrate"]): - return "execute_migration" - if any(word in lowered for word in ["backup", "бэкап", "резерв"]): - return "run_backup" - if any(word in lowered for word in ["валидац", "validation", "validate"]): - return "run_llm_validation" - if any(word in lowered for word in ["документ", "documentation", "docs"]): - return "run_llm_documentation" - if any(word in lowered for word in ["инструмент", "tool", "capabilit", "умеешь", "можешь"]): - return "show_capabilities" - return None + inferred = "start_maintenance" + elif any(word in lowered for word in ["end", "закрой", "заверши", "останов"]): + inferred = "end_maintenance" + else: + inferred = "list_maintenance_events" + elif any(word in lowered for word in ["дашборд", "dashboard", "dashboards", "дашборды"]): + inferred = "search_dashboards" + elif any(word in lowered for word in ["здоров", "health", "статус системы", "system status"]): + inferred = "get_health_summary" + elif any(word in lowered for word in ["задач", "task", "таск"]): + inferred = "get_task_status" + elif any(word in lowered for word in ["llm", "provider", "провайдер", "модель"]): + inferred = "list_llm_providers" + elif any(word in lowered for word in ["branch", "ветк"]): + inferred = "create_branch" + elif any(word in lowered for word in ["commit", "коммит"]): + inferred = "commit_changes" + elif any(word in lowered for word in ["deploy", "депло", "разверн"]): + inferred = "deploy_dashboard" + elif any(word in lowered for word in ["миграц", "migration", "migrate"]): + inferred = "execute_migration" + elif any(word in lowered for word in ["backup", "бэкап", "резерв"]): + inferred = "run_backup" + elif any(word in lowered for word in ["валидац", "validation", "validate"]): + inferred = "run_llm_validation" + elif any(word in lowered for word in ["документ", "documentation", "docs"]): + inferred = "run_llm_documentation" + elif any(word in lowered for word in ["инструмент", "tool", "capabilit", "умеешь", "можешь"]): + inferred = "show_capabilities" + + if inferred: + logger.reason("Tool inferred from user text", + payload={"tool": inferred, "text_preview": (text or "")[:80]}, + extra={"src": "AgentChat.ToolResolver.InferFromText"}) + return inferred # #endregion AgentChat.ToolResolver.InferFromText @@ -155,7 +166,11 @@ def extract_tool_call_from_state(state, user_text: str = "") -> tuple[str | None if tool_name: return (str(tool_name), tool_args) except Exception as exc: - log("AgentChat.ToolResolver", "EXPLORE", "tool_call extraction failed", {"error": str(exc)}) + logger.explore( + "tool_call extraction failed", + payload={"error": str(exc)}, + extra={"src": "AgentChat.ToolResolver"}, + ) if getattr(state, "next", None): node_or_tool = str(state.next[0]) @@ -164,7 +179,11 @@ def extract_tool_call_from_state(state, user_text: str = "") -> tuple[str | None inferred_tool = infer_tool_from_text(user_text) if inferred_tool: - log("AgentChat.ToolResolver", "EXPLORE", "tool_call inferred from user text", {"tool": inferred_tool}) + logger.explore( + "tool_call inferred from user text", + payload={"tool": inferred_tool}, + extra={"src": "AgentChat.ToolResolver"}, + ) return (inferred_tool, {}) return (None, {}) # #endregion AgentChat.ToolResolver.ExtractFromState diff --git a/backend/src/agent/app.py b/backend/src/agent/app.py index 48d77dc4..da7f73ca 100644 --- a/backend/src/agent/app.py +++ b/backend/src/agent/app.py @@ -49,7 +49,7 @@ from src.agent.langgraph_setup import create_agent from src.agent.middleware import log_tool_event from src.agent.tools import get_tools_for_query from src.core.auth.jwt import decode_token -from src.core.cot_logger import log +from src.core.logger import logger JWT_SECRET = os.getenv("JWT_SECRET", "super-secret-key") MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024 # 10 MB @@ -96,8 +96,11 @@ def _persist_chat_file(file_path: str, conv_id: str) -> str | None: shutil.copy2(str(src), str(dest_file)) rel_path = str(dest_file.relative_to(Path(storage_root))) - log("AgentChat.PersistFile", "REFLECT", "Chat file persisted to storage", - {"original": src.name, "rel_path": rel_path, "size": dest_file.stat().st_size}) + logger.reflect( + "Chat file persisted to storage", + payload={"original": src.name, "rel_path": rel_path, "size": dest_file.stat().st_size}, + extra={"src": "AgentChat.PersistFile"}, + ) return rel_path # #endregion AgentChat.GradioApp.PersistFile # Per-conversation mutex for HITL resume (FR-026): keyed by conversation_id @@ -159,6 +162,13 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio try: # ── Resolve conversation ID early (needed for file persistence) ── conv_id = conversation_id or str(uuid.uuid4()) + is_resume = action in ("confirm", "deny") + logger.reason( + "Agent handler invoked", + payload={"user_id": user_id, "conv_id": conv_id, "action": action, "env_id": env_id, + "is_resume": is_resume, "msg_len": len(str(message))}, + extra={"src": "AgentChat.GradioApp.Handler"}, + ) # ── Parse message ── text = message.get("text", "") if isinstance(message, dict) else str(message) @@ -335,12 +345,24 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio if attempt < max_attempts - 1: text = "Respond with valid JSON only. Previous response was malformed.\n\n" + text continue + logger.explore( + "LLM malformed output", + payload={"conv_id": conv_id, "attempt": attempt}, + error=str(e), + extra={"src": "AgentChat.GradioApp.Handler"}, + ) yield json.dumps({ "content": "❌ Ошибка обработки ответа LLM. Пожалуйста, уточните запрос.", "metadata": {"type": "error", "code": "LLM_MALFORMED_OUTPUT", "detail": str(e)}, }) except Exception as exc: + logger.explore( + "Agent handler failed", + payload={"conv_id": conv_id, "user_id": user_id}, + error=str(exc), + extra={"src": "AgentChat.GradioApp.Handler"}, + ) try: state = await agent.aget_state(config) if getattr(state, "next", None): @@ -358,6 +380,11 @@ async def agent_handler( # noqa: C901 — intentionally complex C4 orchestratio await save_conversation(conv_id, text, user_id, assistant_text="".join(assistant_parts)) # Fire-and-forget: generate LLM title in background (best-effort) asyncio.create_task(generate_llm_title(conv_id, text)) + logger.reflect( + "Agent handler completed", + payload={"conv_id": conv_id, "assistant_len": len("".join(assistant_parts))}, + extra={"src": "AgentChat.GradioApp.Handler"}, + ) finally: _user_locks[user_id] = False diff --git a/backend/src/agent/langgraph_setup.py b/backend/src/agent/langgraph_setup.py index 88a0ef34..1d8bd615 100644 --- a/backend/src/agent/langgraph_setup.py +++ b/backend/src/agent/langgraph_setup.py @@ -20,6 +20,8 @@ from langgraph.checkpoint.postgres.aio import AsyncPostgresSaver from langgraph.prebuilt import create_react_agent from psycopg.rows import dict_row +from src.core.logger import logger + # ── Monkey-patch: OpenAI SDK for Pydantic BaseModel classes ── # LangChain BaseTool objects carry an ``args_schema`` field that is a Pydantic # BaseModel *class* reference (not an instance). When the OpenAI SDK recursively @@ -111,8 +113,9 @@ async def _fetch_llm_config() -> dict | None: if config.get("configured"): _llm_config = config return config - except Exception: - pass # Keep existing config on failure + except Exception as e: + logger.explore("Failed to fetch LLM config from FastAPI", error=str(e), + extra={"src": "AgentChat.LangGraph.Setup"}) return _llm_config @@ -151,10 +154,24 @@ async def create_agent( api_key = config["api_key"] base_url = config.get("base_url") or "https://api.openai.com/v1" model = config.get("default_model") or "gpt-4o-mini" + config_source = "FastAPI" else: api_key = os.getenv("LLM_API_KEY") base_url = os.getenv("LLM_BASE_URL", "https://api.openai.com/v1") model = os.getenv("LLM_MODEL", "gpt-4o") + config_source = "env vars" + logger.explore( + "LLM config not found in FastAPI, falling back to env vars", + payload={"model": model, "provider_type": config.get("provider_type") if config else None}, + error="No configured LLM provider in FastAPI", + extra={"src": "AgentChat.LangGraph.Setup"}, + ) + + logger.reason( + "Creating LangGraph agent", + payload={"model": model, "config_source": config_source, "tools_count": len(tools), "env_id": env_id}, + extra={"src": "AgentChat.LangGraph.Setup"}, + ) llm = ChatOpenAI( model=model, @@ -179,6 +196,11 @@ async def create_agent( checkpointer = _CHECKPOINTER else: checkpointer = InMemorySaver() + logger.explore( + "Postgres checkpointer unavailable, falling back to InMemorySaver", + error="_CHECKPOINTER is None — checkpoints will be lost on restart", + extra={"src": "AgentChat.LangGraph.Setup"}, + ) graph = create_react_agent( model=llm, @@ -189,5 +211,10 @@ async def create_agent( interrupt_before=_interrupt_before_from_env() if interrupt_before is None else interrupt_before, ) + logger.reflect( + "LangGraph agent created", + payload={"model": model, "checkpointer_type": type(checkpointer).__name__, "tools_count": len(tools)}, + extra={"src": "AgentChat.LangGraph.Setup"}, + ) return graph # #endregion AgentChat.LangGraph.Setup diff --git a/backend/src/agent/middleware.py b/backend/src/agent/middleware.py index b4df238b..5bfdfb98 100644 --- a/backend/src/agent/middleware.py +++ b/backend/src/agent/middleware.py @@ -7,11 +7,9 @@ # @REJECTED ConfirmationRiskMiddleware rejected — LangGraph interrupt_before=DANGEROUS_TOOLS handles HITL natively. from datetime import UTC, datetime -import logging from src.agent.context import get_user_jwt - -logger = logging.getLogger("cot") +from src.core.logger import logger # #region AgentChat.Middleware.LoggingMiddleware [C:3] [TYPE Function] [SEMANTICS audit,tool,logging] @@ -48,9 +46,10 @@ async def log_tool_event(event: dict, conversation_id: str) -> None: elif kind == "on_tool_error": audit_payload["error"] = str(data.get("error", ""))[:500] - logger.info( - "Tool audit: %(event_type)s — %(tool)s — conv=%(conversation_id)s", - audit_payload, + logger.reason( + "Tool audit event", + payload=audit_payload, + extra={"src": "AgentChat.Middleware.LoggingMiddleware"}, ) # TODO: Async write to assistant_audit table via REST call to FastAPI diff --git a/backend/src/agent/run.py b/backend/src/agent/run.py index 03c7c132..485dd173 100644 --- a/backend/src/agent/run.py +++ b/backend/src/agent/run.py @@ -11,9 +11,8 @@ import os import socket import httpx -import logging -logger = logging.getLogger("cot") +from src.core.logger import logger FASTAPI_URL = os.getenv("FASTAPI_URL", "http://localhost:8000") @@ -46,16 +45,37 @@ def _fetch_llm_config() -> dict | None: resp.raise_for_status() config = resp.json() if config.get("configured"): - logger.info("LLM config fetched from FastAPI: %s (%s)", config.get("provider_type"), config.get("default_model")) + logger.reason( + "LLM config fetched from FastAPI", + payload={"provider_type": config.get("provider_type"), "model": config.get("default_model")}, + extra={"src": "AgentChat.Run.FetchLlmConfig"}, + ) return config - logger.warning("FastAPI returned no active LLM provider: %s", config.get("reason")) + logger.explore( + "FastAPI returned no active LLM provider", + payload={"reason": config.get("reason")}, + error="No configured LLM provider", + extra={"src": "AgentChat.Run.FetchLlmConfig"}, + ) except Exception as e: if attempt < 5: - logger.info("Waiting for FastAPI (attempt %d/6): %s", attempt + 1, e) + logger.reason( + f"Waiting for FastAPI (attempt {attempt + 1}/6)", + payload={"error": str(e)}, + extra={"src": "AgentChat.Run.FetchLlmConfig"}, + ) time.sleep(5) else: - logger.warning("Failed to fetch LLM config from FastAPI after 6 attempts: %s", e) - logger.info("Falling back to env vars for LLM config") + logger.explore( + "Failed to fetch LLM config after 6 attempts", + error=str(e), + extra={"src": "AgentChat.Run.FetchLlmConfig"}, + ) + logger.explore( + "Falling back to env vars for LLM config", + error="FastAPI unreachable", + extra={"src": "AgentChat.Run.FetchLlmConfig"}, + ) return None @@ -85,9 +105,18 @@ if __name__ == "__main__": try: port = _find_free_port(configured_port) if port != configured_port: - logger.warning("Port %d is in use, falling back to port %d", configured_port, port) + logger.explore( + "Port in use, falling back", + payload={"configured_port": configured_port, "actual_port": port}, + error=f"Port {configured_port} is in use", + extra={"src": "AgentChat.Run.PortBinding"}, + ) except OSError as e: - logger.error("Failed to find a free port: %s", e) + logger.explore( + "Failed to find a free port", + error=str(e), + extra={"src": "AgentChat.Run.PortBinding"}, + ) raise else: port = configured_port diff --git a/backend/src/agent/tools.py b/backend/src/agent/tools.py index 3346977e..a7348363 100644 --- a/backend/src/agent/tools.py +++ b/backend/src/agent/tools.py @@ -14,6 +14,7 @@ from langchain_core.tools import tool from pydantic import BaseModel, Field from src.agent.context import get_service_jwt, get_user_jwt +from src.core.logger import logger FASTAPI_URL = os.getenv("FASTAPI_URL", "http://localhost:8000") TOOL_RESPONSE_LIMIT = 4000 @@ -178,9 +179,15 @@ class SearchDashboardsInput(BaseModel): @tool(args_schema=SearchDashboardsInput) async def search_dashboards(query: str, env_id: str | None = None) -> str: """Search and list dashboards by name, with optional environment filter.""" + logger.reason("Search dashboards", payload={"query": query, "env_id": env_id}, + extra={"src": "AgentChat.Tools.SearchDashboards"}) params = {"q": query, "env_id": env_id or ""} resp = await _get("/api/dashboards", params=params) if resp.status_code != 200: + logger.explore("Dashboard search failed", + payload={"status": resp.status_code, "query": query}, + error=resp.text[:200], + extra={"src": "AgentChat.Tools.SearchDashboards"}) return f"Error {resp.status_code}: Unable to fetch dashboards. {resp.text}" try: @@ -189,6 +196,9 @@ async def search_dashboards(query: str, env_id: str | None = None) -> str: total = data.get("total", 0) if total == 0: + logger.reflect("No dashboards found", + payload={"query": query, "total": 0}, + extra={"src": "AgentChat.Tools.SearchDashboards"}) return f"No dashboards found matching '{query}' in environment '{env_id or 'default'}'." lines = [f"Found {total} dashboard(s) in environment '{env_id or 'default'}':"] @@ -199,8 +209,14 @@ async def search_dashboards(query: str, env_id: str | None = None) -> str: lines.append(f" - {title}") lines.append(f" Owners: {owners}") lines.append(f" Last modified: {modified}") + logger.reflect("Dashboards listed", + payload={"query": query, "total": total}, + extra={"src": "AgentChat.Tools.SearchDashboards"}) return "\n".join(lines) except (json.JSONDecodeError, KeyError) as e: + logger.explore("Failed to parse dashboards response", + payload={"query": query}, error=str(e), + extra={"src": "AgentChat.Tools.SearchDashboards"}) return f"Could not parse dashboards response: {e}" # #endregion AgentChat.Tools.SearchDashboards @@ -220,9 +236,18 @@ class HealthSummaryInput(BaseModel): @tool(args_schema=HealthSummaryInput) async def get_health_summary(env_id: str | None = None) -> str: """Get system health summary — dashboard validation status, recent failures.""" + logger.reason("Get health summary", payload={"env_id": env_id}, + extra={"src": "AgentChat.Tools.HealthSummary"}) resp = await _get("/api/health/summary", params={"environment_id": env_id or ""}) if resp.status_code != 200: + logger.explore("Health check failed", + payload={"status": resp.status_code}, + error=resp.text[:200], + extra={"src": "AgentChat.Tools.HealthSummary"}) return f"Health check failed: HTTP {resp.status_code}" + logger.reflect("Health summary retrieved", + payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.HealthSummary"}) return _trim_response(resp.text, 2000) # #endregion AgentChat.Tools.HealthSummary @@ -236,8 +261,11 @@ async def get_health_summary(env_id: str | None = None) -> str: @tool async def list_environments() -> str: """List configured deployment environments.""" + logger.reason("List environments", extra={"src": "AgentChat.Tools.ListEnvironments"}) resp = await _get("/api/settings/environments") result = _api_result(resp) + logger.reflect("Environments listed", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.ListEnvironments"}) return _trim_response(_safe_json_text(result)) # #endregion AgentChat.Tools.ListEnvironments @@ -251,8 +279,13 @@ async def list_environments() -> str: @tool async def get_task_status(task_id: str) -> str: """Check the status of a background task by its task_id.""" + logger.reason("Get task status", payload={"task_id": task_id}, + extra={"src": "AgentChat.Tools.TaskStatus"}) resp = await _get(f"/api/tasks/{task_id}") - return _api_result(resp) + result = _api_result(resp) + logger.reflect("Task status retrieved", payload={"task_id": task_id, "status": resp.status_code}, + extra={"src": "AgentChat.Tools.TaskStatus"}) + return result # #endregion AgentChat.Tools.TaskStatus @@ -278,8 +311,12 @@ async def show_capabilities() -> str: @tool async def list_llm_providers() -> str: """List configured LLM providers.""" + logger.reason("List LLM providers", extra={"src": "AgentChat.Tools.ListLlmProviders"}) resp = await _get("/api/llm/providers") - return _api_result(resp) + result = _api_result(resp) + logger.reflect("LLM providers listed", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.ListLlmProviders"}) + return result # #endregion AgentChat.Tools.ListLlmProviders @@ -292,8 +329,12 @@ async def list_llm_providers() -> str: @tool async def get_llm_status() -> str: """Check whether the LLM runtime is configured and usable.""" + logger.reason("Get LLM status", extra={"src": "AgentChat.Tools.LlmStatus"}) resp = await _get("/api/llm/status") - return _api_result(resp) + result = _api_result(resp) + logger.reflect("LLM status retrieved", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.LlmStatus"}) + return result # #endregion AgentChat.Tools.LlmStatus @@ -318,12 +359,17 @@ async def run_backup( dashboard_ids: list[int] | None = None, ) -> str: """Run a Superset backup for an environment, optionally scoped to dashboard IDs.""" + vars = {"env_id": environment_id, "dashboard_id": dashboard_id, "dashboard_ids": dashboard_ids} + logger.reason("Run backup", payload=vars, extra={"src": "AgentChat.Tools.RunBackup"}) params: dict[str, Any] = {"environment_id": environment_id} - ids = _dashboard_ids(dashboard_id, dashboard_ids) - if ids: - params["dashboard_ids"] = ids + ids_l = _dashboard_ids(dashboard_id, dashboard_ids) + if ids_l: + params["dashboard_ids"] = ids_l resp = await _post("/api/tasks", {"plugin_id": "superset-backup", "params": params}) - return _api_result(resp, {201}) + result = _api_result(resp, {201}) + logger.reflect("Backup task created", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.RunBackup"}) + return result # #endregion AgentChat.Tools.RunBackup @@ -352,6 +398,9 @@ async def execute_migration( fix_cross_filters: bool = True, ) -> str: """Execute dashboard migration between two environments.""" + logger.reason("Execute migration", + payload={"source": source_env_id, "target": target_env_id, "ids_count": len(selected_ids)}, + extra={"src": "AgentChat.Tools.ExecuteMigration"}) payload = { "source_env_id": source_env_id, "target_env_id": target_env_id, @@ -360,7 +409,10 @@ async def execute_migration( "fix_cross_filters": fix_cross_filters, } resp = await _post("/api/migration/execute", payload) - return _api_result(resp) + result = _api_result(resp) + logger.reflect("Migration executed", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.ExecuteMigration"}) + return result # #endregion AgentChat.Tools.ExecuteMigration @@ -394,12 +446,18 @@ async def create_branch( env_id: str | None = None, ) -> str: """Create a branch in a dashboard Git repository.""" + logger.reason("Create branch", + payload={"dashboard_ref": dashboard_ref, "branch_name": branch_name, "from_branch": from_branch}, + extra={"src": "AgentChat.Tools.CreateBranch"}) resp = await _post( f"/api/git/repositories/{dashboard_ref}/branches", {"name": branch_name, "from_branch": from_branch}, params={"env_id": env_id} if env_id else None, ) - return _api_result(resp) + result = _api_result(resp) + logger.reflect("Branch created", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.CreateBranch"}) + return result # #endregion AgentChat.Tools.CreateBranch @@ -424,12 +482,18 @@ async def commit_changes( env_id: str | None = None, ) -> str: """Stage and commit changes in a dashboard Git repository.""" + logger.reason("Commit changes", + payload={"dashboard_ref": dashboard_ref, "message": message[:80]}, + extra={"src": "AgentChat.Tools.CommitChanges"}) resp = await _post( f"/api/git/repositories/{dashboard_ref}/commit", {"message": message, "files": files or []}, params={"env_id": env_id} if env_id else None, ) - return _api_result(resp) + result = _api_result(resp) + logger.reflect("Changes committed", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.CommitChanges"}) + return result # #endregion AgentChat.Tools.CommitChanges @@ -452,12 +516,18 @@ async def deploy_dashboard( env_id: str | None = None, ) -> str: """Deploy a dashboard from Git to a target environment.""" + logger.reason("Deploy dashboard", + payload={"dashboard_ref": dashboard_ref, "target_env": environment_id}, + extra={"src": "AgentChat.Tools.DeployDashboard"}) resp = await _post( f"/api/git/repositories/{dashboard_ref}/deploy", {"environment_id": environment_id}, params={"env_id": env_id} if env_id else None, ) - return _api_result(resp) + result = _api_result(resp) + logger.reflect("Dashboard deployed", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.DeployDashboard"}) + return result # #endregion AgentChat.Tools.DeployDashboard @@ -482,6 +552,9 @@ async def run_llm_documentation( provider_id: str | None = None, ) -> str: """Generate dataset documentation via the LLM documentation task.""" + logger.reason("Run LLM documentation", + payload={"dataset_id": dataset_id, "environment_id": environment_id}, + extra={"src": "AgentChat.Tools.LlmDocumentation"}) resolved_provider = await _resolve_active_provider_id(provider_id) params = { "dataset_id": str(dataset_id), @@ -490,7 +563,11 @@ async def run_llm_documentation( if resolved_provider: params["provider_id"] = resolved_provider resp = await _post("/api/tasks", {"plugin_id": "llm_documentation", "params": params}) - return _api_result(resp, {201}) + result = _api_result(resp, {201}) + logger.reflect("LLM documentation task created", + payload={"status": resp.status_code, "provider_id": resolved_provider}, + extra={"src": "AgentChat.Tools.LlmDocumentation"}) + return result # #endregion AgentChat.Tools.LlmDocumentation @@ -524,10 +601,19 @@ async def run_llm_validation( ) -> str: """Create and immediately run an LLM dashboard validation policy, or run an existing policy.""" if policy_id: + logger.reason("Run existing validation policy", + payload={"policy_id": policy_id}, + extra={"src": "AgentChat.Tools.LlmValidation"}) resp = await _post(f"/api/validation-tasks/{policy_id}/run") - return _api_result(resp) + result = _api_result(resp) + logger.reflect("Validation policy run", payload={"policy_id": policy_id, "status": resp.status_code}, + extra={"src": "AgentChat.Tools.LlmValidation"}) + return result ids = _dashboard_ids(dashboard_id, dashboard_ids) + logger.reason("Create LLM validation policy", + payload={"environment_id": environment_id, "dashboard_count": len(ids)}, + extra={"src": "AgentChat.Tools.LlmValidation"}) sources = [] if dashboard_url: sources.append({"type": "dashboard_url", "value": dashboard_url}) @@ -560,7 +646,11 @@ async def run_llm_validation( except (json.JSONDecodeError, KeyError) as exc: return f"Validation policy created but response could not be parsed: {exc}. Raw: {create_resp.text}" run_resp = await _post(f"/api/validation-tasks/{created_id}/run") - return _api_result(run_resp) + result = _api_result(run_resp) + logger.reflect("Validation policy created and run", + payload={"policy_id": created_id, "status": run_resp.status_code}, + extra={"src": "AgentChat.Tools.LlmValidation"}) + return result # #endregion AgentChat.Tools.LlmValidation @@ -573,8 +663,12 @@ async def run_llm_validation( @tool async def list_maintenance_events() -> str: """List active and completed maintenance events.""" + logger.reason("List maintenance events", extra={"src": "AgentChat.Tools.ListMaintenance"}) resp = await _get("/api/maintenance/events") - return _api_result(resp) + result = _api_result(resp) + logger.reflect("Maintenance events listed", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.ListMaintenance"}) + return result # #endregion AgentChat.Tools.ListMaintenance @@ -603,6 +697,9 @@ async def start_maintenance( message: str | None = None, ) -> str: """Start a maintenance event and apply banners to affected dashboards.""" + logger.reason("Start maintenance", + payload={"environment_id": environment_id, "tables": tables}, + extra={"src": "AgentChat.Tools.StartMaintenance"}) payload = { "tables": tables, "environment_id": environment_id, @@ -611,7 +708,10 @@ async def start_maintenance( "message": message, } resp = await _post("/api/maintenance/start", payload) - return _api_result(resp, {200, 202}) + result = _api_result(resp, {200, 202}) + logger.reflect("Maintenance started", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.StartMaintenance"}) + return result # #endregion AgentChat.Tools.StartMaintenance @@ -637,13 +737,25 @@ async def end_maintenance( ) -> str: """End one maintenance event, or end all active events when end_all is true.""" if end_all: + logger.reason("End all maintenance events", + payload={"environment_id": environment_id}, + extra={"src": "AgentChat.Tools.EndMaintenance"}) payload = {"environment_id": environment_id} if environment_id else {} resp = await _post("/api/maintenance/end-all", payload) - return _api_result(resp, {202}) + result = _api_result(resp, {202}) + logger.reflect("All maintenance ended", payload={"status": resp.status_code}, + extra={"src": "AgentChat.Tools.EndMaintenance"}) + return result if not maintenance_id: return "Error: maintenance_id is required unless end_all=true." + logger.reason("End maintenance event", + payload={"maintenance_id": maintenance_id}, + extra={"src": "AgentChat.Tools.EndMaintenance"}) resp = await _post(f"/api/maintenance/{maintenance_id}/end") - return _api_result(resp, {202}) + result = _api_result(resp, {202}) + logger.reflect("Maintenance ended", payload={"status": resp.status_code, "maintenance_id": maintenance_id}, + extra={"src": "AgentChat.Tools.EndMaintenance"}) + return result # #endregion AgentChat.Tools.EndMaintenance diff --git a/backend/src/core/cot_logger.py b/backend/src/core/cot_logger.py index b832c95f..5a4854b3 100644 --- a/backend/src/core/cot_logger.py +++ b/backend/src/core/cot_logger.py @@ -170,6 +170,9 @@ def log( # #region MarkerLogger [C:2] [TYPE Class] [SEMANTICS logger, proxy, marker, syntactic-sugar] # @ingroup Core # @BRIEF Thin proxy class providing .reason(), .reflect(), .explore() convenience methods. +# @DEPRECATED Use logger.reason/reflect/explore from src.core.logger instead — MarkerLogger has +# zero usages in the codebase and adds a redundant abstraction layer. +# @REPLACED_BY logger.reason/reflect/explore (monkey-patched on superset_tools_app logger) class MarkerLogger: """Thin proxy over the cot_logger.log() function.