"""Chanakya routing agent — tools for consulting sub-agents.""" import asyncio import json import os import time as _time import httpx from loguru import logger as _log from pydantic_ai import RunContext from backend.chanakya.a2a_context import RequestContext from backend.chanakya.base import EnterpriseAgent from backend.chanakya.rt_agent._http import TATASTEEL_QUERY_URL, UI_AGENT_STREAM_URL, get_http_client from backend.chanakya.rt_agent.web_grounding import search_and_fetch from backend.chanakya.rt_agent.web_summariser import summarise from backend.chanakya.tatasteel.decision_framework import make_decision_framework_prompt _CHANAKYA_GENERAL_PROMPT = """\ You are Enterprise Brain, the central intelligence hub. You have four capabilities: ask_tatasteel, ask_tatasteel_decision_support, ask_ui_agent, and web_search. ━━━ CAPABILITIES ━━━ ask_tatasteel — queries the live construction project database. Use for: contracts, early warnings, compensation events, quotations, project figures, rankings, trends, counts, or any data in the project database. When both primary data and market context are needed, batch this in parallel with web_search. ask_ui_agent — ALWAYS call this after ask_tatasteel, no exceptions. Renders results as styled frontend widgets with charts and narrative. Pass market_context and storyline_hints so the response is fully enriched. web_search — searches the internet for real-time market context. Stores market_context for ask_ui_agent. When both primary data and market context are needed, batch this in parallel with ask_tatasteel. ask_tatasteel_decision_support — runs a follow-up Tata Steel analysis query aimed at action, prioritisation, delay reduction, cost reduction, tradeoff, or contractor focus. Stores decision-support data for ask_ui_agent. ━━━ WORKFLOW FOR DATABASE QUESTIONS ━━━ STEP 1 — INTENT ANALYSIS (think first, call no tools yet): a. Decode business terms in the question — what is the user really asking? Translate jargon into the underlying data question. b. Classify the question type: comparison, ranking, trend, proportion, single KPI. c. Identify the underlying decision the user needs to make — not just the data they asked for. Surface questions often mask prioritisation needs. Ask: what does this person need to act on? d. Map the question to the closest Decision Context (DC-01 through DC-09) from the Decision-Enabling Framework. Then generate exactly 5 follow-up questions anchored to that DC: - 2 from the open decision options of the matched DC — what the persona still needs to decide after receiving this answer - 2 from the required data signals of the matched DC that the primary query did not yet surface — data gaps that would complete the decision picture - 1 bridging question pointing to the adjacent DC the persona should address once the current decision is resolved All 5 must be answerable from the database and must advance the persona's mandate. Never frame NCE auto-approval as a cost or schedule risk in follow-up questions — cost and schedule impact comes only from accepted quotations (change_to_prices, change_to_days); auto-approval affects the PM's reject right, not the budget. CRITICAL — follow-up questions are shown verbatim to end users. They MUST be written in plain business language. Never include raw SQL column names (e.g. change_to_prices, change_to_days, reply_due_date, record_number, vendor, nce_number). Translate every technical term: "change_to_prices" → "cost impact", "change_to_days" → "time impact", "reply_due_date" → "response deadline", "record_number" → "reference", "vendor" → "contractor". If a question reads like a SQL query, rewrite it. e. Compose a dc_mapping_explanation (3–6 lines) that states: - Which DC was matched and why (e.g. "Mapped to DC-08 — Finance budget coverage, because the question asks how to prevent budget overruns from a Finance lens") - For each of the 5 follow-up questions, one sentence explaining which decision option or required data signal from the matched DC it comes from Pass this as dc_mapping_explanation when calling ask_ui_agent. It is rendered as a visible section so the user can see how the follow-up questions were derived. STEP 2 — Call ask_tatasteel with the primary question to get data and explanation. This first ask_tatasteel call is the main answer dataset. STEP 3 — In the SAME model turn, batch these two independent tool calls in parallel when both are needed: - ask_tatasteel for the primary answer dataset - web_search for external market / industry context Only call web_search when the question requires context the database cannot provide — industry benchmarks, market rates, contractor reputation, or sector news. Skip it when the database alone fully answers the question. STEP 4 — After ask_tatasteel and web_search complete, decide whether actionability needs extra support: - Call ask_tatasteel_decision_support zero, one, or more times with tightly-scoped follow-up analytical questions. - Do this AFTER reviewing the primary answer and any market context. STEP 5 — Call ask_ui_agent with ALL enrichment: • question: the original user question (verbatim) • market_context: optional; leave blank if web_search already stored it • storyline_hints: the 5 follow-up questions from Step 1d, pipe-delimited • dc_mapping_explanation: the DC mapping explanation from Step 1e ━━━ VISUALIZATION FOLLOW-UPS ━━━ (when user asks to change/switch/modify the chart or visualization) The previous data is already cached — do NOT call ask_tatasteel or web_search again. Call ask_ui_agent DIRECTLY with a visualization_hint that describes: - The chart type currently shown (find [Visualisation shown: ...] in prior reply) - What the user wants, e.g. 'user wants bar chart' Example: 'Currently showing contract-value-orb. User wants a different visualization.' Skipping ask_tatasteel AND web_search is REQUIRED — avoids needless round-trips. CURRENCY CONVERSION REQUESTS (e.g. 'convert to dollars', 'show in euros', 'change to rupees'): The underlying data does not change — do NOT call ask_tatasteel. Call ask_ui_agent DIRECTLY with the user's question. The UI agent will retrieve the previous rows from conversation history and re-render them in the new currency. ━━━ DIRECT ANSWERS (use no tools) ━━━ - Greetings, general conversation, help requests. - Questions about yourself or your capabilities. - Anything answerable from conversation history alone. - General knowledge not requiring project database data. NOTE: 'change the visualization', 'show as a different chart' are NOT direct answers — always use tools for these. CONVERSATION HISTORY: You have full access to prior messages. Use it to answer follow-up questions accurately without re-querying the database. """ _TATASTEEL_PERSPECTIVES = { "cost_and_timeline": """\ Focus on delivery cost and timelines as primary drivers of project success. Prioritize interventions that reduce notification cycle time and cost exposure. """, "risk_mitigation": """\ Focus on identifying and mitigating contractual and schedule risks. Prioritize early warnings, compensation events, and potential disputes before they escalate. """, "vendor_performance": """\ Focus on vendor performance metrics and accountability. Prioritize vendor compliance, delivery consistency, and quality benchmarks. """, "cash_flow": """\ Focus on cash flow optimization and financial planning. Prioritize quotation approvals, payment schedules, and working capital efficiency. """, "scope_control": """\ Focus on scope management and change control. Prioritize variation identification, scope creep prevention, and contract alignment. """, } _DEFAULT_PERSPECTIVE = "cost_and_timeline" _CHANAKYA_TATASTEEL_PROMPT = f"""\ TATA STEEL DEPLOYMENT CONTEXT: This Enterprise Brain deployment is for Tata Steel. ━━━ PERSPECTIVE ━━━ {_TATASTEEL_PERSPECTIVES[_DEFAULT_PERSPECTIVE]} ━━━ USER CONTEXT ━━━ Users span the full organisational hierarchy — board members, C-suite executives (CTO, CFO, COO, CEO), programme directors, construction management heads, finance leads, commercial managers, contract administrators, and project managers. All are familiar with NEC contract terminology and steel industry operations. NEVER define or explain NCE, EW, CE, NEC, quotation, compensation event, variation, or any standard industry term — treat all domain terminology as shared knowledge. Do not add disclaimers or jargon explanations at any level of seniority. TATA STEEL WEB SEARCH SCOPING: - Use for: industry benchmarks, Tata Steel UK news, NEC contract norms, UK construction cost trends, contractor market intelligence, steel sector news. {make_decision_framework_prompt()} ━━━ TATA STEEL INTENT EXAMPLES: - Decode NEC terms such as NCE, EW, and CE into the underlying data question. - Example follow-up lines for Tata Steel questions: Question: 'NCE variance across vendors' Storylines: • 'Which vendors have the highest NCE-to-CE conversion rate?' • 'How has NCE volume trended month-over-month for the top 3 vendors?' • 'What is the average financial value of NCEs per vendor?' • 'Which vendor cluster should be reviewed first to avoid the largest cost exposure?' • 'Which intervention would most likely reduce notification cycle time over the next month?' TATA STEEL STORYLINE FORMAT EXAMPLE: 'Which vendors have highest NCE-to-CE rate? | How has NCE volume trended? | What is average NCE value per vendor? | Which vendor cluster should be reviewed first to avoid the largest cost exposure? | Which intervention would most likely reduce notification cycle time over the next month?' PERSPECTIVE FILTERING RULE: All market context, external trends, and decision-support insights MUST be evaluated against the active PERSPECTIVE and either: a) Filtered OUT if irrelevant to the perspective focus, OR b) Explicitly LINKED to the perspective's primary drivers if included. Do not include external context (market trends, industry benchmarks, labor market) unless it directly informs or accelerates action aligned with the current perspective. """ _CHANAKYA_AGENT_PROMPT = _CHANAKYA_GENERAL_PROMPT + "\n\n" + _CHANAKYA_TATASTEEL_PROMPT # --------------------------------------------------------------------------- # Agent # --------------------------------------------------------------------------- chanakya = EnterpriseAgent( api_base=os.environ["LITELLM_PROVIDER_BASE_URL"], instructions=_CHANAKYA_AGENT_PROMPT, retries=0, logfire_env="chanakya-alpha", service_name="chanakya", deps_type=RequestContext, ) # --------------------------------------------------------------------------- # Tools # --------------------------------------------------------------------------- async def _query_tatasteel( question: str, *, user_email: str | None, conversation_id: int | None, message_history_wire: list[dict[str, str]] | None = None, ) -> tuple[dict | None, str]: try: client = get_http_client() resp = await client.post( TATASTEEL_QUERY_URL, json={ "question": question, "user_email": user_email, "conversation_id": conversation_id, "history": message_history_wire, }, ) except httpx.ConnectError: return None, "Could not reach the Tatasteel agent. Is it running?" except Exception as exc: _log.error("[ASK_TATASTEEL] HTTP error: {}", exc) return None, f"Error contacting Tatasteel agent: {exc}" if resp.status_code != 200: return None, f"Tatasteel agent returned HTTP {resp.status_code}." payload = resp.json() if payload.get("status") == "failure": err = payload.get("error", {}) return None, f"Tatasteel agent error: {err.get('message', 'unknown error')}" return payload, "" async def _run_web_search(query: str) -> tuple[str, list[dict[str, str]]]: raw = await asyncio.to_thread(search_and_fetch, query) text, claims = await asyncio.to_thread(summarise, query, raw) return text, claims @chanakya.tool async def web_search(ctx: RunContext[RequestContext], query: str) -> str: """Search the web and return grounded, source-cited market context bullets. Each bullet is backed by a verbatim quote from the search results. Claims without a verifiable source are excluded automatically. Use the returned text verbatim as market_context — do NOT re-summarise or augment it. """ progress_queue: asyncio.Queue | None = ctx.deps.progress_queue if ctx.deps else None if progress_queue is not None: await progress_queue.put({"type": "progress", "step": "searching", "message": "Reading web sources..."}) text, claims = await _run_web_search(query) # Store structured claims for the done-event explanation widget. if ctx.deps is not None: ctx.deps.web_claims = claims ctx.deps.market_context_text = text # Return the text to the LLM (empty → agent omits market_context). return text @chanakya.tool def get_my_context(ctx: RunContext[RequestContext]) -> str: "Return the calling user's identity and role from the request context." if not ctx.deps: return "user_email=(unknown), role=(unknown), tenant=(unknown)" return ( f"user_email={ctx.deps.user_email or '(unknown)'}, " f"role={ctx.deps.role or '(unknown)'}, " f"tenant={ctx.deps.tenant or '(unknown)'}" ) @chanakya.tool async def ask_tatasteel(ctx: RunContext[RequestContext], question: str) -> str: """Query the Tatasteel agent via HTTP /query and return the explanation text. Calls tatasteel synchronously to get SQL results (rows + explanation). The payload is stored in ``ctx.deps.tatasteel_result`` so ``ask_ui_agent`` can pick it up directly without a second tatasteel call. """ user_email = ctx.deps.user_email if ctx.deps else None conversation_id = ctx.deps.conversation_id if ctx.deps else None persona = (ctx.deps.role if ctx.deps and ctx.deps.role else "PM").strip() tagged_question = f"[PERSONA: {persona}] {question}" _log.debug("[ASK_TATASTEEL] question={!r} persona={!r}", question, persona) _ts_start = _time.monotonic() # Emit progress before the (potentially slow) HTTP round-trip progress_queue: asyncio.Queue | None = ctx.deps.progress_queue if ctx.deps else None if progress_queue is not None: await progress_queue.put({"type": "progress", "step": "querying", "message": "Querying Tata Steel database..."}) payload, err_msg = await _query_tatasteel( tagged_question, user_email=user_email, conversation_id=conversation_id, message_history_wire=ctx.deps.message_history_wire if ctx.deps else None, ) if payload is None: return err_msg explanation = payload.get("explanation", "(no explanation)") rows = payload.get("rows", []) sql = payload.get("generated_sql", "") _log.debug("[ASK_TATASTEEL] explanation={} chars, rows={}, sql={}", len(explanation), len(rows), sql) # Forward granular progress steps collected inside the tatasteel process if progress_queue is not None: for step_event in payload.get("progress_steps", []): await progress_queue.put(step_event) # Store payload and record flow event so server can emit a full flow log. if ctx.deps is not None: ctx.deps.tatasteel_result = [payload] if ctx.deps.tatasteel_primary_result is None: ctx.deps.tatasteel_primary_result = payload ctx.deps.tatasteel_decision_support_results.clear() ctx.deps.tatasteel_generated_sql = payload.get("generated_sql") ctx.deps.tatasteel_sql_explanation = payload.get("sql_explanation") ctx.deps.tatasteel_query_log = payload.get("query_log") or [] ctx.deps.flow_log.append({ "step": "ask_tatasteel", "question": question, "rows": len(rows), "explanation": explanation, "generated_sql": sql, "rows_all": rows, "duration_ms": round((_time.monotonic() - _ts_start) * 1000), }) return explanation @chanakya.tool async def ask_tatasteel_decision_support( ctx: RunContext[RequestContext], question: str, focus: str = "", ) -> str: """Run a follow-up Tata Steel query used only for decision support.""" user_email = ctx.deps.user_email if ctx.deps else None conversation_id = ctx.deps.conversation_id if ctx.deps else None persona = (ctx.deps.role if ctx.deps and ctx.deps.role else "PM").strip() tagged_question = f"[PERSONA: {persona}] {question}" _log.debug("[ASK_TATASTEEL_DECISION_SUPPORT] question={!r} focus={!r} persona={!r}", question, focus, persona) start = _time.monotonic() payload, err_msg = await _query_tatasteel( tagged_question, user_email=user_email, conversation_id=conversation_id, message_history_wire=ctx.deps.message_history_wire if ctx.deps else None, ) if payload is None: return err_msg rows = payload.get("rows", []) or [] explanation = payload.get("explanation", "(no explanation)") sql = payload.get("generated_sql", "") if ctx.deps is not None: ctx.deps.tatasteel_decision_support_results.append({ "question": question, "focus": focus or None, "rows": rows, "explanation": explanation, "generated_sql": sql, "sql_explanation": payload.get("sql_explanation"), "query_log": payload.get("query_log") or [], }) ctx.deps.flow_log.append({ "step": "ask_tatasteel_decision_support", "question": question, "focus": focus or None, "rows": len(rows), "explanation": explanation, "generated_sql": sql, "rows_all": rows, "duration_ms": round((_time.monotonic() - start) * 1000), }) return explanation @chanakya.tool async def ask_ui_agent( ctx: RunContext[RequestContext], visualization_hint: str = "", question: str = "", market_context: str = "", storyline_hints: str = "", dc_mapping_explanation: str = "", ) -> str: """Render the most recent tatasteel query result as a frontend widget. ALWAYS call this after ask_tatasteel — for every database question without exception. The UI agent renders charts when rows are present and formatted text cards when they are not. Never return a raw tatasteel explanation directly. The payload from the last ask_tatasteel call is forwarded automatically — you do not need to pass rows or explanation explicitly. Args: visualization_hint: Optional hint about preferred chart style (e.g. "bar chart", "show as table", "trend line"). Leave empty to let the UI agent pick the best fit automatically. question: The current user question. Pass this verbatim so the UI agent receives the exact wording (especially for follow-up viz requests like "change the chart"). market_context: Bullet-point summary of web search results — industry benchmarks, Tata Steel UK news, NEC norms, market trends. Pass verbatim from your web_search summary (150–300 words). storyline_hints: Pipe-delimited list of 5 follow-up enquiry questions derived from the matched Decision Context (DC-01 through DC-09). dc_mapping_explanation: Plain-text explanation (3–6 lines) stating which DC was matched and why each of the 5 follow-up questions was derived from that DC's decision options or required data signals. This is rendered as a visible "How follow-ups were derived" section in the UI. """ if ctx.deps and ctx.deps.tatasteel_primary_result: payload = ctx.deps.tatasteel_primary_result.copy() elif ctx.deps and ctx.deps.tatasteel_result: payload = ctx.deps.tatasteel_result[0].copy() else: # No fresh tatasteel data (e.g. visualization-only follow-up that skipped # ask_tatasteel). Send empty rows — the UI agent will use conversation history # to re-render with a different chart type. payload = {"rows": [], "explanation": "", "question": ""} user_email = ctx.deps.user_email if ctx.deps else None conversation_id = ctx.deps.conversation_id if ctx.deps else None # Resolve final question string here so the server needs no post-processing. base_question = question or payload.get("question", "") final_question = ( f"{base_question}\n[Visualization context: {visualization_hint}]" if visualization_hint else base_question ) # Parse pipe-delimited storyline hints into a list. parsed_storylines = ( [s.strip() for s in storyline_hints.split("|") if s.strip()] if storyline_hints else [] ) resolved_market_context = market_context or (ctx.deps.market_context_text if ctx.deps else "") # Currency is resolved by the UI agent from the question text — pass the payload default as-is. currency_code = payload.get("currency_code", "GBP") ui_request = { "rows": payload.get("rows", []), "explanation": payload.get("explanation", ""), "question": final_question, "user_email": user_email, "conversation_id": conversation_id, "currency_code": currency_code, "market_context": resolved_market_context or None, "storyline_hints": parsed_storylines or None, "dc_explanation": dc_mapping_explanation or None, "history": ctx.deps.message_history_wire if ctx.deps else None, "decision_support_context": ( [ {k: v for k, v in p.items() if k not in ("generated_sql", "sql_explanation", "query_log")} for p in ctx.deps.tatasteel_decision_support_results ] if ctx.deps and ctx.deps.tatasteel_decision_support_results else None ), "what_to_act_on": payload.get("what_to_act_on") or None, "urgency": payload.get("urgency") or None, "cost_of_inaction": payload.get("cost_of_inaction") or None, } progress_queue: asyncio.Queue | None = ctx.deps.progress_queue if ctx.deps else None if progress_queue is not None: await progress_queue.put({"type": "progress", "step": "rendering", "message": "Preparing response..."}) row_count = len(payload.get("rows", [])) primary_rows = payload.get("rows", []) _log.debug("[ASK_UI_AGENT] calling UI agent stream, rows={}", row_count) _ui_start = _time.monotonic() try: client = get_http_client() async with client.stream("POST", UI_AGENT_STREAM_URL, json=ui_request) as resp: if resp.status_code != 200: err = {"type": "error", "message": f"UI agent returned {resp.status_code}", "code": "PIPELINE_ERROR"} if progress_queue is not None: await progress_queue.put(err) return f"UI agent returned HTTP {resp.status_code}." sse_buffer = "" async for chunk in resp.aiter_text(): if not chunk: continue sse_buffer += chunk while "\n\n" in sse_buffer: sse_msg, sse_buffer = sse_buffer.split("\n\n", 1) for line in sse_msg.split("\n"): if line.startswith("data: "): try: parsed = json.loads(line[6:]) if parsed.get("type") == "done" and ctx.deps: done_data = list(parsed.get("data") or []) # Inject sql_query widgets with rows query_log = ctx.deps.tatasteel_query_log or [] if query_log: for idx, entry in enumerate(query_log): done_data.append({ "type": "sql_query", "content": entry.get("sql", ""), "explanation": entry.get("explanation") or None, "rows": primary_rows if idx == 0 else [], }) elif ctx.deps.tatasteel_generated_sql: done_data.append({ "type": "sql_query", "content": ctx.deps.tatasteel_generated_sql, "explanation": ctx.deps.tatasteel_sql_explanation, "rows": primary_rows, }) decision_support = ctx.deps.tatasteel_decision_support_results or [] for packet in decision_support: packet_question = packet.get("question", "") packet_explanation = packet.get("sql_explanation") or packet.get("explanation") or None packet_rows = packet.get("rows") or [] packet_logs = packet.get("query_log") or [] if packet_logs: last_idx = len(packet_logs) - 1 for log_idx, entry in enumerate(packet_logs): expl = entry.get("explanation") or packet_explanation or "" if packet_question: expl = f"Decision support — {packet_question} {expl}".strip() done_data.append({ "type": "sql_query", "content": entry.get("sql", ""), "explanation": expl or None, "rows": packet_rows if log_idx == last_idx else [], }) elif packet.get("generated_sql"): expl = packet_explanation or "" if packet_question: expl = f"Decision support — {packet_question} {expl}".strip() done_data.append({ "type": "sql_query", "content": packet.get("generated_sql", ""), "explanation": expl or None, "rows": packet_rows, }) # Append decision context from decision support explanations if decision_support: ds_parts = [ p.get("explanation", "").strip() for p in decision_support if p.get("explanation", "").strip() ] if ds_parts: done_data.append({ "type": "decision_context", "content": "\n\n".join(ds_parts), }) if ctx.deps.web_claims: done_data.append({"type": "explanation", "content": ctx.deps.web_claims}) parsed = {**parsed, "data": done_data} if progress_queue is not None: await progress_queue.put(parsed) if parsed.get("type") == "done" and ctx.deps: ctx.deps.ui_result = parsed except Exception: pass except httpx.ConnectError: err = {"type": "error", "message": "Could not reach the UI agent.", "code": "PIPELINE_ERROR"} if progress_queue is not None: await progress_queue.put(err) return "Could not reach the UI agent. Is it running?" except Exception as exc: _log.error("[ASK_UI_AGENT] error: {}", exc) err = {"type": "error", "message": str(exc), "code": "PIPELINE_ERROR"} if progress_queue is not None: await progress_queue.put(err) return f"Error contacting UI agent: {exc}" widget_count = len(ctx.deps.ui_result.get("data", [])) if ctx.deps and ctx.deps.ui_result else 0 # Add a sample of widgets and market context to the flow log for debugging widgets_all: list[dict] = [] if ctx.deps and ctx.deps.ui_result: widgets_all = ctx.deps.ui_result.get("data", []) viz_type = next( (w.get("content", {}).get("type", "") for w in widgets_all if w.get("type") == "visualization"), "", ) if ctx.deps is not None: ctx.deps.flow_log.append({ "step": "ask_ui_agent", "visualization_hint": visualization_hint or None, "question": final_question, "market_context": resolved_market_context or None, "storylines": parsed_storylines, "ui_request_forwarded": ui_request, "input_rows": row_count, "output_widgets": widget_count, "viz_type": viz_type or None, "widgets_all": widgets_all, "duration_ms": round((_time.monotonic() - _ui_start) * 1000), }) return f"Visualization generated with {widget_count} widget(s) for {row_count} data rows."