"""Stage 7 orchestrator — the AgentLoop ties everything together. Each user utterance is processed by a single `handle_turn(...)` call which: 1. Routes the utterance (continue / switch / question / correction / offtopic / terminate). 2. Extracts slots, classifies intent, or answers a question as appropriate. 3. Updates AgentState. 4. Runs the deterministic resolver against the new state. 5. Decides the next action (ask slot, confirm lock, fire APIs, redirect, …). 6. Generates the warm conversational reply. 7. Emits trace events at every decision. The returned TurnResult contains the agent's reply, UI hint, and the full trace for this turn. """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Optional from backend.agent.executor import ExecutionResult, execute, execution_to_dict from backend.agent.kb import JourneyKB, KnowledgeBase from backend.agent.llm_layer import ( answer_question, classify_intent, extract_slots, generate_reply, route, ) from backend.agent.resolver import resolve, ResolverResult from backend.agent.state import AgentState from backend.agent.trace import TraceEvent, TraceRecorder @dataclass class TurnResult: reply: str ui_hint: dict trace_events: list[dict] = field(default_factory=list) state_snapshot: dict = field(default_factory=dict) terminated: bool = False execution: Optional[dict] = None # set when APIs fire # ============================================================================ # Helpers # ============================================================================ def _last_assistant(state: AgentState) -> str: for t in reversed(state.turns): if t.role == "assistant": return t.text return "" def _recent_turns(state: AgentState, n: int = 4) -> list[dict]: return [{"role": t.role, "text": t.text} for t in state.turns[-n:]] def _slot_value_label(journey_kb: JourneyKB, slot: str, value: Any) -> Any: """If `slot` is a dropdown with a known vocab and `value` matches an enum, return the label. Otherwise return the value verbatim. (Placeholder until we have human labels per code.)""" return value def _build_slot_question_details(journey_kb: JourneyKB, slot: str) -> dict: """Compose the UI hint payload for asking about `slot`.""" ctrl = journey_kb.slots_by_btsynonym.get(slot.lower(), {}) vocab = journey_kb.slot_value_vocab.get(slot.lower(), []) kind = ctrl.get("classification", "basic") dtype = ctrl.get("datatype", "char") label = slot # TODO: pull a human label from somewhere return { "slot": slot, "label": label, "kind": kind, "datatype": dtype, "options": vocab, "ui_kind": ( "dropdown" if (kind == "dropdown" and vocab) else "date" if dtype == "date" else "text" ), } # Human-friendly labels for the slots that appear in the multi-slot form. The kb's btsynonym # names (qtyml, needdateml, drpshpidml, …) leak Ramco-internal naming that real users don't # recognise. This map turns them into natural labels for form fields. _SLOT_LABEL = { "potypeenum": "PO type", "supplier_code": "Supplier", "buyerhdr": "Buyer", "currencycode": "Currency", "item_codeml": "Item", "qtyml": "Quantity", "cost": "Unit cost", "warehouse_code_mul": "Warehouse", "customercode_ml": "Customer", "needdateml": "Need date", "refsono": "Reference doc no.", "adhocitemclassml": "Ad-hoc item class", } # How each form field is rendered when there's no master-data dropdown. _SLOT_KIND = { "qtyml": "number", "cost": "number", "needdateml": "date", } # Friendly dropdown options for enums that aren't in master_data_stub but are vocab-constrained. _SLOT_ENUM_OPTIONS = { # PO type code 1/2/3/4 → user-readable labels (extractor canonicalises back to digits) "potypeenum": ["General", "Capital", "Dropship", "Consignment"], } def _missing_priority_slots(journey_kb: JourneyKB, choices: dict) -> list[str]: """The set of conversational-priority slots NOT yet filled, in their natural order. Used by `_reply_ask_slot` to decide single-ask vs multi-slot form.""" # Import here to avoid circular import with resolver. from backend.agent.resolver import _conversational_priority, _user_fillable_slots fillable = _user_fillable_slots(journey_kb) out: list[str] = [] for s in _conversational_priority(choices): if s.lower() in choices: continue if fillable and s.lower() not in fillable: continue out.append(s.lower()) return out def _build_multi_slot_form(journey_kb: JourneyKB, slots_to_ask: list[str]) -> dict: """Return a ui_hint for a multi-field form. For each slot we pick the best control: - Master-data lookup (supplier, item, buyer, customer, warehouse, currency) → dropdown populated from kb//master_data_stub.json. - Enumerated code (potypeenum) → dropdown with human labels. - Numeric (qty, cost) → number input. - Date (need date) → date picker. - Everything else → text input. """ master = (journey_kb.master_data or {}).get("tables") or {} fields: list[dict] = [] for s in slots_to_ask: sl = s.lower() label = _SLOT_LABEL.get(sl, sl) # 1) Master-data-backed dropdown if sl in master: options = list(master[sl].get("valid_values") or []) fields.append({"slot": sl, "label": label, "kind": "dropdown", "options": options}) continue # 2) Enum override (potypeenum) if sl in _SLOT_ENUM_OPTIONS: fields.append({"slot": sl, "label": label, "kind": "dropdown", "options": _SLOT_ENUM_OPTIONS[sl]}) continue # 3) Specific kind override (number / date) if sl in _SLOT_KIND: fields.append({"slot": sl, "label": label, "kind": _SLOT_KIND[sl]}) continue # 4) Default to text fields.append({"slot": sl, "label": label, "kind": "text"}) return { "kind": "multi_slot_form", "title": "Fill in the remaining PO details", "fields": fields, } # ============================================================================ # Main agent loop # ============================================================================ class AgentLoop: """Stateful conversation handler. Use it as: loop.handle_turn(user_msg) for each turn.""" def __init__(self, kb: KnowledgeBase, *, state: AgentState | None = None, fire_apis: bool = True): self.kb = kb self.state = state or AgentState(submodule=kb.submodule) self.trace = TraceRecorder() self.fire_apis = fire_apis # Per-turn ephemeral: vocab-rejected slot values from the LATEST slot extraction. # Cleared at the start of every handle_turn(); set after each extract_slots() call. # Carried into the next reply's plan_details so the agent can acknowledge "I couldn't # recognise X" rather than silently re-asking. self._turn_rejected_slots: list[dict] = [] # ----- public API ------------------------------------------------------ def handle_turn(self, user_message: str) -> TurnResult: turn_idx = self.trace.begin_turn() self.state.add_turn("user", user_message) # Reset per-turn ephemerals self._turn_rejected_slots = [] try: if not self.state.current_journey: result = self._handle_no_journey(user_message) else: result = self._handle_in_journey(user_message) except Exception as e: self.trace.error("agent_exception", f"Unhandled error: {type(e).__name__}: {e}") result = TurnResult( reply="Sorry — I hit an unexpected hiccup. Could you say that again?", ui_hint={"kind": "text"}, ) # Record assistant turn and attach trace self.state.add_turn("assistant", result.reply, ui_hint=result.ui_hint) result.trace_events = [e.to_dict() for e in self.trace.for_turn(turn_idx)] result.state_snapshot = self.state.to_dict() result.terminated = self.state.terminated self.trace.end_turn(f"replied: {result.reply[:80]}") return result # ----- branches -------------------------------------------------------- def _handle_no_journey(self, user_message: str) -> TurnResult: """No journey active yet — try to map the utterance to a P2P journey. If there are suspended journeys, also check whether the user is asking to RESUME one before going into intent classification.""" # Resume check: if any journey is suspended, route the message to see if the user is # asking to go back to it. if self.state.suspended_journeys: snapshot = { "current_journey": None, "slots": {}, "last_assistant": _last_assistant(self.state), "suspended_journeys": [sj.activity_desc for sj in self.state.suspended_journeys], } r_decision = route(user_message, snapshot, self.kb) self.trace.emit( "route_decided", f"action={r_decision.action} (no-journey routing, suspended exist)", action=r_decision.action, confidence=r_decision.confidence, reasoning=r_decision.reasoning, target_journey=r_decision.target_journey, ) if r_decision.action == "resume": return self._handle_resume(user_message, r_decision.target_journey) self.trace.emit("phase", "intent_classification_no_journey") # If we have earlier turns (because turn 1 mis-classified), pass them so the classifier # can recover with accumulated context. convo = [{"role": t.role, "text": t.text} for t in self.state.turns[:-1]] decision = classify_intent( user_message, self.kb, conversation_so_far=convo if convo else None, ) self.trace.emit( "intent_classified", f"intent: {decision.journey} ({decision.confidence})", journey=decision.journey, confidence=decision.confidence, reasoning=decision.reasoning, extracted_slots=dict(decision.extracted_slots), raw=decision.raw, ) # If intent classifier returned nothing OR matched an unavailable journey, emit a # synthetic route_decided so the trace shows the action consistently. if not decision.journey: self.trace.emit( "route_decided", f"action=offtopic (no journey matched)", action="offtopic", confidence=decision.confidence, reasoning=decision.reasoning, ) elif decision.journey not in self.kb.available_journeys(): self.trace.emit( "route_decided", f"action=offtopic (journey '{decision.journey}' not yet implemented)", action="offtopic", confidence=decision.confidence, reasoning=f"recognised but unavailable: {decision.journey}", recognised_journey=decision.journey, ) if decision.journey and decision.journey in self.kb.available_journeys(): # Successful first-turn intent classification — emit a synthetic route_decided # so the trace shows action=continue consistently for downstream consumers. self.trace.emit( "route_decided", f"action=continue (intent matched: {decision.journey})", action="continue", confidence=decision.confidence, reasoning=f"first turn matched journey {decision.journey}", ) self.state.current_journey = decision.journey # Slot extraction is delegated to the dedicated slot_extractor (intent.extracted_slots # is intentionally empty — see classify_intent). The slot extractor uses the journey's # full slot vocab + canonical-key normaliser, so we don't pollute state with raw # natural-language keys like "supplier_name". journey_kb = self.kb.journey(decision.journey) focus = list(journey_kb.gating_slots[:8]) extraction = extract_slots( user_message=user_message, journey_kb=journey_kb, focus_slots=focus, recent_facts=self.state.recent_facts, ) self.trace.emit( "slots_extracted", f"got {len(extraction.extracted_slots)} slot(s) on first turn" + (f", rejected {len(extraction.rejected_slots)}" if extraction.rejected_slots else ""), extracted=extraction.extracted_slots, unset=extraction.unset_slots, rejected=extraction.rejected_slots, user_intent_summary=extraction.user_intent_summary, ) self._turn_rejected_slots = list(extraction.rejected_slots) for r in extraction.rejected_slots: self.trace.warn("slot_rejected", f"dropped {r['slot']}={r['value']!r}: {r['reason']}", slot=r['slot'], value=r['value'], reason=r['reason']) for k, v in extraction.extracted_slots.items(): if not self.state.has_slot(k): self.state.set_slot(k, v) for k in extraction.unset_slots: self.state.unset_slot(k) return self._advance(user_message) # Journey not available or no match — clarifying / redirect if decision.journey and decision.journey not in self.kb.available_journeys(): self.trace.warn("journey_unavailable", f"Recognised intent {decision.journey}, but not yet implemented", journey=decision.journey) plan_kind = "clarify_intent" plan_details = { "reason": "journey_recognised_but_unavailable", "recognised_journey": decision.journey, "available_journeys": self.kb.available_journeys(), } elif decision.candidate_journeys: # Ambiguous — the LLM lists the candidate journeys it considered. Present them. self.trace.emit("clarify_required", "ambiguous intent — asking user to choose", candidates=decision.candidate_journeys) plan_kind = "clarify_ambiguous" plan_details = { "candidate_journeys": decision.candidate_journeys, "clarification_question": decision.clarification_question, "available_journeys": self.kb.available_journeys(), } else: plan_kind = "clarify_intent" plan_details = { "reason": "no_journey_match", "clarification_question": decision.clarification_question, "available_journeys": self.kb.available_journeys(), } reply = generate_reply( plan_kind=plan_kind, plan_details=plan_details, state_snapshot={ "current_journey": None, "recent_turns": _recent_turns(self.state), }, user_message=user_message, ) # If the LLM was ambiguous AND named candidate journeys, surface them as a dropdown. ui_hint = reply.ui_hint if plan_kind == "clarify_ambiguous" and decision.candidate_journeys: ui_hint = { "kind": "dropdown", "slot": "_journey_choice", "label": "Which would you like?", "options": decision.candidate_journeys, } self.trace.emit("reply_generated", f"plan={plan_kind}", ui_hint=ui_hint) return TurnResult(reply=reply.reply, ui_hint=ui_hint) def _handle_in_journey(self, user_message: str) -> TurnResult: """A journey is active — route + handle the user's move.""" snapshot = { "current_journey": self.state.current_journey, "slots": dict(self.state.slots), "variant_locked": self.state.variant_locked, "last_assistant": _last_assistant(self.state), "suspended_journeys": [sj.activity_desc for sj in self.state.suspended_journeys], } decision = route(user_message, snapshot, self.kb) self.trace.emit( "route_decided", f"action={decision.action} ({decision.confidence}) — {decision.reasoning}", action=decision.action, confidence=decision.confidence, target_journey=decision.target_journey, question_topic=decision.question_topic, raw=decision.raw, ) if decision.action == "offtopic": return self._reply_offtopic(user_message) if decision.action == "question": return self._handle_question(user_message, decision.question_topic) if decision.action == "switch": return self._handle_switch(user_message, decision.target_journey) if decision.action == "resume": return self._handle_resume(user_message, decision.target_journey) if decision.action == "abandon": return self._handle_abandon(user_message) if decision.action == "terminate": return self._handle_terminate(user_message) # continue or correction → extract slots and advance return self._handle_continue_or_correction(user_message, is_correction=(decision.action == "correction")) # ----- branch handlers ------------------------------------------------- def _reply_offtopic(self, user_message: str) -> TurnResult: # If a journey is active, the redirect should name it and the next-asked slot — so the # user understands EXACTLY where we're picking up. Otherwise it's a generic redirect. next_slot = None next_slot_reason = None if self.state.current_journey: try: jkb = self.kb.journey(self.state.current_journey) res = resolve(jkb, self.state.slots) next_slot = res.next_slot_to_ask next_slot_reason = res.next_slot_reason except Exception: pass reply = generate_reply( plan_kind="redirect_offtopic", plan_details={ "current_journey": self.state.current_journey, "filled_slots": dict(self.state.slots), "next_slot": next_slot, "next_slot_reason": next_slot_reason, }, state_snapshot={ "current_journey": self.state.current_journey, "recent_turns": _recent_turns(self.state), }, user_message=user_message, ) self.trace.emit("reply_generated", "plan=redirect_offtopic", ui_hint=reply.ui_hint) return TurnResult(reply=reply.reply, ui_hint=reply.ui_hint) def _handle_question(self, user_message: str, question_topic: Optional[str]) -> TurnResult: journey_kb = self.kb.journey(self.state.current_journey) # type: ignore[arg-type] # Recompute the resolver to know what we were about to ask res = resolve(journey_kb, self.state.slots) qa = answer_question( user_message=user_message, journey_kb=journey_kb, filled_slots=dict(self.state.slots), next_slot=res.next_slot_to_ask, question_topic=question_topic, ) self.trace.emit("qa_answered", f"topic={question_topic}", answer=qa.answer) # Compose reply: answer + resume prompt text = f"{qa.answer}\n\n{qa.resume_prompt}".strip() ui_hint: dict[str, Any] if res.next_slot_to_ask: details = _build_slot_question_details(journey_kb, res.next_slot_to_ask) ui_hint = { "kind": details["ui_kind"], "slot": details["slot"], "label": details["label"], "options": details["options"], } else: ui_hint = {"kind": "text"} return TurnResult(reply=text, ui_hint=ui_hint) def _handle_switch(self, user_message: str, target_journey: Optional[str]) -> TurnResult: if not target_journey: return self._reply_offtopic(user_message) prev_journey = self.state.current_journey # If the previous journey already terminated successfully, there's nothing meaningful # to suspend — just clear it instead of pushing dead state onto the suspended stack. # Otherwise suspend so the user can `resume` later. if self.state.terminated: self.state.current_journey = None self.state.slots = {} self.state.variant_locked = False self.state.locked_at_turn = None self.trace.emit("journey_cleared_after_terminate", f"cleared {prev_journey} (already terminated) before considering switch to {target_journey}", from_journey=prev_journey, to_journey=target_journey) else: self.state.suspend_current_journey(reason=f"user wants to switch to {target_journey}") self.trace.emit("journey_suspended", f"suspended {prev_journey} in favor of {target_journey}", from_journey=prev_journey, to_journey=target_journey) if target_journey in self.kb.available_journeys(): self.state.current_journey = target_journey # Switching to a new fresh journey resets the terminated flag for THIS new journey. self.state.terminated = False self.state.termination_reason = None self.trace.emit("journey_started", f"journey -> {target_journey}", journey=target_journey) return self._advance(user_message) # Switching to an unavailable journey — honestly tell the user. plan_details = { "target_journey": target_journey, "previous_journey": prev_journey, "available_journeys": self.kb.available_journeys(), "previous_terminated": self.state.terminated, } reply = generate_reply( plan_kind="switch_journey", plan_details=plan_details, state_snapshot={"current_journey": None, "recent_turns": _recent_turns(self.state)}, user_message=user_message, ) self.trace.emit("reply_generated", "plan=switch_journey (unavailable)", ui_hint=reply.ui_hint) return TurnResult(reply=reply.reply, ui_hint=reply.ui_hint) def _handle_resume(self, user_message: str, target_journey: Optional[str]) -> TurnResult: """User wants to return to a previously-suspended journey.""" if not self.state.suspended_journeys: self.trace.warn("resume_no_suspended", "user asked to resume but nothing is suspended") return self._reply_offtopic(user_message) # If target_journey is None or not in suspended, pick the most recently suspended one tgt = target_journey if not tgt or not any(sj.activity_desc == tgt for sj in self.state.suspended_journeys): tgt = self.state.suspended_journeys[-1].activity_desc # Save current state (if any) before resuming if self.state.current_journey and self.state.current_journey != tgt: self.state.suspend_current_journey(reason="resume to previous journey") ok = self.state.restore_journey(tgt) self.trace.emit("journey_resumed", f"resumed {tgt} (ok={ok})", journey=tgt, ok=ok, slots_restored=list(self.state.slots.keys())) if not ok: return self._reply_offtopic(user_message) return self._advance(user_message) def _handle_abandon(self, user_message: str) -> TurnResult: """User wants to step away from the current journey. Crucially, we do NOT wipe the slots — we SUSPEND the journey onto the suspended stack. That way if the user later says "go back" / "actually let's continue", _handle_resume can restore everything they had filled. This matches how real conversations work: "forget it" usually means "let me set this aside", not "burn the work I just did". If the user genuinely wants a hard reset, they can send the WS `reset` message or say "let's start over from scratch" (which the router will treat as a fresh first turn after the suspend). """ prev_journey = self.state.current_journey prev_slot_count = len(self.state.slots) # suspend_current_journey() pushes onto suspended_journeys AND clears current_journey # + slots + variant_locked. That's exactly what we want. if self.state.current_journey: self.state.suspend_current_journey(reason="user said abandon/never mind") self.trace.emit("journey_abandoned", f"user abandoned {prev_journey} ({prev_slot_count} slot(s)); " f"preserved on suspended stack for possible resume", previous_journey=prev_journey, slot_count=prev_slot_count, suspended_count=len(self.state.suspended_journeys)) reply = generate_reply( plan_kind="abandon", plan_details={ "previous_journey": prev_journey, "slot_count": prev_slot_count, "can_resume": prev_slot_count > 0, # tell the LLM to offer "go back" if work was done "available_journeys": self.kb.available_journeys(), }, state_snapshot={"current_journey": None, "recent_turns": _recent_turns(self.state)}, user_message=user_message, ) return TurnResult(reply=reply.reply, ui_hint={"kind": "text"}) def _handle_terminate(self, user_message: str) -> TurnResult: journey_kb = self.kb.journey(self.state.current_journey) # type: ignore[arg-type] res = resolve(journey_kb, self.state.slots) self.trace.emit("terminate_requested", f"locked_status={res.locked_status}", locked_status=res.locked_status, violation_count=len(res.active_violations), pending_slots=res.slots_pending) if not res.locked: # Cannot terminate yet — fall back to advance behaviour, which will surface what's missing return self._advance(user_message) return self._fire_apis(journey_kb, res, user_message) def _handle_continue_or_correction(self, user_message: str, *, is_correction: bool) -> TurnResult: journey_kb = self.kb.journey(self.state.current_journey) # type: ignore[arg-type] res_before = resolve(journey_kb, self.state.slots) focus = [res_before.next_slot_to_ask] if res_before.next_slot_to_ask else [] # also include the top requirements as "interested" slots for req in res_before.active_requirements[:5]: if req.slot not in focus: focus.append(req.slot) extraction = extract_slots( user_message=user_message, journey_kb=journey_kb, focus_slots=focus, recent_facts=self.state.recent_facts, ) # Plain-English summary: what we PULLED OUT of this turn. extracted_keys = list(extraction.extracted_slots.keys()) if not extracted_keys and not extraction.unset_slots and not extraction.rejected_slots: ext_summary = "Nothing new extracted from this turn" else: parts = [] if extracted_keys: shown = ", ".join(extracted_keys[:4]) more = f" (+{len(extracted_keys)-4} more)" if len(extracted_keys) > 4 else "" parts.append(f"Captured {len(extracted_keys)} slot(s): {shown}{more}") if extraction.unset_slots: parts.append(f"Cleared: {', '.join(extraction.unset_slots)}") if extraction.rejected_slots: rej = ", ".join(f"{r['slot']}={r['value']!r}" for r in extraction.rejected_slots) parts.append(f"Dropped (failed vocab): {rej}") ext_summary = " · ".join(parts) self.trace.emit("slots_extracted", ext_summary, extracted=extraction.extracted_slots, unset=extraction.unset_slots, rejected=extraction.rejected_slots, uncertain=extraction.uncertain_slots, user_intent_summary=extraction.user_intent_summary) # Record vocab-rejected values so the next reply can acknowledge them self._turn_rejected_slots = list(extraction.rejected_slots) for r in extraction.rejected_slots: self.trace.warn("slot_rejected", f"dropped {r['slot']}={r['value']!r}: {r['reason']}", slot=r['slot'], value=r['value'], reason=r['reason']) # Apply the extracted slots. R4: always overwrite — the slot extractor only emits # values that the user explicitly mentioned in this turn (per its prompt), so trusting # them is structurally safer than guarding "set only if new" (which silently dropped # user revisions whenever the router didn't recognise correction-phrasing). # is_correction is preserved as a TRACE LABEL but no longer gates the write. for k, v in extraction.extracted_slots.items(): self.state.set_slot(k, v) # Apply unsets — user explicitly asked to clear these for k in extraction.unset_slots: removed = self.state.unset_slot(k) if removed: self.trace.emit("slot_unset", f"cleared slot '{k}' on user request", slot=k) # Update recent facts (small heuristic: keep latest value of common back-reference slots) for ref_key in ("supplier_code", "buyerhdr", "currencycode", "potypeenum"): if ref_key in extraction.extracted_slots: self.state.recent_facts[ref_key] = extraction.extracted_slots[ref_key] return self._advance(user_message) # ----- the central advance() ------------------------------------------ def _advance(self, user_message: str) -> TurnResult: """After state update, decide what to do next: ask, lock, or fire APIs.""" journey_kb = self.kb.journey(self.state.current_journey) # type: ignore[arg-type] res = resolve(journey_kb, self.state.slots) # Human-readable summary for the trace pane — surfaces what the resolver decided in # plain English, instead of the internal field names. if res.locked_status == "locked": summary = "Ready to create — all required slots filled, no blocking issues" elif res.locked_status == "blocked_by_violation": first = res.active_violations[0] if res.active_violations else None err = (first.error_message or "(no message)") if first else "(unknown rule)" summary = f"Blocked: {err[:120]}" elif res.locked_status == "missing_required": pending_str = ", ".join(res.slots_pending[:3]) more = f" (+{len(res.slots_pending)-3} more)" if len(res.slots_pending) > 3 else "" summary = f"Still need {len(res.slots_pending)} field(s): {pending_str}{more}" elif res.locked_status == "no_choices_yet": summary = "Waiting for the first slot value" else: summary = res.locked_status # Note: deferred_violations are intentionally NOT mentioned in the headline summary. # They're master-data lookups (does supplier X exist? is item Y effective?) checked at # API time, not user-actionable. The count is still in event.data.deferred_violations # for anyone inspecting the trace event, but doesn't clutter the conversational summary. self.trace.emit( "resolved", summary, locked_status=res.locked_status, next_slot=res.next_slot_to_ask, next_slot_reason=res.next_slot_reason, slots_filled=res.slots_filled, slots_pending=res.slots_pending, violations=[{ "rule_kind": v.rule_kind, "sp_name": v.sp_name, "error_code": v.error_code, "error_message": v.error_message, "antecedents": v.antecedents, "severity": v.severity, } for v in res.active_violations], deferred_violations=[{ "rule_kind": v.rule_kind, "sp_name": v.sp_name, "error_code": v.error_code, "error_message": v.error_message, "severity": v.severity, } for v in res.deferred_violations], ) if res.active_violations: return self._reply_violation(journey_kb, res, user_message) if res.next_slot_to_ask: return self._reply_ask_slot(journey_kb, res, user_message) if res.locked: # Variant locked — but the user hasn't asked to terminate yet. Confirm before firing. return self._reply_confirm_lock(journey_kb, res, user_message) # No more to ask, not locked — odd state; redirect return self._reply_offtopic(user_message) def _reply_violation(self, journey_kb: JourneyKB, res: ResolverResult, user_message: str) -> TurnResult: v = res.active_violations[0] plan_details = { "violation_kind": v.rule_kind, "antecedents": v.antecedents, "error_message": v.error_message, "filled_slots": res.slots_filled, } reply = generate_reply( plan_kind="violation_block", plan_details=plan_details, state_snapshot={ "current_journey": self.state.current_journey, "recent_turns": _recent_turns(self.state), }, user_message=user_message, ) self.trace.emit("reply_generated", "plan=violation_block", error_message=v.error_message, ui_hint=reply.ui_hint) return TurnResult(reply=reply.reply, ui_hint=reply.ui_hint) def _reply_ask_slot(self, journey_kb: JourneyKB, res: ResolverResult, user_message: str) -> TurnResult: # Decide between SINGLE-SLOT ASK and MULTI-SLOT FORM. Heuristic: if 3+ priority slots # remain unfilled, the form is much faster than asking one at a time — the user # clicks dropdowns / fills fields and submits a structured message. priority_remaining = _missing_priority_slots(journey_kb, self.state.slots) if len(priority_remaining) >= 3: return self._reply_ask_multi(journey_kb, res, priority_remaining, user_message) details = _build_slot_question_details(journey_kb, res.next_slot_to_ask) # type: ignore[arg-type] details["reason"] = res.next_slot_reason details["slots_remaining"] = res.slots_pending[:8] # If the user just gave us a value that failed vocab validation, surface it so the # reply can acknowledge "I couldn't recognise X" instead of silently re-asking. if self._turn_rejected_slots: details["rejected_in_last_turn"] = list(self._turn_rejected_slots) reply = generate_reply( plan_kind="ask_slot", plan_details=details, state_snapshot={ "current_journey": self.state.current_journey, "recent_turns": _recent_turns(self.state), }, user_message=user_message, ) # Force the UI hint to a structured form when we have options/dates ui_hint: dict[str, Any] if details["ui_kind"] == "dropdown" and details["options"]: ui_hint = { "kind": "dropdown", "slot": details["slot"], "label": details["label"], "options": details["options"], } elif details["ui_kind"] == "date": ui_hint = {"kind": "date", "slot": details["slot"], "label": details["label"]} else: ui_hint = {"kind": "text", "slot": details["slot"], "label": details["label"]} self.trace.emit("reply_generated", f"Asking the user for: {details['slot']}", ui_hint=ui_hint, slot_detail=details) return TurnResult(reply=reply.reply, ui_hint=ui_hint) def _reply_ask_multi( self, journey_kb: JourneyKB, res: ResolverResult, priority_remaining: list[str], user_message: str, ) -> TurnResult: """Surface a multi-slot form instead of asking one slot at a time.""" form = _build_multi_slot_form(journey_kb, priority_remaining) plan_details = { "remaining_slots": priority_remaining, "slot_labels": {f["slot"]: f["label"] for f in form["fields"]}, "already_filled": res.slots_filled, "po_type_known": "potypeenum" not in priority_remaining, } if self._turn_rejected_slots: plan_details["rejected_in_last_turn"] = list(self._turn_rejected_slots) reply = generate_reply( plan_kind="ask_multi", plan_details=plan_details, state_snapshot={ "current_journey": self.state.current_journey, "recent_turns": _recent_turns(self.state), }, user_message=user_message, ) self.trace.emit( "reply_generated", f"Showing the multi-slot form for: {', '.join(priority_remaining[:5])}" + (f" (+{len(priority_remaining)-5} more)" if len(priority_remaining) > 5 else ""), ui_hint=form, field_count=len(form["fields"]), ) return TurnResult(reply=reply.reply, ui_hint=form) def _reply_confirm_lock(self, journey_kb: JourneyKB, res: ResolverResult, user_message: str) -> TurnResult: plan_details = { "filled_slots": res.slots_filled, "active_defaults": res.active_defaults[:6], } reply = generate_reply( plan_kind="confirm_lock", plan_details=plan_details, state_snapshot={ "current_journey": self.state.current_journey, "recent_turns": _recent_turns(self.state), }, user_message=user_message, ) self.trace.emit("variant_locked", f"variant locked; awaiting user confirmation to fire APIs", filled_slots=res.slots_filled) self.state.variant_locked = True return TurnResult( reply=reply.reply, ui_hint={"kind": "confirm", "prompt": "Create the PO?"}, ) def _fire_apis(self, journey_kb: JourneyKB, res: ResolverResult, user_message: str) -> TurnResult: self.trace.emit("fire_apis_begin", f"firing APIs for journey '{journey_kb.meta.activity_desc}'") exec_result = execute(journey_kb, self.state.slots, fire=self.fire_apis) exec_dict = execution_to_dict(exec_result) self.state.fired_apis.append(exec_dict) self.trace.emit( "fire_apis_done" if exec_result.success else "fire_apis_failed", f"success={exec_result.success} calls={len(exec_result.calls)}", execution=exec_dict, ) # Mark terminal if success if exec_result.success: self.state.terminated = True self.state.termination_reason = "apis_succeeded" # If the failure was a master-data validation (bad supplier code, bad item, etc.) the # error payload carries `valid_options_sample` — pull it out so the reply prompt can # show options, and so the UI can render a dropdown the user can click. master_data_error: dict | None = None for c in exec_result.calls: if c.response and isinstance(c.response.get("error"), dict): err = c.response["error"] if err.get("code") == "MASTER_DATA_INVALID": master_data_error = err break # SLOT-level unset so the user's bad value doesn't get re-sent on retry if master_data_error: slot = master_data_error.get("slot") if slot: self.state.unset_slot(slot) reply = generate_reply( plan_kind="fire_apis_result", plan_details={ "success": exec_result.success, "calls": [ {"screen": c.screen, "api_file": c.api_file, "status": "ok" if c.success else "error", "response": c.response, "error": c.error} for c in exec_result.calls ], "po_no": next( (c.response.get("po_no") for c in exec_result.calls if c.response and c.response.get("po_no")), None, ), "error_summary": exec_result.error_summary, "master_data_error": master_data_error, # may be None }, state_snapshot={ "current_journey": self.state.current_journey, "recent_turns": _recent_turns(self.state), }, user_message=user_message, ) # When the failure was master-data, present the valid options as a dropdown so the # tester can click instead of typing. This is honest UI: the same set of values the # mock executor would accept on retry. ui_hint: dict = {"kind": "text"} if master_data_error and master_data_error.get("valid_options_sample"): ui_hint = { "kind": "dropdown", "slot": master_data_error.get("slot") or "_", "label": f"Pick a valid {master_data_error.get('slot')}", "options": list(master_data_error["valid_options_sample"]), } return TurnResult( reply=reply.reply, ui_hint=ui_hint, execution=exec_dict, )