diff --git a/agents/critic.py b/agents/critic.py index 79d942f..54656cd 100644 --- a/agents/critic.py +++ b/agents/critic.py @@ -26,6 +26,8 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: summary, findings = build_critic_review( state.latest_plan, decision_log.candidate_evaluations, + state=state, + event=event, ) state.latest_plan.critic_summary = summary decision_log.critic_summary = summary diff --git a/agents/demand.py b/agents/demand.py index 2f38731..54dab0e 100644 --- a/agents/demand.py +++ b/agents/demand.py @@ -1,12 +1,13 @@ -from __future__ import annotations - -from collections import defaultdict - -import pandas as pd - -from agents.base import BaseAgent -from core.enums import ActionType, EventType -from core.models import Action, AgentProposal, Event, SystemState +from __future__ import annotations + +from collections import defaultdict + +import pandas as pd + +from agents.base import BaseAgent +from core.scenario_scope import payload_demand_changes, resolve_scenario_scope +from core.enums import ActionType, EventType +from core.models import Action, AgentProposal, Event, SystemState class DemandAgent(BaseAgent): @@ -14,25 +15,37 @@ class DemandAgent(BaseAgent): custom_system_prompt = """You are the Demand Agent for a supply chain management system. Your role is to analyze demand spikes, historical demand patterns, and forecast future demand. When a disruption or event occurs (like a DEMAND_SPIKE), you evaluate how it impacts the required inventory levels. -Provide a concise, professional summary (under 4 sentences) of your demand analysis, the specific SKU affected, and your recommendations. -Write the summary entirely in English. Do NOT use markdown formatting or bullet points in the summary.""" + Provide a concise, professional summary (under 4 sentences) of your demand analysis, the specific SKU affected, and your recommendations. + Write the summary entirely in English. Do NOT use markdown formatting or bullet points in the summary.""" - def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: - proposal = AgentProposal(agent=self.name) - grouped: dict[str, list[tuple[int, int]]] = defaultdict(list) - for demand in state.demands: - grouped[demand.sku].append((demand.day_index, demand.quantity)) + @staticmethod + def _event_demand_changes(event: Event | None) -> dict[str, float]: + if event is None: + return {} + if event.type not in {EventType.DEMAND_SPIKE, EventType.COMPOUND}: + return {} + return { + str(item["sku"]): float(item.get("multiplier", 1.0)) + for item in payload_demand_changes(event) + } - multiplier = 1.0 - event_sku = None - if event and event.type == EventType.DEMAND_SPIKE: - multiplier = float(event.payload.get("multiplier", 1.5)) - event_sku = event.payload.get("sku") - proposal.observations.append( - f"demand spike multiplier applied: {multiplier:.2f}" - ) + def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: + proposal = AgentProposal(agent=self.name) + grouped: dict[str, list[tuple[int, int]]] = defaultdict(list) + for demand in state.demands: + grouped[demand.sku].append((demand.day_index, demand.quantity)) + + event_changes = self._event_demand_changes(event) + scope = resolve_scenario_scope(state, event) + affected_skus = set(scope.demand_affected_skus or event_changes) + for sku, multiplier in event_changes.items(): + proposal.observations.append( + f"{sku} demand spike multiplier applied: {multiplier:.2f}" + ) for sku, points in grouped.items(): + if affected_skus and sku not in affected_skus: + continue df = pd.DataFrame(points, columns=["day_index", "quantity"]).sort_values( "day_index" ) @@ -44,8 +57,8 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: forecast = int(round(avg_d)) if pd.notna(avg_d) else 0 std_d_val = float(std_d) if pd.notna(std_d) else 0.0 - if sku == event_sku: - forecast = int(round(forecast * multiplier)) + if sku in event_changes: + forecast = int(round(forecast * event_changes[sku])) if sku in state.inventory: state.inventory[sku].forecast_qty = max(0, forecast) state.inventory[sku].std_demand = std_d_val @@ -53,26 +66,27 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: f"{sku} historical demand (30 days): Avg = {forecast}, Std Dev = {std_d_val:.2f}" ) - if ( - event - and event.type == EventType.DEMAND_SPIKE - and event_sku in state.inventory - ): - item = state.inventory[event_sku] - proposal.proposals.append( - Action( - action_id=f"act_demand_rebalance_{event_sku}", - action_type=ActionType.REBALANCE, - target_id=event_sku, - parameters={"quantity": max(item.forecast_qty - item.on_hand, 0)}, - estimated_cost_delta=40.0, - estimated_service_delta=0.08, - estimated_risk_delta=-0.10, - estimated_recovery_hours=8.0, - reason=f"pre-position stock for {event_sku} demand spike", - priority=0.75, + if event and event.type in {EventType.DEMAND_SPIKE, EventType.COMPOUND}: + for event_sku in affected_skus: + if event_sku not in state.inventory: + continue + item = state.inventory[event_sku] + proposal.proposals.append( + Action( + action_id=f"act_demand_rebalance_{event_sku}", + action_type=ActionType.REBALANCE, + target_id=event_sku, + parameters={ + "quantity": max(item.forecast_qty - item.on_hand, 0) + }, + estimated_cost_delta=40.0, + estimated_service_delta=0.08, + estimated_risk_delta=-0.10, + estimated_recovery_hours=8.0, + reason=f"pre-position stock for {event_sku} demand spike", + priority=0.75, + ) ) - ) affected_inventory = { sku: { @@ -83,17 +97,18 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: "safety_stock": item.safety_stock, } for sku, item in state.inventory.items() - if event_sku is None or sku == event_sku + if not affected_skus or sku in affected_skus } if event is not None or proposal.proposals: self.enrich_with_llm( state=state, event=event, - proposal=proposal, - state_slice={ - "event_sku": event_sku, - "multiplier": multiplier, - "affected_inventory": affected_inventory, - }, - ) - return proposal + proposal=proposal, + state_slice={ + "event_skus": sorted(affected_skus), + "demand_changes": event_changes, + "scenario_scope": scope.to_dict(), + "affected_inventory": affected_inventory, + }, + ) + return proposal diff --git a/agents/inventory.py b/agents/inventory.py index 07fbc87..b0759e3 100644 --- a/agents/inventory.py +++ b/agents/inventory.py @@ -4,51 +4,56 @@ from collections import Counter from agents.base import BaseAgent +from core.scenario_scope import resolve_scenario_scope from core.enums import ActionType from core.models import Action, AgentProposal, Event, SystemState - - -def approximate_z_score(service_level: float) -> float: - """Approximation of the inverse CDF of the normal distribution.""" - p = max(0.5001, min(0.9999, service_level)) - t = math.sqrt(-2.0 * math.log(1.0 - p)) - c0, c1, c2 = 2.515517, 0.802853, 0.010328 - d1, d2, d3 = 1.432788, 0.189269, 0.001308 - z = t - ((c2 * t + c1) * t + c0) / (((d3 * t + d2) * t + d1) * t + 1.0) - return max(0.0, z) - - + + +def approximate_z_score(service_level: float) -> float: + """Approximation of the inverse CDF of the normal distribution.""" + p = max(0.5001, min(0.9999, service_level)) + t = math.sqrt(-2.0 * math.log(1.0 - p)) + c0, c1, c2 = 2.515517, 0.802853, 0.010328 + d1, d2, d3 = 1.432788, 0.189269, 0.001308 + z = t - ((c2 * t + c1) * t + c0) / (((d3 * t + d2) * t + d1) * t + 1.0) + return max(0.0, z) + + class InventoryAgent(BaseAgent): name = "inventory" custom_system_prompt = """You are the Inventory Agent for a supply chain control tower. Explain inventory risk like an operations manager would. Be specific about stock position, expected demand, days of cover, reorder point, safety stock, and the business impact of inaction. -If replenishment is proposed, explain why that action is better than waiting. -Do not invent inventory math, customer orders, or supplier facts that are not in the provided state.""" + If replenishment is proposed, explain why that action is better than waiting. + Do not invent inventory math, customer orders, or supplier facts that are not in the provided state.""" def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: proposal = AgentProposal(agent=self.name) sku_order_counts = Counter(order.sku for order in state.orders) risk_rows: list[dict[str, object]] = [] + scope = resolve_scenario_scope(state, event) + affected_skus = set(scope.affected_skus) for sku, item in state.inventory.items(): + if affected_skus and sku not in affected_skus: + continue # --- 1. Calculate parameters from Demand and Supplier --- avg_d = item.forecast_qty std_d = getattr(item, "std_demand", 0.0) - - supplier_key = f"{item.preferred_supplier_id}_{sku}" - supplier = state.suppliers.get(supplier_key) - - lead_time = supplier.lead_time_days if supplier else 0 - reliability = supplier.reliability if supplier else 0.95 - - z_score = approximate_z_score(reliability) - - # --- 2. Calculate LTD, SS, ROP --- - ltd = avg_d * lead_time - ss = z_score * std_d * math.sqrt(lead_time) if lead_time > 0 else 0.0 - rop = ltd + ss - - # --- 3. Save back to InventoryItem --- + + supplier_key = f"{item.preferred_supplier_id}_{sku}" + supplier = state.suppliers.get(supplier_key) + + lead_time = supplier.lead_time_days if supplier else 0 + reliability = supplier.reliability if supplier else 0.95 + + z_score = approximate_z_score(reliability) + + # --- 2. Calculate LTD, SS, ROP --- + ltd = avg_d * lead_time + ss = z_score * std_d * math.sqrt(lead_time) if lead_time > 0 else 0.0 + rop = ltd + ss + + # --- 3. Save back to InventoryItem --- item.reorder_point = int(round(rop)) item.safety_stock = int(round(ss)) @@ -56,9 +61,7 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: daily_demand = max(float(avg_d), 0.0) available_stock = item.on_hand + item.incoming_qty days_of_cover = ( - available_stock / daily_demand - if daily_demand > 0 - else float("inf") + available_stock / daily_demand if daily_demand > 0 else float("inf") ) order_count = sku_order_counts.get(sku, 0) root_cause = [] @@ -70,7 +73,9 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: root_cause.append("below-target supplier reliability") if daily_demand > max(item.on_hand, 1): root_cause.append("demand running ahead of current on-hand stock") - cause_text = ", ".join(root_cause) if root_cause else "normal replenishment cycle" + cause_text = ( + ", ".join(root_cause) if root_cause else "normal replenishment cycle" + ) if projected < item.safety_stock: inventory_status = "stockout risk" @@ -92,7 +97,9 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: "projected_stock": projected, "reorder_point": item.reorder_point, "safety_stock": item.safety_stock, - "days_of_cover": None if days_of_cover == float("inf") else round(days_of_cover, 1), + "days_of_cover": None + if days_of_cover == float("inf") + else round(days_of_cover, 1), "lead_time_days": lead_time, "supplier_reliability": reliability, "pending_orders": order_count, @@ -104,27 +111,33 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: quantity = max(item.reorder_point + item.safety_stock - projected, 0) proposal.proposals.append( Action( - action_id=f"act_reorder_{sku}", - action_type=ActionType.REORDER, - target_id=sku, - parameters={"quantity": int(quantity)}, - estimated_cost_delta=quantity * item.unit_cost * 1.05, - estimated_service_delta=0.12, - estimated_risk_delta=-0.15, - estimated_recovery_hours=24.0, - reason=f"projected stock ({projected}) for {sku} falls below ROP ({item.reorder_point})", + action_id=f"act_reorder_{sku}", + action_type=ActionType.REORDER, + target_id=sku, + parameters={"quantity": int(quantity)}, + estimated_cost_delta=quantity * item.unit_cost * 1.05, + estimated_service_delta=0.12, + estimated_risk_delta=-0.15, + estimated_recovery_hours=24.0, + reason=f"projected stock ({projected}) for {sku} falls below ROP ({item.reorder_point})", priority=0.8 if projected < item.safety_stock else 0.6, ) ) ranked_rows = sorted( risk_rows, key=lambda row: ( - 0 if row["inventory_status"] == "stockout risk" else 1 if row["inventory_status"] == "reorder risk" else 2, + 0 + if row["inventory_status"] == "stockout risk" + else 1 + if row["inventory_status"] == "reorder risk" + else 2, row["days_of_cover"] if row["days_of_cover"] is not None else 9999, ), ) critical_rows = [ - row for row in ranked_rows if row["inventory_status"] in {"stockout risk", "reorder risk"} + row + for row in ranked_rows + if row["inventory_status"] in {"stockout risk", "reorder risk"} ] if critical_rows: top = critical_rows[0] @@ -140,7 +153,7 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: else "limited demand visibility" ) proposal.domain_summary = ( - f"Inventory pressure is concentrated on {top['name']} ({top['sku']}). " + f"Inventory pressure is concentrated on {top['name']} ({top['sku']}) within the direct disruption scope. " f"Projected stock falls to {int(top['projected_stock'])} units versus a reorder point of {int(top['reorder_point'])} " f"and safety stock of {int(top['safety_stock'])}. At the current demand pace, that leaves {coverage_text}. " f"The main driver is {top['root_cause']}; if no action is taken, stock availability could tighten within the next replenishment cycle{order_text}." @@ -165,10 +178,10 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: f"{critical_rows[0]['days_of_cover'] or 0:.1f} days of cover." ) elif ranked_rows: - stable_count = sum(1 for row in ranked_rows if row["inventory_status"] == "stable") - proposal.domain_summary = ( - f"Inventory is broadly stable. {stable_count} SKUs remain above reorder point after applying current demand, lead time, and safety stock assumptions." + stable_count = sum( + 1 for row in ranked_rows if row["inventory_status"] == "stable" ) + proposal.domain_summary = f"Inventory is broadly stable. {stable_count} SKUs remain above reorder point after applying current demand, lead time, and safety stock assumptions." proposal.tradeoffs.append( "Holding current inventory avoids unnecessary replenishment cost, but coverage should still be refreshed as demand and supplier conditions change." ) @@ -180,6 +193,7 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: event=event, proposal=proposal, state_slice={ + "scenario_scope": scope.to_dict(), "inventory_focus": ranked_rows[:5], "critical_sku_count": len(critical_rows), "proposed_reorders": [ diff --git a/agents/logistics.py b/agents/logistics.py index 4d199b7..c55dc50 100644 --- a/agents/logistics.py +++ b/agents/logistics.py @@ -1,8 +1,9 @@ -from __future__ import annotations - -from agents.base import BaseAgent -from core.enums import ActionType, EventType -from core.models import Action, AgentProposal, Event, SystemState +from __future__ import annotations + +from agents.base import BaseAgent +from core.scenario_scope import payload_route_ids, resolve_scenario_scope +from core.enums import ActionType +from core.models import Action, AgentProposal, Event, SystemState class LogisticsAgent(BaseAgent): @@ -12,69 +13,72 @@ class LogisticsAgent(BaseAgent): Provide a concise, professional summary (under 4 sentences) of the routing issues, your analysis of alternative routes, and your specific rerouting recommendations. Write the summary entirely in English. Do NOT use markdown formatting or bullet points in the summary.""" - def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: - proposal = AgentProposal(agent=self.name) - blocked_route = ( - event.payload.get("route_id") - if event and event.type in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND} - else None - ) - available_routes = [ - route for route in state.routes.values() if route.status != "blocked" - ] - - for sku, item in state.inventory.items(): - current_route = state.routes.get(item.preferred_route_id) - if not current_route: - continue - - valid_routes = [ - r - for r in available_routes - if r.destination == current_route.destination + def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: + proposal = AgentProposal(agent=self.name) + scope = resolve_scenario_scope(state, event) + blocked_routes = set(payload_route_ids(event)) + affected_skus = set(scope.route_affected_skus) + available_routes = [ + route + for route in state.routes.values() + if route.status != "blocked" and route.route_id not in blocked_routes + ] + + for sku, item in state.inventory.items(): + if affected_skus and sku not in affected_skus: + continue + current_route = state.routes.get(item.preferred_route_id) + if not current_route: + continue + if blocked_routes and current_route.route_id not in blocked_routes: + continue + + valid_routes = [ + r + for r in available_routes + if r.destination == current_route.destination and r.origin == current_route.origin ] - ranked = sorted( - valid_routes, - key=lambda route: (route.risk_score, route.transit_days, route.cost), - ) - - best = ranked[0] if ranked else current_route - if best.route_id == blocked_route and len(ranked) > 1: - best = ranked[1] - if best.route_id != current_route.route_id and ( - current_route.route_id == blocked_route - or best.risk_score < current_route.risk_score - ): - proposal.proposals.append( - Action( - action_id=f"act_reroute_{sku}_{best.route_id}", - action_type=ActionType.REROUTE, - target_id=sku, + ranked = sorted( + valid_routes, + key=lambda route: (route.risk_score, route.transit_days, route.cost), + ) + + best = ranked[0] if ranked else None + if best is not None and best.route_id != current_route.route_id: + proposal.proposals.append( + Action( + action_id=f"act_reroute_{sku}_{best.route_id}", + action_type=ActionType.REROUTE, + target_id=sku, parameters={"route_id": best.route_id}, - estimated_cost_delta=max(best.cost - current_route.cost, 0.0), - estimated_service_delta=0.04, - estimated_risk_delta=-0.10, - estimated_recovery_hours=float(best.transit_days * 8), - reason=f"reroute {sku} from {current_route.route_id} to {best.route_id}", - priority=0.85 - if current_route.route_id == blocked_route - else 0.45, - ) - ) - proposal.observations.append( - f"{sku} reroute candidate: {best.route_id}" - ) - - ranked_all = sorted( - available_routes, - key=lambda route: (route.risk_score, route.transit_days, route.cost), - ) - if event is not None or proposal.proposals: - route_snapshot = [ - { - "route_id": route.route_id, - "origin": route.origin, + estimated_cost_delta=max(best.cost - current_route.cost, 0.0), + estimated_service_delta=0.04, + estimated_risk_delta=-0.10, + estimated_recovery_hours=float(best.transit_days * 8), + reason=f"reroute {sku} from {current_route.route_id} to {best.route_id}", + priority=0.85, + ) + ) + proposal.observations.append( + f"{sku} reroute candidate: {best.route_id}" + ) + + ranked_all = sorted( + available_routes, + key=lambda route: (route.risk_score, route.transit_days, route.cost), + ) + if blocked_routes: + proposal.domain_summary = ( + f"Routing disruption is limited to {len(blocked_routes)} blocked lane(s), " + f"directly exposing {len(affected_skus)} SKU(s) across {len(scope.warehouse_ids)} warehouse(s). " + f"Only those directly exposed SKUs were evaluated for reroute recovery." + ) + if event is not None or proposal.proposals: + route_snapshot = [ + { + "route_id": route.route_id, + "origin": route.origin, "destination": route.destination, "transit_days": route.transit_days, "cost": route.cost, @@ -84,12 +88,13 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: for route in ranked_all ] self.enrich_with_llm( - state=state, - event=event, - proposal=proposal, - state_slice={ - "blocked_route": blocked_route, - "route_options": route_snapshot, - }, - ) - return proposal + state=state, + event=event, + proposal=proposal, + state_slice={ + "blocked_routes": sorted(blocked_routes), + "scenario_scope": scope.to_dict(), + "route_options": route_snapshot, + }, + ) + return proposal diff --git a/agents/planner.py b/agents/planner.py index 4bb67ad..15a0857 100644 --- a/agents/planner.py +++ b/agents/planner.py @@ -3,17 +3,17 @@ from uuid import uuid4 from agents.base import BaseAgent -from core.enums import ActionType, ApprovalStatus, PlanStatus -from core.models import ( - Action, - AgentProposal, - CandidatePlanDraft, - CandidatePlanEvaluation, - DecisionLog, - Event, - Plan, - SystemState, - PlanMetadata +from core.enums import ActionType, ApprovalStatus, PlanStatus +from core.models import ( + Action, + AgentProposal, + CandidatePlanDraft, + CandidatePlanEvaluation, + DecisionLog, + Event, + Plan, + SystemState, + PlanMetadata, ) from core.logger import get_logger from llm.service import enrich_plan_and_decision, generate_candidate_plan_drafts @@ -22,12 +22,17 @@ build_winning_factors, explain_rejected_actions, ) -from policies.constraints import evaluate_plan_constraints, mode_rationale, evaluate_hard_constraints, evaluate_soft_constraints +from policies.constraints import ( + evaluate_plan_constraints, + mode_rationale, + evaluate_hard_constraints, + evaluate_soft_constraints, +) from policies.guardrails import approval_required from policies.scoring import compute_score from policies.strategic_prompt import ( - retrieve_relevant_cases, - compute_memory_influence, + retrieve_relevant_cases, + compute_memory_influence, derive_strategy_rationale, build_strategic_prompt, ) @@ -35,67 +40,69 @@ logger = get_logger(__name__) - + STRATEGY_ORDER = ("cost_first", "balanced", "resilience_first") - - -def _dedupe_actions(actions: list[Action], limit: int) -> list[Action]: - selected: list[Action] = [] - seen: set[tuple[str, str]] = set() - for action in actions: - key = (action.action_type.value, action.target_id) - if key in seen: - continue - selected.append(action) - seen.add(key) - if len(selected) >= limit: - break - return selected - - -def _fallback_sort_key(strategy_label: str, action: Action) -> tuple[float, float, float, float]: - if strategy_label == "cost_first": - return ( - action.estimated_cost_delta, - action.estimated_recovery_hours, - -action.estimated_service_delta, - action.estimated_risk_delta, - ) - if strategy_label == "resilience_first": - return ( - action.estimated_risk_delta, - action.estimated_recovery_hours, - -action.estimated_service_delta, - action.estimated_cost_delta, - ) - return ( - -action.priority, - action.estimated_risk_delta, - -action.estimated_service_delta, - action.estimated_cost_delta, - ) - - -def _strategy_reason(strategy_label: str) -> str: - if strategy_label == "cost_first": - return "fallback cost-first strategy favors lower incremental cost and operational simplicity" - if strategy_label == "resilience_first": - return "fallback resilience-first strategy favors risk reduction and faster recovery" - return "fallback balanced strategy blends specialist priority, risk reduction, and service protection" - - + + +def _dedupe_actions(actions: list[Action], limit: int) -> list[Action]: + selected: list[Action] = [] + seen: set[tuple[str, str]] = set() + for action in actions: + key = (action.action_type.value, action.target_id) + if key in seen: + continue + selected.append(action) + seen.add(key) + if len(selected) >= limit: + break + return selected + + +def _fallback_sort_key( + strategy_label: str, action: Action +) -> tuple[float, float, float, float]: + if strategy_label == "cost_first": + return ( + action.estimated_cost_delta, + action.estimated_recovery_hours, + -action.estimated_service_delta, + action.estimated_risk_delta, + ) + if strategy_label == "resilience_first": + return ( + action.estimated_risk_delta, + action.estimated_recovery_hours, + -action.estimated_service_delta, + action.estimated_cost_delta, + ) + return ( + -action.priority, + action.estimated_risk_delta, + -action.estimated_service_delta, + action.estimated_cost_delta, + ) + + +def _strategy_reason(strategy_label: str) -> str: + if strategy_label == "cost_first": + return "fallback cost-first strategy favors lower incremental cost and operational simplicity" + if strategy_label == "resilience_first": + return "fallback resilience-first strategy favors risk reduction and faster recovery" + return "fallback balanced strategy blends specialist priority, risk reduction, and service protection" + + def _ensure_actions(candidate_actions: list[Action]) -> list[Action]: ensured = list(candidate_actions) if any(action.action_type == ActionType.NO_OP for action in ensured): return ensured - ensured.append( - Action( - action_id="act_no_op", - action_type=ActionType.NO_OP, - target_id="system", - reason="no action required", - priority=0.1, - ) + ensured.append( + Action( + action_id="act_no_op", + action_type=ActionType.NO_OP, + target_id="system", + reason="no action required", + priority=0.1, + ) ) return ensured @@ -136,57 +143,71 @@ def _should_suppress_no_op(state: SystemState) -> bool: or pressure["critical_skus"] >= 8 or pressure["below_safety_stock"] >= 2 ) - - -def _fallback_drafts(candidate_actions: list[Action], action_limit: int) -> list[CandidatePlanDraft]: - drafts: list[CandidatePlanDraft] = [] - for strategy_label in STRATEGY_ORDER: - sorted_actions = sorted(candidate_actions, key=lambda action: _fallback_sort_key(strategy_label, action)) - selected = _dedupe_actions(sorted_actions, action_limit) - drafts.append( - CandidatePlanDraft( - strategy_label=strategy_label, - action_ids=[action.action_id for action in selected], - rationale=_strategy_reason(strategy_label), - llm_used=False, - ) - ) - return drafts - - + + +def _fallback_drafts( + candidate_actions: list[Action], action_limit: int +) -> list[CandidatePlanDraft]: + drafts: list[CandidatePlanDraft] = [] + for strategy_label in STRATEGY_ORDER: + sorted_actions = sorted( + candidate_actions, + key=lambda action: _fallback_sort_key(strategy_label, action), + ) + selected = _dedupe_actions(sorted_actions, action_limit) + drafts.append( + CandidatePlanDraft( + strategy_label=strategy_label, + action_ids=[action.action_id for action in selected], + rationale=_strategy_reason(strategy_label), + llm_used=False, + ) + ) + return drafts + + def _normalize_drafts( drafts: list[CandidatePlanDraft], candidate_actions: list[Action], action_limit: int, ) -> tuple[list[CandidatePlanDraft], int]: - allowed_ids = {action.action_id for action in candidate_actions} - by_id = {action.action_id: action for action in candidate_actions} - normalized: list[CandidatePlanDraft] = [] - repaired_count = 0 - fallback_map = {draft.strategy_label: draft for draft in _fallback_drafts(candidate_actions, action_limit)} - draft_map = {draft.strategy_label: draft for draft in drafts if draft.strategy_label in STRATEGY_ORDER} - - for strategy_label in STRATEGY_ORDER: - draft = draft_map.get(strategy_label) - if draft is None: - normalized.append(fallback_map[strategy_label]) - repaired_count += 1 - continue - action_ids = [action_id for action_id in draft.action_ids if action_id in allowed_ids] - selected_actions = _dedupe_actions( - [by_id[action_id] for action_id in action_ids if action_id in by_id], - action_limit, - ) - if not selected_actions: - normalized.append(fallback_map[strategy_label]) - repaired_count += 1 - continue - normalized.append( - CandidatePlanDraft( - strategy_label=strategy_label, - action_ids=[action.action_id for action in selected_actions], - rationale=draft.rationale or _strategy_reason(strategy_label), - llm_used=draft.llm_used, + allowed_ids = {action.action_id for action in candidate_actions} + by_id = {action.action_id: action for action in candidate_actions} + normalized: list[CandidatePlanDraft] = [] + repaired_count = 0 + fallback_map = { + draft.strategy_label: draft + for draft in _fallback_drafts(candidate_actions, action_limit) + } + draft_map = { + draft.strategy_label: draft + for draft in drafts + if draft.strategy_label in STRATEGY_ORDER + } + + for strategy_label in STRATEGY_ORDER: + draft = draft_map.get(strategy_label) + if draft is None: + normalized.append(fallback_map[strategy_label]) + repaired_count += 1 + continue + action_ids = [ + action_id for action_id in draft.action_ids if action_id in allowed_ids + ] + selected_actions = _dedupe_actions( + [by_id[action_id] for action_id in action_ids if action_id in by_id], + action_limit, + ) + if not selected_actions: + normalized.append(fallback_map[strategy_label]) + repaired_count += 1 + continue + normalized.append( + CandidatePlanDraft( + strategy_label=strategy_label, + action_ids=[action.action_id for action in selected_actions], + rationale=draft.rationale or _strategy_reason(strategy_label), + llm_used=draft.llm_used, ) ) return normalized, repaired_count @@ -303,16 +324,17 @@ def _decision_priority_score( ) -> float: coverage = _coverage_metrics(evaluation, gap_by_sku=gap_by_sku, by_id=by_id) pressure = _inventory_pressure_summary(state) - stressed_normal = ( - mode == "normal" - and ( - state.kpis.stockout_risk >= 0.05 - or pressure["critical_skus"] >= 8 - or pressure["below_safety_stock"] >= 2 - ) + stressed_normal = mode == "normal" and ( + state.kpis.stockout_risk >= 0.05 + or pressure["critical_skus"] >= 8 + or pressure["below_safety_stock"] >= 2 + ) + coverage_weight = ( + 0.016 if mode == "crisis" else (0.075 if stressed_normal else 0.012) + ) + unresolved_penalty = ( + 0.0012 if mode == "crisis" else (0.0025 if stressed_normal else 0.0008) ) - coverage_weight = 0.016 if mode == "crisis" else (0.075 if stressed_normal else 0.012) - unresolved_penalty = 0.0012 if mode == "crisis" else (0.0025 if stressed_normal else 0.0008) return round( evaluation.score + (coverage["coverage_fraction"] * coverage_weight) @@ -322,31 +344,31 @@ def _decision_priority_score( def _evaluate_candidate( - *, - state: SystemState, - event: Event | None, - before_kpis, - strategy_label: str, - actions: list[Action], - rationale: str, - llm_used: bool, -) -> CandidatePlanEvaluation: - selected_mode_rationale = mode_rationale(state, event) - violations = evaluate_plan_constraints( - state=state, - event=event, - actions=actions, - ) - feasible = not violations + *, + state: SystemState, + event: Event | None, + before_kpis, + strategy_label: str, + actions: list[Action], + rationale: str, + llm_used: bool, +) -> CandidatePlanEvaluation: + selected_mode_rationale = mode_rationale(state, event) + violations = evaluate_plan_constraints( + state=state, + event=event, + actions=actions, + ) + feasible = not violations if not feasible: return CandidatePlanEvaluation( strategy_label=strategy_label, action_ids=[action.action_id for action in actions], score=0.0, - score_breakdown={ - "service_level": 0.0, - "total_cost": 0.0, - "disruption_risk": 0.0, + score_breakdown={ + "service_level": 0.0, + "total_cost": 0.0, + "disruption_risk": 0.0, "recovery_speed": 0.0, }, projected_kpis=before_kpis.model_copy(deep=True), @@ -376,14 +398,14 @@ def _evaluate_candidate( mode=state.mode, baseline_cost=before_kpis.total_cost, ) - transient_plan = Plan( - plan_id=f"plan_eval_{uuid4().hex[:8]}", - mode=state.mode, - trigger_event_ids=[event.event_id] if event else [], - actions=actions, - score=score, - score_breakdown=breakdown, - strategy_label=strategy_label, + transient_plan = Plan( + plan_id=f"plan_eval_{uuid4().hex[:8]}", + mode=state.mode, + trigger_event_ids=[event.event_id] if event else [], + actions=actions, + score=score, + score_breakdown=breakdown, + strategy_label=strategy_label, generated_by="llm" if llm_used else "deterministic_fallback", planner_reasoning=rationale, status=PlanStatus.PROPOSED, @@ -409,8 +431,10 @@ def _evaluate_candidate( violations=[], mode_rationale=selected_mode_rationale, approval_required=needs_approval, - approval_reason=reason if needs_approval else "no approval required: thresholds not triggered", - rationale=rationale, + approval_reason=reason + if needs_approval + else "no approval required: thresholds not triggered", + rationale=rationale, llm_used=llm_used, ) @@ -529,7 +553,9 @@ def _expand_selected_for_coverage( ) max_unresolved = max(int(len(gap_by_sku) * (1.0 - target_coverage)), 0) - selected_action_ids = [action_id for action_id in selected.action_ids if action_id in by_id] + selected_action_ids = [ + action_id for action_id in selected.action_ids if action_id in by_id + ] selected_actions = [by_id[action_id] for action_id in selected_action_ids] selected_targeted = { action.target_id @@ -539,7 +565,9 @@ def _expand_selected_for_coverage( total_gap = sum(gap_by_sku.values()) or 1.0 selected_covered_gap = sum(gap_by_sku.get(sku, 0.0) for sku in selected_targeted) selected_coverage = selected_covered_gap / total_gap - selected_unresolved = max(len(gap_by_sku) - len(selected_targeted & set(gap_by_sku)), 0) + selected_unresolved = max( + len(gap_by_sku) - len(selected_targeted & set(gap_by_sku)), 0 + ) if selected_coverage >= target_coverage and selected_unresolved <= max_unresolved: return selected @@ -596,18 +624,20 @@ def _expand_selected_for_coverage( gap_by_sku=gap_by_sku, by_id=by_id, ) - expanded_evaluation.coverage_fraction = round(expanded_coverage["coverage_fraction"], 4) + expanded_evaluation.coverage_fraction = round( + expanded_coverage["coverage_fraction"], 4 + ) expanded_evaluation.critical_covered = int(expanded_coverage["critical_covered"]) - expanded_evaluation.unresolved_critical = int(expanded_coverage["unresolved_critical"]) + expanded_evaluation.unresolved_critical = int( + expanded_coverage["unresolved_critical"] + ) if ( expanded_evaluation.unresolved_critical < selected.unresolved_critical or expanded_evaluation.coverage_fraction > selected.coverage_fraction ): expanded_evaluation.approval_required = True - expanded_evaluation.approval_reason = ( - "coverage guardrail expanded the selected strategy to reduce unresolved critical SKUs" - ) + expanded_evaluation.approval_reason = "coverage guardrail expanded the selected strategy to reduce unresolved critical SKUs" return expanded_evaluation return selected @@ -621,7 +651,9 @@ def _optimize_draft_scope( before_kpis, gap_by_sku: dict[str, float], ) -> CandidatePlanEvaluation: - ordered_actions = [by_id[action_id] for action_id in draft.action_ids if action_id in by_id] + ordered_actions = [ + by_id[action_id] for action_id in draft.action_ids if action_id in by_id + ] min_actions, max_actions = _adaptive_scope_bounds( state=state, event=event, @@ -670,17 +702,14 @@ def _optimize_draft_scope( by_id=by_id, ) candidates.append((evaluation, adjusted_score, coverage["coverage_fraction"])) - if ( - adjusted_score > best_adjusted - or ( - adjusted_score == best_adjusted - and ( - priority_score > best_priority - or ( - priority_score == best_priority - and best_evaluation is not None - and len(evaluation.action_ids) < len(best_evaluation.action_ids) - ) + if adjusted_score > best_adjusted or ( + adjusted_score == best_adjusted + and ( + priority_score > best_priority + or ( + priority_score == best_priority + and best_evaluation is not None + and len(evaluation.action_ids) < len(best_evaluation.action_ids) ) ) ): @@ -739,9 +768,9 @@ def _select_best_evaluation( -item.projected_kpis.total_cost, 1 if item.strategy_label == "balanced" else 0, ), - ) - - + ) + + def _selection_reason( selected: CandidatePlanEvaluation, evaluations: list[CandidatePlanEvaluation], @@ -773,7 +802,7 @@ def _selection_reason( item.projected_kpis.service_level, -item.projected_kpis.total_cost, ), - reverse=True, + reverse=True, ) runner_up = ordered[1] if len(ordered) > 1 else None base = ( @@ -803,30 +832,32 @@ def _selection_reason( f" It also covered more at-risk SKUs ({int(selected_coverage['critical_covered'])} vs " f"{int(runner_up_coverage['critical_covered'])})." ) - elif selected_coverage["coverage_fraction"] > runner_up_coverage["coverage_fraction"]: + elif ( + selected_coverage["coverage_fraction"] > runner_up_coverage["coverage_fraction"] + ): coverage_note = " It addressed a larger share of current inventory exposure." return ( f"{base}. It ranked ahead of {runner_up.strategy_label} ({runner_up_priority_score:.4f}) after comparing " f"service protection, cost, disruption risk, and recovery speed.{coverage_note}" ) - - -def _safe_hold_evaluation( - *, - state: SystemState, - event: Event | None, - before_kpis, - safe_action: Action, - infeasible_count: int, -) -> CandidatePlanEvaluation: - score, breakdown = compute_score( - service_level=before_kpis.service_level, - total_cost=before_kpis.total_cost, - disruption_risk=before_kpis.disruption_risk, - recovery_speed=before_kpis.recovery_speed, - mode=state.mode, - baseline_cost=before_kpis.total_cost, - ) + + +def _safe_hold_evaluation( + *, + state: SystemState, + event: Event | None, + before_kpis, + safe_action: Action, + infeasible_count: int, +) -> CandidatePlanEvaluation: + score, breakdown = compute_score( + service_level=before_kpis.service_level, + total_cost=before_kpis.total_cost, + disruption_risk=before_kpis.disruption_risk, + recovery_speed=before_kpis.recovery_speed, + mode=state.mode, + baseline_cost=before_kpis.total_cost, + ) return CandidatePlanEvaluation( strategy_label="safe_hold", action_ids=[safe_action.action_id], @@ -843,20 +874,20 @@ def _safe_hold_evaluation( mode_rationale=mode_rationale(state, event), approval_required=False, approval_reason="no approval required: retaining current state because all candidate plans were infeasible", - rationale=( - f"no candidate plan satisfied hard constraints; retaining the current network state after excluding " - f"{infeasible_count} infeasible candidate plan(s)" - ), - llm_used=False, - ) - - -class PlannerAgent(BaseAgent): - name = "planner" - + rationale=( + f"no candidate plan satisfied hard constraints; retaining the current network state after excluding " + f"{infeasible_count} infeasible candidate plan(s)" + ), + llm_used=False, + ) + + +class PlannerAgent(BaseAgent): + name = "planner" + def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: proposal = AgentProposal(agent=self.name) - + candidate_actions = _ensure_actions(state.candidate_actions) if event is not None: candidate_actions = _drop_no_op_when_actions_exist(candidate_actions) @@ -867,31 +898,46 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: feasible_candidates = [] infeasible_reasons = {} for act in candidate_actions: - dummy_plan = Plan(plan_id="tmp", mode=state.mode, score=0, score_breakdown={}, actions=[act]) - is_feas, vios = evaluate_hard_constraints(dummy_plan, state) - if is_feas: - feasible_candidates.append(act) - else: - infeasible_reasons[act.action_id] = vios - - if not feasible_candidates and candidate_actions: - feasible_candidates = [ + dummy_plan = Plan( + plan_id="tmp", + mode=state.mode, + score=0, + score_breakdown={}, + actions=[act], + ) + is_feas, vios = evaluate_hard_constraints(dummy_plan, state) + if is_feas: + feasible_candidates.append(act) + else: + infeasible_reasons[act.action_id] = vios + + if not feasible_candidates and candidate_actions: + feasible_candidates = [ Action( - action_id="act_no_op_fallback", action_type=ActionType.NO_OP, target_id="system", - reason="all candidate actions violated hard constraints", priority=0.0 + action_id="act_no_op_fallback", + action_type=ActionType.NO_OP, + target_id="system", + reason="all candidate actions violated hard constraints", + priority=0.0, ) ] action_limit = _action_limit_for_mode(state, len(feasible_candidates)) - - effective_event = event or (state.active_events[-1] if state.active_events else None) - historical_cases = retrieve_relevant_cases(effective_event, state.memory, top_k=3) - memory_influence = compute_memory_influence(historical_cases) - strategic_prompt = build_strategic_prompt( - mode=state.mode.value, event=effective_event, - historical_cases=historical_cases, candidate_actions=feasible_candidates - ) - - before_kpis = state.kpis.model_copy(deep=True) + + effective_event = event or ( + state.active_events[-1] if state.active_events else None + ) + historical_cases = retrieve_relevant_cases( + effective_event, state.memory, top_k=3 + ) + memory_influence = compute_memory_influence(historical_cases) + strategic_prompt = build_strategic_prompt( + mode=state.mode.value, + event=effective_event, + historical_cases=historical_cases, + candidate_actions=feasible_candidates, + ) + + before_kpis = state.kpis.model_copy(deep=True) llm_drafts, planner_error = generate_candidate_plan_drafts( state=state, event=event, candidate_actions=feasible_candidates ) @@ -903,7 +949,7 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: suppress_no_op, action_limit, ) - + if planner_error == "llm_disabled": drafts = _fallback_drafts(feasible_candidates, action_limit) repaired_count = 0 @@ -913,7 +959,9 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: action_limit, ) else: - drafts, repaired_count = _normalize_drafts(llm_drafts, feasible_candidates, action_limit) + drafts, repaired_count = _normalize_drafts( + llm_drafts, feasible_candidates, action_limit + ) drafts, duplicate_repair_count = _enforce_distinct_drafts( drafts, feasible_candidates, @@ -922,7 +970,9 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: repaired_count += duplicate_repair_count fallback_used = repaired_count > 0 if fallback_used and not planner_error: - planner_error = "planner returned duplicate, incomplete, or invalid candidate plans" + planner_error = ( + "planner returned duplicate, incomplete, or invalid candidate plans" + ) if fallback_used: logger.warning( "planner fallback activated repaired_count=%s total_strategies=%s reason=%s", @@ -932,8 +982,8 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: ) else: logger.info("planner used llm candidate drafts without repair") - - by_id = {action.action_id: action for action in feasible_candidates} + + by_id = {action.action_id: action for action in feasible_candidates} gap_by_sku = _inventory_gap_map(state) evaluations: list[CandidatePlanEvaluation] = [] for draft in drafts: @@ -959,21 +1009,38 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: evaluation.unresolved_critical = int(coverage["unresolved_critical"]) if not any(item.feasible for item in evaluations): - safe_action = next((action for action in feasible_candidates if action.action_type == ActionType.NO_OP), Action(action_id="act_no_op_fallback", action_type=ActionType.NO_OP, target_id="system", reason="all candidate actions violated hard constraints", priority=0.0)) + safe_action = next( + ( + action + for action in feasible_candidates + if action.action_type == ActionType.NO_OP + ), + Action( + action_id="act_no_op_fallback", + action_type=ActionType.NO_OP, + target_id="system", + reason="all candidate actions violated hard constraints", + priority=0.0, + ), + ) evaluations.append( _safe_hold_evaluation( - state=state, event=event, before_kpis=before_kpis, - safe_action=safe_action, infeasible_count=len(evaluations), - ) - ) - - + state=state, + event=event, + before_kpis=before_kpis, + safe_action=safe_action, + infeasible_count=len(evaluations), + ) + ) + selected_evaluation = _select_best_evaluation( evaluations, state=state, by_id=by_id, ) - guardrail_event = event or (state.active_events[-1] if state.active_events else None) + guardrail_event = event or ( + state.active_events[-1] if state.active_events else None + ) selected_evaluation = _expand_selected_for_coverage( selected=selected_evaluation, state=state, @@ -982,7 +1049,11 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: by_id=by_id, gap_by_sku=gap_by_sku, ) - selected_actions = [by_id[action_id] for action_id in selected_evaluation.action_ids if action_id in by_id] + selected_actions = [ + by_id[action_id] + for action_id in selected_evaluation.action_ids + if action_id in by_id + ] requires_manual_review, manual_review_reason = _coverage_approval_guardrail( selected=selected_evaluation, state=state, @@ -998,7 +1069,7 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: selected_evaluation.approval_reason = ( f"{prior_reason}; {manual_review_reason}" ) - + ordered_evaluations = sorted( [item for item in evaluations if item.feasible] or evaluations, key=lambda item: ( @@ -1010,7 +1081,11 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: reverse=True, ) runner_up = next( - (item for item in ordered_evaluations if item.strategy_label != selected_evaluation.strategy_label), + ( + item + for item in ordered_evaluations + if item.strategy_label != selected_evaluation.strategy_label + ), None, ) summary = build_plan_summary( @@ -1025,13 +1100,15 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: projection_summary=selected_evaluation.projection_summary, simulation_horizon_days=selected_evaluation.simulation_horizon_days, worst_case_kpis=selected_evaluation.worst_case_kpis, + state=state, + event=event, ) - - final_plan = Plan( - plan_id=f"plan_{uuid4().hex[:8]}", - mode=state.mode, - trigger_event_ids=[event.event_id] if event else [], - actions=selected_actions, + + final_plan = Plan( + plan_id=f"plan_{uuid4().hex[:8]}", + mode=state.mode, + trigger_event_ids=[event.event_id] if event else [], + actions=selected_actions, score=selected_evaluation.score, score_breakdown=selected_evaluation.score_breakdown, feasible=selected_evaluation.feasible, @@ -1044,48 +1121,57 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: else ( "llm_planner" if repaired_count == 0 - else ("llm_planner_repaired" if repaired_count < len(STRATEGY_ORDER) else "hybrid_fallback") + else ( + "llm_planner_repaired" + if repaired_count < len(STRATEGY_ORDER) + else "hybrid_fallback" + ) ) ), approval_required=selected_evaluation.approval_required, - approval_reason=selected_evaluation.approval_reason if selected_evaluation.approval_required else "", + approval_reason=selected_evaluation.approval_reason + if selected_evaluation.approval_required + else "", planner_reasoning=summary, status=PlanStatus.PROPOSED, ) - - is_hard_feas, hard_violations = evaluate_hard_constraints(final_plan, state) - soft_violations = evaluate_soft_constraints(final_plan, state) - - final_plan.feasible = is_hard_feas - final_plan.violations = hard_violations + soft_violations - - - strategy_rationale = derive_strategy_rationale(effective_event, historical_cases, selected_actions) - final_plan.metadata = PlanMetadata( - referenced_cases=historical_cases, - memory_influence_score=memory_influence, - strategy_rationale=strategy_rationale, - strategic_prompt=strategic_prompt - ) - + + is_hard_feas, hard_violations = evaluate_hard_constraints(final_plan, state) + soft_violations = evaluate_soft_constraints(final_plan, state) + + final_plan.feasible = is_hard_feas + final_plan.violations = hard_violations + soft_violations + + strategy_rationale = derive_strategy_rationale( + effective_event, historical_cases, selected_actions + ) + final_plan.metadata = PlanMetadata( + referenced_cases=historical_cases, + memory_influence_score=memory_influence, + strategy_rationale=strategy_rationale, + strategic_prompt=strategic_prompt, + ) + winning_factors = build_winning_factors( selected_actions, before_kpis, selected_evaluation.projected_kpis, selected_evaluation.score_breakdown, ) - rejection_reasons = explain_rejected_actions(candidate_actions, selected_actions, action_limit) - - decision_log = DecisionLog( - decision_id=f"dec_{uuid4().hex[:8]}", - plan_id=final_plan.plan_id, - event_ids=final_plan.trigger_event_ids, - before_kpis=before_kpis, - after_kpis=selected_evaluation.projected_kpis, - selected_actions=[action.action_id for action in selected_actions], - rejected_actions=rejection_reasons, - score_breakdown=selected_evaluation.score_breakdown, - mode_rationale=selected_evaluation.mode_rationale, + rejection_reasons = explain_rejected_actions( + candidate_actions, selected_actions, action_limit + ) + + decision_log = DecisionLog( + decision_id=f"dec_{uuid4().hex[:8]}", + plan_id=final_plan.plan_id, + event_ids=final_plan.trigger_event_ids, + before_kpis=before_kpis, + after_kpis=selected_evaluation.projected_kpis, + selected_actions=[action.action_id for action in selected_actions], + rejected_actions=rejection_reasons, + score_breakdown=selected_evaluation.score_breakdown, + mode_rationale=selected_evaluation.mode_rationale, rationale=summary, candidate_evaluations=evaluations, selection_reason=_selection_reason( @@ -1095,29 +1181,42 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: by_id=by_id, ), winning_factors=winning_factors, - approval_required=selected_evaluation.approval_required, - approval_reason=selected_evaluation.approval_reason, - approval_status=ApprovalStatus.PENDING if selected_evaluation.approval_required else ApprovalStatus.AUTO_APPLIED, - planner_error=planner_error if fallback_used else None, - metadata=final_plan.metadata, - ) - - enrich_plan_and_decision(state=state, event=event, plan=final_plan, decision_log=decision_log) - - state.latest_plan = final_plan - state.latest_plan_id = final_plan.plan_id - state.pending_plan = final_plan if final_plan.approval_required else None - state.decision_logs.append(decision_log) - + approval_required=selected_evaluation.approval_required, + approval_reason=selected_evaluation.approval_reason, + approval_status=ApprovalStatus.PENDING + if selected_evaluation.approval_required + else ApprovalStatus.AUTO_APPLIED, + planner_error=planner_error if fallback_used else None, + metadata=final_plan.metadata, + ) + + enrich_plan_and_decision( + state=state, event=event, plan=final_plan, decision_log=decision_log + ) + + state.latest_plan = final_plan + state.latest_plan_id = final_plan.plan_id + state.pending_plan = final_plan if final_plan.approval_required else None + state.decision_logs.append(decision_log) + proposal.observations.append( f"built {final_plan.plan_id} using {final_plan.strategy_label} with score {final_plan.score:.4f}" ) + planner_trace_summary = decision_log.selection_reason + if selected_evaluation.projection_summary: + planner_trace_summary = ( + f"{planner_trace_summary} {selected_evaluation.projection_summary}" + if planner_trace_summary + else selected_evaluation.projection_summary + ) if fallback_used and planner_error: proposal.risks.append( f"planner {'repair' if repaired_count < len(STRATEGY_ORDER) else 'fallback'} used: {planner_error}" ) - proposal.domain_summary = final_plan.llm_planner_narrative or final_plan.planner_reasoning - proposal.notes_for_planner = decision_log.selection_reason + proposal.domain_summary = ( + final_plan.llm_planner_narrative or final_plan.planner_reasoning + ) + proposal.notes_for_planner = planner_trace_summary proposal.downstream_impacts = winning_factors[:3] if runner_up is not None: proposal.tradeoffs.append( @@ -1128,7 +1227,11 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: "Planner suppressed the no-action option because current inventory stress already exceeds the normal operating tolerance." ) proposal.llm_used = any(evaluation.llm_used for evaluation in evaluations) - proposal.llm_error = planner_error if (fallback_used and planner_error != "llm_disabled") else None + proposal.llm_error = ( + planner_error + if (fallback_used and planner_error != "llm_disabled") + else None + ) logger.info( "planner final selection strategy=%s generated_by=%s llm_used=%s llm_error=%s", final_plan.strategy_label, @@ -1136,5 +1239,5 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: proposal.llm_used, proposal.llm_error or "none", ) - + return proposal diff --git a/agents/risk.py b/agents/risk.py index b8c6680..b81036e 100644 --- a/agents/risk.py +++ b/agents/risk.py @@ -3,6 +3,7 @@ import requests from agents.base import BaseAgent +from core.scenario_scope import direct_scope_summary, resolve_scenario_scope from core.models import AgentProposal, Event, SystemState from policies.modes import select_mode @@ -13,15 +14,60 @@ class RiskAgent(BaseAgent): custom_system_prompt = ( "Role: {agent_name} specialist agent in an autonomous supply chain control tower. " "CRITICAL: Your ONLY goal is to write a comprehensive 'domain_summary' in English " - "summarizing the disruption based on external_api_data. " + "summarizing the disruption for the declared scenario scope first, then only mention external_api_data that is directly relevant to those affected entities. " + "Do not broaden the blast radius beyond scenario_scope. Do not mention unrelated suppliers, routes, or regions just because they appear in external_api_data. " + "If external_api_data is generic or not directly tied to the affected SKUs, treat it as background context and keep the summary centered on scenario_scope. " "Make it sound natural but precise. Do not invent new actions or make up false information. " "Return empty arrays for impacts, tradeoffs, and recommended actions." ) + @staticmethod + def _scenario_brief(event: Event | None, scenario_scope: dict[str, object]) -> str: + if event is None: + return "No active disruption." + affected_skus = scenario_scope.get("affected_skus", []) + route_ids = scenario_scope.get("route_ids", []) + supplier_ids = scenario_scope.get("supplier_ids", []) + if event.type.value == "supplier_delay": + delay_hours = event.payload.get("delay_hours") + return ( + f"Primary scenario: supplier delay at {', '.join(supplier_ids) or 'an unspecified supplier'} with a declared delay of {delay_hours} hours " + f"affecting SKUs {', '.join(affected_skus) if affected_skus else 'not specified'}." + ) + if event.type.value == "demand_spike": + changes = scenario_scope.get("demand_changes", []) + change_text = ", ".join( + f"{item.get('sku')} x{float(item.get('multiplier', 1.0)):.1f}" + for item in changes + if isinstance(item, dict) and item.get("sku") + ) + return ( + "Primary scenario: demand spike affecting only the declared SKU set" + f" ({change_text or ', '.join(affected_skus)})." + ) + if event.type.value == "route_blockage": + return ( + f"Primary scenario: route blockage on {', '.join(route_ids) or 'the declared lanes'} " + f"directly affecting SKUs {', '.join(affected_skus) if affected_skus else 'derived from current route dependencies'}." + ) + if event.type.value == "compound": + return ( + f"Primary scenario: compound disruption across routes {', '.join(route_ids) or 'none declared'} " + f"and suppliers {', '.join(supplier_ids) or 'none declared'}, with direct impact on " + f"{', '.join(affected_skus) if affected_skus else 'the declared entities'}." + ) + return ( + f"Primary scenario: {event.type.value.replace('_', ' ')} affecting " + f"{', '.join(affected_skus) if affected_skus else 'the declared entities'}." + ) + def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: proposal = AgentProposal(agent=self.name) state.mode = select_mode(state, event) api_payloads = {} + resolved_scope = resolve_scenario_scope(state, event) + scenario_scope = resolved_scope.to_dict() + scenario_brief = self._scenario_brief(event, scenario_scope) if event is None: proposal.observations.append( f"Network operating in {state.mode.value} mode with no active trigger event" @@ -38,7 +84,8 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: ) proposal.domain_summary = ( f"{event.type.value.replace('_', ' ').title()} requires risk review for " - f"{', '.join(event.entity_ids) if event.entity_ids else 'the network'}." + f"{', '.join(event.entity_ids) if event.entity_ids else 'the network'}. " + f"{direct_scope_summary(resolved_scope)}" ) # 1. Fetch Weather API @@ -78,6 +125,8 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: "event_severity": event.severity if event else 0.0, "entity_ids": event.entity_ids if event else [], "payload": event.payload if event else {}, + "scenario_scope": scenario_scope, + "scenario_brief": scenario_brief, "active_event_count": len(state.active_events), "external_api_data": api_payloads, }, @@ -93,5 +142,3 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: proposal.tradeoffs.clear() return proposal - - diff --git a/agents/supplier.py b/agents/supplier.py index 6c5f767..35c8734 100644 --- a/agents/supplier.py +++ b/agents/supplier.py @@ -1,33 +1,36 @@ -from __future__ import annotations - -from collections import defaultdict - -from agents.base import BaseAgent -from core.enums import ActionType, EventType -from core.models import Action, AgentProposal, Event, SystemState +from __future__ import annotations + +from collections import defaultdict + +from agents.base import BaseAgent +from core.scenario_scope import payload_supplier_ids, resolve_scenario_scope +from core.enums import ActionType, EventType +from core.models import Action, AgentProposal, Event, SystemState class SupplierAgent(BaseAgent): name = "supplier" custom_system_prompt = """You are the Supplier Management Agent for a supply chain system. When supplier delays or disruptions occur, your job is to analyze alternative suppliers, factoring in reliability, cost, and lead times. -Provide a concise, professional summary (under 4 sentences) of the supplier issues, your evaluation of the alternatives, and the recommended supplier switch. -Write the summary entirely in English. Do NOT use markdown formatting or bullet points in the summary.""" - - def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: - proposal = AgentProposal(agent=self.name) - by_sku: dict[str, list] = defaultdict(list) - for supplier in state.suppliers.values(): - by_sku[supplier.sku].append(supplier) + Provide a concise, professional summary (under 4 sentences) of the supplier issues, your evaluation of the alternatives, and the recommended supplier switch. + Write the summary entirely in English. Do NOT use markdown formatting or bullet points in the summary.""" - delayed_supplier = ( - event.payload.get("supplier_id") - if event and event.type in {EventType.SUPPLIER_DELAY, EventType.COMPOUND} - else None - ) - for sku, item in state.inventory.items(): - current = item.preferred_supplier_id - current_supplier = state.suppliers.get(f"{current}_{sku}") + def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: + proposal = AgentProposal(agent=self.name) + by_sku: dict[str, list] = defaultdict(list) + for supplier in state.suppliers.values(): + by_sku[supplier.sku].append(supplier) + + delayed_suppliers = set(payload_supplier_ids(event)) + scope = resolve_scenario_scope(state, event) + affected_skus = set(scope.supplier_affected_skus) + if event and event.type == EventType.SUPPLIER_DELAY: + affected_skus = affected_skus or set(scope.affected_skus) + for sku, item in state.inventory.items(): + if affected_skus and sku not in affected_skus: + continue + current = item.preferred_supplier_id + current_supplier = state.suppliers.get(f"{current}_{sku}") if current_supplier is None: current_supplier = state.suppliers.get(current) ranked = sorted( @@ -38,38 +41,38 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: supplier.lead_time_days, ), ) - if not ranked: - proposal.risks.append(f"no suppliers available for {sku}") - continue - best = ranked[0] - if best.supplier_id == delayed_supplier and len(ranked) > 1: - best = ranked[1] - if current_supplier and best.supplier_id != current_supplier.supplier_id: - cost_delta = best.unit_cost - current_supplier.unit_cost - proposal.proposals.append( - Action( + if not ranked: + proposal.risks.append(f"no suppliers available for {sku}") + continue + best = ranked[0] + if best.supplier_id in delayed_suppliers and len(ranked) > 1: + best = ranked[1] + if current_supplier and best.supplier_id != current_supplier.supplier_id: + cost_delta = best.unit_cost - current_supplier.unit_cost + proposal.proposals.append( + Action( action_id=f"act_supplier_{sku}_{best.supplier_id}", action_type=ActionType.SWITCH_SUPPLIER, target_id=sku, parameters={"supplier_id": best.supplier_id}, - estimated_cost_delta=max( - cost_delta * max(item.forecast_qty, 1), 0.0 - ), - estimated_service_delta=0.06 - if best.reliability >= current_supplier.reliability - else 0.02, - estimated_risk_delta=-0.12 - if best.supplier_id != delayed_supplier - else -0.02, - estimated_recovery_hours=float(best.lead_time_days * 12), - reason=f"switch {sku} from {current_supplier.supplier_id} to {best.supplier_id}", - priority=0.9 - if delayed_supplier == current_supplier.supplier_id - else 0.55, - ) - ) - proposal.observations.append( - f"{sku} best supplier candidate: {best.supplier_id}" + estimated_cost_delta=max( + cost_delta * max(item.forecast_qty, 1), 0.0 + ), + estimated_service_delta=0.06 + if best.reliability >= current_supplier.reliability + else 0.02, + estimated_risk_delta=-0.12 + if best.supplier_id not in delayed_suppliers + else -0.02, + estimated_recovery_hours=float(best.lead_time_days * 12), + reason=f"switch {sku} from {current_supplier.supplier_id} to {best.supplier_id}", + priority=0.9 + if current_supplier.supplier_id in delayed_suppliers + else 0.55, + ) + ) + proposal.observations.append( + f"{sku} best supplier candidate: {best.supplier_id}" ) if event is not None or proposal.proposals: supplier_snapshot = { @@ -84,14 +87,17 @@ def run(self, state: SystemState, event: Event | None = None) -> AgentProposal: for supplier in ranked_suppliers ] for sku, ranked_suppliers in by_sku.items() + if not affected_skus or sku in affected_skus } self.enrich_with_llm( - state=state, - event=event, - proposal=proposal, - state_slice={ - "delayed_supplier": delayed_supplier, - "supplier_options": supplier_snapshot, - }, - ) - return proposal + state=state, + event=event, + proposal=proposal, + state_slice={ + "delayed_suppliers": sorted(delayed_suppliers), + "affected_skus": sorted(affected_skus), + "scenario_scope": scope.to_dict(), + "supplier_options": supplier_snapshot, + }, + ) + return proposal diff --git a/core/scenario_scope.py b/core/scenario_scope.py new file mode 100644 index 0000000..bb33630 --- /dev/null +++ b/core/scenario_scope.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +from dataclasses import dataclass + +from core.enums import EventType +from core.models import Event, SystemState + + +@dataclass +class ResolvedScenarioScope: + event_type: str + route_ids: list[str] + supplier_ids: list[str] + affected_skus: list[str] + route_affected_skus: list[str] + supplier_affected_skus: list[str] + demand_affected_skus: list[str] + warehouse_ids: list[str] + node_ids: list[str] + demand_changes: list[dict[str, float | str]] + component_types: list[str] + + def to_dict(self) -> dict[str, object]: + return { + "event_type": self.event_type, + "route_ids": list(self.route_ids), + "supplier_ids": list(self.supplier_ids), + "affected_skus": list(self.affected_skus), + "route_affected_skus": list(self.route_affected_skus), + "supplier_affected_skus": list(self.supplier_affected_skus), + "demand_affected_skus": list(self.demand_affected_skus), + "warehouse_ids": list(self.warehouse_ids), + "node_ids": list(self.node_ids), + "demand_changes": list(self.demand_changes), + "component_types": list(self.component_types), + "direct_scope_summary": direct_scope_summary(self), + } + + +def _sorted(values: set[str]) -> list[str]: + return sorted(str(value) for value in values if str(value).strip()) + + +def _list_from_payload( + payload: dict[str, object], + plural_key: str, + singular_key: str | None = None, +) -> list[str]: + values = payload.get(plural_key) + result: list[str] = [] + if isinstance(values, list): + for value in values: + text = str(value).strip() + if text and text not in result: + result.append(text) + if singular_key: + single = str(payload.get(singular_key) or "").strip() + if single and single not in result: + result.append(single) + return result + + +def payload_route_ids(event: Event | None) -> list[str]: + if event is None: + return [] + return _list_from_payload(event.payload, "route_ids", "route_id") + + +def payload_supplier_ids(event: Event | None) -> list[str]: + if event is None: + return [] + return _list_from_payload(event.payload, "supplier_ids", "supplier_id") + + +def payload_affected_skus(event: Event | None) -> list[str]: + if event is None: + return [] + return _list_from_payload(event.payload, "affected_skus", "sku") + + +def payload_demand_changes(event: Event | None) -> list[dict[str, float | str]]: + if event is None: + return [] + raw = event.payload.get("demand_changes") + if isinstance(raw, list) and raw: + normalized: list[dict[str, float | str]] = [] + for item in raw: + if not isinstance(item, dict): + continue + sku = str(item.get("sku") or "").strip() + if not sku: + continue + normalized.append( + { + "sku": sku, + "multiplier": float(item.get("multiplier", 1.0)), + } + ) + if normalized: + return normalized + if event.type in {EventType.DEMAND_SPIKE, EventType.COMPOUND}: + sku = str(event.payload.get("sku") or "").strip() + if sku: + return [ + { + "sku": sku, + "multiplier": float(event.payload.get("multiplier", 1.0)), + } + ] + return [] + + +def direct_scope_summary(scope: ResolvedScenarioScope) -> str: + parts: list[str] = [] + if scope.route_ids: + label = "route" if len(scope.route_ids) == 1 else "routes" + preview = ", ".join(scope.route_ids[:3]) + parts.append(f"{len(scope.route_ids)} {label} ({preview})") + if scope.supplier_ids: + label = "supplier" if len(scope.supplier_ids) == 1 else "suppliers" + preview = ", ".join(scope.supplier_ids[:3]) + parts.append(f"{len(scope.supplier_ids)} {label} ({preview})") + if scope.affected_skus: + preview = ", ".join(scope.affected_skus[:4]) + parts.append(f"{len(scope.affected_skus)} SKU(s) ({preview})") + if scope.warehouse_ids: + parts.append(f"{len(scope.warehouse_ids)} warehouse(s)") + if scope.node_ids: + parts.append(f"{len(scope.node_ids)} node(s)") + if not parts: + return "No direct disruption scope was declared." + return "Directly impacted scope includes " + ", ".join(parts) + "." + + +def resolve_scenario_scope( + state: SystemState, + event: Event | None, +) -> ResolvedScenarioScope: + if event is None: + return ResolvedScenarioScope( + event_type="none", + route_ids=[], + supplier_ids=[], + affected_skus=[], + route_affected_skus=[], + supplier_affected_skus=[], + demand_affected_skus=[], + warehouse_ids=[], + node_ids=[], + demand_changes=[], + component_types=[], + ) + + route_ids = payload_route_ids(event) + supplier_ids = payload_supplier_ids(event) + declared_skus = payload_affected_skus(event) + demand_changes = payload_demand_changes(event) + demand_skus = { + str(item["sku"]) + for item in demand_changes + if str(item.get("sku") or "").strip() + } + route_skus = { + sku + for sku, item in state.inventory.items() + if item.preferred_route_id in route_ids + } + supplier_skus = { + sku + for sku, item in state.inventory.items() + if item.preferred_supplier_id in supplier_ids + } + affected_skus = set(declared_skus) | demand_skus | route_skus | supplier_skus + warehouses = { + state.inventory[sku].warehouse_id + for sku in affected_skus + if sku in state.inventory + } + nodes: set[str] = set() + for route_id in route_ids: + route = state.routes.get(route_id) + if route is None: + continue + nodes.add(route.origin) + nodes.add(route.destination) + + component_types: list[str] = [] + if route_ids: + component_types.append("routing") + if supplier_ids: + component_types.append("sourcing") + if demand_changes: + component_types.append("demand") + + return ResolvedScenarioScope( + event_type=event.type.value, + route_ids=route_ids, + supplier_ids=supplier_ids, + affected_skus=_sorted(affected_skus), + route_affected_skus=_sorted(route_skus), + supplier_affected_skus=_sorted(supplier_skus), + demand_affected_skus=_sorted(demand_skus), + warehouse_ids=_sorted(warehouses), + node_ids=_sorted(nodes), + demand_changes=demand_changes, + component_types=component_types, + ) diff --git a/data/scenarios.json b/data/scenarios.json index 252b62a..3e40de2 100644 --- a/data/scenarios.json +++ b/data/scenarios.json @@ -12,7 +12,7 @@ "event_count": 1 }, "compound_disruption": { - "description": "Supplier delay plus demand spike", - "event_count": 2 + "description": "Supplier delay plus route disruption under demand pressure", + "event_count": 1 } } diff --git a/orchestrator/graph.py b/orchestrator/graph.py index 8095f5b..448efeb 100644 --- a/orchestrator/graph.py +++ b/orchestrator/graph.py @@ -27,6 +27,7 @@ TraceStep, ) from core.runtime_tracking import new_run_id +from core.scenario_scope import resolve_scenario_scope from core.state import recompute_kpis, utc_now from langgraph.graph import END, START, StateGraph from orchestrator.router import ( @@ -80,6 +81,7 @@ def _emit( try: from streaming.event_bus import event_bus from streaming.schemas import ThinkingEvent + event_bus.publish( run_id, ThinkingEvent( @@ -337,9 +339,17 @@ def risk_node(self, graph_state: OrchestrationState) -> OrchestrationState: event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - self._emit(graph_state, type="start", agent="risk", step="init", - message="Starting orchestration cycle", - data={"mode": state.mode.value, "pending_plan": state.pending_plan is not None}) + self._emit( + graph_state, + type="start", + agent="risk", + step="init", + message="Starting orchestration cycle", + data={ + "mode": state.mode.value, + "pending_plan": state.pending_plan is not None, + }, + ) self._reset_cycle(state) self._begin_trace(state, event) @@ -351,29 +361,54 @@ def risk_node(self, graph_state: OrchestrationState) -> OrchestrationState: item.dedupe_key for item in state.active_events }: state.active_events.append(event) - + # Step 1: Mode selection - self._emit(graph_state, type="thinking", agent="risk", step="mode_selection", - message=f"Selecting operating mode based on {len(state.active_events)} active events", - data={"active_events": len(state.active_events)}) + self._emit( + graph_state, + type="thinking", + agent="risk", + step="mode_selection", + message=f"Selecting operating mode based on {len(state.active_events)} active events", + data={"active_events": len(state.active_events)}, + ) # Step 2: External API calls - self._emit(graph_state, type="action", agent="risk", step="api_weather", - message="Fetching external Weather API for disruption intel", - data={"api": "weather", "endpoint": "/mock/weather"}) + self._emit( + graph_state, + type="action", + agent="risk", + step="api_weather", + message="Fetching external Weather API for disruption intel", + data={"api": "weather", "endpoint": "/mock/weather"}, + ) - self._emit(graph_state, type="action", agent="risk", step="api_routes", - message="Fetching external Routes API for transport status", - data={"api": "routes", "endpoint": "/mock/routes"}) + self._emit( + graph_state, + type="action", + agent="risk", + step="api_routes", + message="Fetching external Routes API for transport status", + data={"api": "routes", "endpoint": "/mock/routes"}, + ) - self._emit(graph_state, type="action", agent="risk", step="api_suppliers", - message="Fetching external Suppliers API for supplier health", - data={"api": "suppliers", "endpoint": "/mock/suppliers"}) + self._emit( + graph_state, + type="action", + agent="risk", + step="api_suppliers", + message="Fetching external Suppliers API for supplier health", + data={"api": "suppliers", "endpoint": "/mock/suppliers"}, + ) # Step 3: LLM enrichment - self._emit(graph_state, type="thinking", agent="risk", step="llm_enrichment", - message="Calling LLM to synthesize risk assessment from API data", - data={"capability": "specialist", "purpose": "domain_summary_generation"}) + self._emit( + graph_state, + type="thinking", + agent="risk", + step="llm_enrichment", + message="Calling LLM to synthesize risk assessment from API data", + data={"capability": "specialist", "purpose": "domain_summary_generation"}, + ) max_sev = max((e.severity for e in state.active_events), default=0.0) output = self.risk_agent.run(state, event) @@ -381,21 +416,34 @@ def risk_node(self, graph_state: OrchestrationState) -> OrchestrationState: self._complete_agent_step(state, "risk", output) # Step 4: Results - self._emit(graph_state, type="observation", agent="risk", step="results", - message=f"Risk assessment complete — {len(output.proposals)} mitigations proposed", - data={ - "proposals": len(output.proposals), - "llm_used": output.llm_used, - "llm_error": output.llm_error, - "mode": state.mode.value, - "max_severity": round(max_sev, 2), - "domain_summary": (output.domain_summary or "")[:200], - }) + self._emit( + graph_state, + type="observation", + agent="risk", + step="results", + message=f"Risk assessment complete — {len(output.proposals)} mitigations proposed", + data={ + "proposals": len(output.proposals), + "llm_used": output.llm_used, + "llm_error": output.llm_error, + "mode": state.mode.value, + "max_severity": round(max_sev, 2), + "domain_summary": (output.domain_summary or "")[:200], + }, + ) risk_route = route_after_risk({"state": state}) - self._emit(graph_state, type="decision", agent="risk", step="routing", - message=f"Next routing → {risk_route}", - data={"next_node": risk_route, "reason": self._risk_route_reason(risk_route)}) + self._emit( + graph_state, + type="decision", + agent="risk", + step="routing", + message=f"Next routing → {risk_route}", + data={ + "next_node": risk_route, + "reason": self._risk_route_reason(risk_route), + }, + ) self._record_route( state, @@ -406,41 +454,80 @@ def risk_node(self, graph_state: OrchestrationState) -> OrchestrationState: ) if state.latest_trace is not None: state.latest_trace.current_branch = state.mode.value - + self._notify_trace_update(state, trace_updater) - return {"state": state, "event": event, "started_at": graph_state["started_at"], - "run_id": graph_state.get("run_id"), "trace_updater": trace_updater} + return { + "state": state, + "event": event, + "started_at": graph_state["started_at"], + "run_id": graph_state.get("run_id"), + "trace_updater": trace_updater, + } def demand_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - + scope = resolve_scenario_scope(state, event) + sku_count = len(state.inventory) - self._emit(graph_state, type="start", agent="demand", step="init", - message=f"Demand Agent beginning forecast analysis for {sku_count} SKUs", - data={"sku_count": sku_count}) - + self._emit( + graph_state, + type="start", + agent="demand", + step="init", + message=f"Demand Agent beginning forecast analysis for {sku_count} SKUs", + data={"sku_count": sku_count}, + ) + self._start_step(state, "demand", "agent", event) self._notify_trace_update(state, trace_updater) # Step 1: Check for demand spike event - has_spike = event and event.type.value == "demand_spike" + has_spike = event and event.type.value in {"demand_spike", "compound"} if has_spike: - multiplier = event.payload.get("multiplier", 1.5) - spike_sku = event.payload.get("sku") - self._emit(graph_state, type="analysis", agent="demand", step="spike_detection", - message=f"Demand spike detected: SKU {spike_sku} at {multiplier}x multiplier", - data={"spike_sku": spike_sku, "multiplier": multiplier}) + demand_changes = scope.demand_changes + if demand_changes: + sku_list = [str(item["sku"]) for item in demand_changes] + preview = ", ".join(sku_list[:3]) + self._emit( + graph_state, + type="analysis", + agent="demand", + step="spike_detection", + message=f"Demand spike detected across {len(sku_list)} SKUs: {preview}", + data={"affected_skus": sku_list, "demand_changes": demand_changes}, + ) + else: + multiplier = event.payload.get("multiplier", 1.5) + spike_sku = event.payload.get("sku") + self._emit( + graph_state, + type="analysis", + agent="demand", + step="spike_detection", + message=f"Demand spike detected: SKU {spike_sku} at {multiplier}x multiplier", + data={"spike_sku": spike_sku, "multiplier": multiplier}, + ) else: - self._emit(graph_state, type="analysis", agent="demand", step="routine_forecast", - message="No demand spike — running routine 30-day rolling forecast", - data={"method": "rolling_average_30d"}) + self._emit( + graph_state, + type="analysis", + agent="demand", + step="routine_forecast", + message="No demand spike — running routine 30-day rolling forecast", + data={"method": "rolling_average_30d"}, + ) # Step 2: Statistical computation - self._emit(graph_state, type="thinking", agent="demand", step="statistics", - message="Computing per-SKU average demand, standard deviation, and forecast adjustments", - data={"method": "pandas_rolling_stats", "window": 30}) + self._emit( + graph_state, + type="thinking", + agent="demand", + step="statistics", + message="Computing per-SKU average demand, standard deviation, and forecast adjustments", + data={"method": "pandas_rolling_stats", "window": 30}, + ) output = self.demand_agent.run(state, event) self._record_output(state, output) @@ -448,21 +535,42 @@ def demand_node(self, graph_state: OrchestrationState) -> OrchestrationState: # Step 3: LLM enrichment (if triggered) if output.llm_used: - self._emit(graph_state, type="observation", agent="demand", step="llm_result", - message="LLM enriched demand summary with contextual analysis", - data={"llm_used": True, "domain_summary": (output.domain_summary or "")[:200]}) + self._emit( + graph_state, + type="observation", + agent="demand", + step="llm_result", + message="LLM enriched demand summary with contextual analysis", + data={ + "llm_used": True, + "domain_summary": (output.domain_summary or "")[:200], + }, + ) else: - self._emit(graph_state, type="observation", agent="demand", step="deterministic_result", - message="Demand analysis completed using deterministic forecasting", - data={"llm_used": False, "llm_error": output.llm_error, - "proposals": len(output.proposals), - "observations": output.observations[:3]}) + self._emit( + graph_state, + type="observation", + agent="demand", + step="deterministic_result", + message="Demand analysis completed using deterministic forecasting", + data={ + "llm_used": False, + "llm_error": output.llm_error, + "proposals": len(output.proposals), + "observations": output.observations[:3], + }, + ) next_node = route_after_demand({"state": state}) - self._emit(graph_state, type="decision", agent="demand", step="routing", - message=f"Demand analysis completed — routing to {next_node}", - data={"next_node": next_node, "proposals": len(output.proposals)}) - + self._emit( + graph_state, + type="decision", + agent="demand", + step="routing", + message=f"Demand analysis completed — routing to {next_node}", + data={"next_node": next_node, "proposals": len(output.proposals)}, + ) + self._record_route( state, "demand", @@ -477,31 +585,63 @@ def inventory_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - + # Step 1: Initial scan + scope = resolve_scenario_scope(state, event) + inventory_focus = set(scope.affected_skus) + scoped_items = [ + item + for sku, item in state.inventory.items() + if not inventory_focus or sku in inventory_focus + ] critical_skus = sum( - 1 for item in state.inventory.values() + 1 + for item in scoped_items if item.on_hand + item.incoming_qty - item.forecast_qty < item.reorder_point ) below_safety = sum( - 1 for item in state.inventory.values() + 1 + for item in scoped_items if item.on_hand + item.incoming_qty - item.forecast_qty < item.safety_stock ) - self._emit(graph_state, type="start", agent="inventory", step="init", - message=f"Inventory Agent checking {len(state.inventory)} SKUs", - data={"sku_count": len(state.inventory), "critical_skus": critical_skus, "below_safety": below_safety}) - + self._emit( + graph_state, + type="start", + agent="inventory", + step="init", + message=f"Inventory Agent checking {len(state.inventory)} SKUs", + data={ + "sku_count": len(scoped_items), + "critical_skus": critical_skus, + "below_safety": below_safety, + "direct_scope_skus": len(inventory_focus), + }, + ) + self._start_step(state, "inventory", "agent", event) self._notify_trace_update(state, trace_updater) # Step 2: Computational analysis - self._emit(graph_state, type="thinking", agent="inventory", step="rop_calculation", - message="Computing Reorder Point (ROP), Safety Stock (SS), and Days of Cover for each SKU", - data={"method": "z_score_based_safety_stock", "formula": "ROP = LTD + z * σ * √LT"}) + self._emit( + graph_state, + type="thinking", + agent="inventory", + step="rop_calculation", + message="Computing Reorder Point (ROP), Safety Stock (SS), and Days of Cover for each SKU", + data={ + "method": "z_score_based_safety_stock", + "formula": "ROP = LTD + z * σ * √LT", + }, + ) - self._emit(graph_state, type="thinking", agent="inventory", step="replenishment_check", - message="Identifying SKUs that need replenishment by comparing projected stock vs ROP", - data={"comparison": "projected_stock < reorder_point"}) + self._emit( + graph_state, + type="thinking", + agent="inventory", + step="replenishment_check", + message="Identifying SKUs that need replenishment by comparing projected stock vs ROP", + data={"comparison": "projected_stock < reorder_point"}, + ) output = self.inventory_agent.run(state, event) self._record_output(state, output) @@ -510,22 +650,44 @@ def inventory_node(self, graph_state: OrchestrationState) -> OrchestrationState: # Step 3: Results reorder_count = len(output.proposals) if output.llm_used: - self._emit(graph_state, type="observation", agent="inventory", step="llm_result", - message=f"LLM enriched inventory analysis — {reorder_count} reorder proposals generated", - data={"llm_used": True, "reorder_proposals": reorder_count, - "domain_summary": (output.domain_summary or "")[:200]}) + self._emit( + graph_state, + type="observation", + agent="inventory", + step="llm_result", + message=f"LLM enriched inventory analysis — {reorder_count} reorder proposals generated", + data={ + "llm_used": True, + "reorder_proposals": reorder_count, + "domain_summary": (output.domain_summary or "")[:200], + }, + ) else: - self._emit(graph_state, type="observation", agent="inventory", step="results", - message=f"Inventory analysis complete — {reorder_count} reorder proposals, {below_safety} SKUs below safety stock", - data={"llm_used": False, "llm_error": output.llm_error, - "reorder_proposals": reorder_count, "below_safety": below_safety, - "observations": output.observations[:3]}) + self._emit( + graph_state, + type="observation", + agent="inventory", + step="results", + message=f"Inventory analysis complete — {reorder_count} reorder proposals, {below_safety} SKUs below safety stock", + data={ + "llm_used": False, + "llm_error": output.llm_error, + "reorder_proposals": reorder_count, + "below_safety": below_safety, + "observations": output.observations[:3], + }, + ) next_node = route_after_inventory({"state": state}) - self._emit(graph_state, type="decision", agent="inventory", step="routing", - message=f"Inventory planning done — routing to {next_node}", - data={"next_node": next_node}) - + self._emit( + graph_state, + type="decision", + agent="inventory", + step="routing", + message=f"Inventory planning done — routing to {next_node}", + data={"next_node": next_node}, + ) + self._record_route( state, "inventory", @@ -540,30 +702,55 @@ def supplier_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - + supplier_count = len(state.suppliers) - self._emit(graph_state, type="start", agent="supplier", step="init", - message=f"Supplier Agent evaluating {supplier_count} suppliers", - data={"supplier_count": supplier_count}) - + scope = resolve_scenario_scope(state, event) + self._emit( + graph_state, + type="start", + agent="supplier", + step="init", + message=f"Supplier Agent evaluating {supplier_count} suppliers", + data={"supplier_count": supplier_count}, + ) + self._start_step(state, "supplier", "agent", event) self._notify_trace_update(state, trace_updater) # Step 1: Detecting delayed suppliers - delayed_supplier = event.payload.get("supplier_id") if event and event.type.value == "supplier_delay" else None - if delayed_supplier: - self._emit(graph_state, type="analysis", agent="supplier", step="delay_detection", - message=f"Supplier delay detected: {delayed_supplier}", - data={"delayed_supplier": delayed_supplier, "event_type": event.type.value}) + delayed_suppliers = scope.supplier_ids + if delayed_suppliers: + self._emit( + graph_state, + type="analysis", + agent="supplier", + step="delay_detection", + message=f"Supplier disruption detected across {len(delayed_suppliers)} supplier(s) impacting {len(scope.supplier_affected_skus or scope.affected_skus)} SKU(s)", + data={ + "delayed_suppliers": delayed_suppliers, + "event_type": event.type.value, + "affected_skus": scope.supplier_affected_skus or scope.affected_skus, + }, + ) else: - self._emit(graph_state, type="analysis", agent="supplier", step="routine_eval", - message="No supplier disruption — running routine supplier ranking", - data={"method": "reliability_cost_leadtime_sort"}) + self._emit( + graph_state, + type="analysis", + agent="supplier", + step="routine_eval", + message="No supplier disruption — running routine supplier ranking", + data={"method": "reliability_cost_leadtime_sort"}, + ) # Step 2: Ranking and switching - self._emit(graph_state, type="thinking", agent="supplier", step="ranking", - message="Ranking suppliers by reliability, unit cost, and lead time for each SKU", - data={"criteria": ["reliability", "unit_cost", "lead_time_days"]}) + self._emit( + graph_state, + type="thinking", + agent="supplier", + step="ranking", + message="Ranking suppliers by reliability, unit cost, and lead time for each SKU", + data={"criteria": ["reliability", "unit_cost", "lead_time_days"]}, + ) output = self.supplier_agent.run(state, event) self._record_output(state, output) @@ -571,22 +758,43 @@ def supplier_node(self, graph_state: OrchestrationState) -> OrchestrationState: # Step 3: Results if output.llm_used: - self._emit(graph_state, type="observation", agent="supplier", step="llm_result", - message=f"LLM enriched supplier analysis — {len(output.proposals)} switch proposals", - data={"llm_used": True, "proposals": len(output.proposals), - "domain_summary": (output.domain_summary or "")[:200]}) + self._emit( + graph_state, + type="observation", + agent="supplier", + step="llm_result", + message=f"LLM enriched supplier analysis — {len(output.proposals)} switch proposals", + data={ + "llm_used": True, + "proposals": len(output.proposals), + "domain_summary": (output.domain_summary or "")[:200], + }, + ) else: - self._emit(graph_state, type="observation", agent="supplier", step="results", - message=f"Supplier Agent proposed {len(output.proposals)} supplier switches", - data={"llm_used": False, "llm_error": output.llm_error, - "proposals": len(output.proposals), - "observations": output.observations[:3]}) + self._emit( + graph_state, + type="observation", + agent="supplier", + step="results", + message=f"Supplier Agent proposed {len(output.proposals)} supplier switches", + data={ + "llm_used": False, + "llm_error": output.llm_error, + "proposals": len(output.proposals), + "observations": output.observations[:3], + }, + ) next_node = route_after_supplier({"state": state}) - self._emit(graph_state, type="decision", agent="supplier", step="routing", - message=f"Supplier evaluation done — routing to {next_node}", - data={"next_node": next_node}) - + self._emit( + graph_state, + type="decision", + agent="supplier", + step="routing", + message=f"Supplier evaluation done — routing to {next_node}", + data={"next_node": next_node}, + ) + self._record_route( state, "supplier", @@ -601,31 +809,61 @@ def logistics_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - + route_count = len(state.routes) - self._emit(graph_state, type="start", agent="logistics", step="init", - message=f"Logistics Agent checking {route_count} transport routes", - data={"route_count": route_count}) - + scope = resolve_scenario_scope(state, event) + self._emit( + graph_state, + type="start", + agent="logistics", + step="init", + message=f"Logistics Agent checking {route_count} transport routes", + data={"route_count": route_count}, + ) + self._start_step(state, "logistics", "agent", event) self._notify_trace_update(state, trace_updater) # Step 1: Route blockage detection - blocked_route = event.payload.get("route_id") if event and event.type.value == "route_blockage" else None - if blocked_route: - self._emit(graph_state, type="analysis", agent="logistics", step="blockage_detection", - message=f"Route blockage detected on {blocked_route}", - data={"blocked_route": blocked_route, "reason": event.payload.get("reason", "unknown")}) + blocked_routes = scope.route_ids + if blocked_routes: + self._emit( + graph_state, + type="analysis", + agent="logistics", + step="blockage_detection", + message=f"Route blockage detected across {len(blocked_routes)} lane(s) affecting {len(scope.route_affected_skus)} SKU(s)", + data={ + "blocked_routes": blocked_routes, + "affected_skus": scope.route_affected_skus, + "warehouse_ids": scope.warehouse_ids, + "node_ids": scope.node_ids, + "reason": event.payload.get("reason", "unknown"), + }, + ) else: - self._emit(graph_state, type="analysis", agent="logistics", step="routine_check", - message="No route blockage — optimizing transport routes", - data={"method": "risk_transit_cost_sort"}) + self._emit( + graph_state, + type="analysis", + agent="logistics", + step="routine_check", + message="No route blockage — optimizing transport routes", + data={"method": "risk_transit_cost_sort"}, + ) # Step 2: Route ranking available = sum(1 for r in state.routes.values() if r.status != "blocked") - self._emit(graph_state, type="thinking", agent="logistics", step="route_ranking", - message=f"Ranking {available} available routes by risk score, transit time, and cost", - data={"available_routes": available, "criteria": ["risk_score", "transit_days", "cost"]}) + self._emit( + graph_state, + type="thinking", + agent="logistics", + step="route_ranking", + message=f"Ranking {available} available routes by risk score, transit time, and cost", + data={ + "available_routes": available, + "criteria": ["risk_score", "transit_days", "cost"], + }, + ) output = self.logistics_agent.run(state, event) self._record_output(state, output) @@ -633,22 +871,43 @@ def logistics_node(self, graph_state: OrchestrationState) -> OrchestrationState: # Step 3: Results if output.llm_used: - self._emit(graph_state, type="observation", agent="logistics", step="llm_result", - message=f"LLM enriched logistics analysis — {len(output.proposals)} reroute proposals", - data={"llm_used": True, "proposals": len(output.proposals), - "domain_summary": (output.domain_summary or "")[:200]}) + self._emit( + graph_state, + type="observation", + agent="logistics", + step="llm_result", + message=f"LLM enriched logistics analysis — {len(output.proposals)} reroute proposals", + data={ + "llm_used": True, + "proposals": len(output.proposals), + "domain_summary": (output.domain_summary or "")[:200], + }, + ) else: - self._emit(graph_state, type="observation", agent="logistics", step="results", - message=f"Logistics Agent proposed {len(output.proposals)} reroute actions", - data={"llm_used": False, "llm_error": output.llm_error, - "proposals": len(output.proposals), - "observations": output.observations[:3]}) + self._emit( + graph_state, + type="observation", + agent="logistics", + step="results", + message=f"Logistics Agent proposed {len(output.proposals)} reroute actions", + data={ + "llm_used": False, + "llm_error": output.llm_error, + "proposals": len(output.proposals), + "observations": output.observations[:3], + }, + ) next_node = route_after_logistics({"state": state}) - self._emit(graph_state, type="decision", agent="logistics", step="routing", - message=f"Logistics planning done — routing to {next_node}", - data={"next_node": next_node}) - + self._emit( + graph_state, + type="decision", + agent="logistics", + step="routing", + message=f"Logistics planning done — routing to {next_node}", + data={"next_node": next_node}, + ) + self._record_route( state, "logistics", @@ -663,34 +922,70 @@ def planner_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - + candidate_count = len(state.candidate_actions) - self._emit(graph_state, type="start", agent="planner", step="init", - message=f"Planner starting with {candidate_count} candidate actions from specialist agents", - data={"candidate_actions": candidate_count, "mode": state.mode.value}) + self._emit( + graph_state, + type="start", + agent="planner", + step="init", + message=f"Planner starting with {candidate_count} candidate actions from specialist agents", + data={"candidate_actions": candidate_count, "mode": state.mode.value}, + ) self._start_step(state, "planner", "agent", event) self._notify_trace_update(state, trace_updater) # Step 1: Feasibility filtering - self._emit(graph_state, type="thinking", agent="planner", step="feasibility_filter", - message="Filtering candidate actions through hard constraints", - data={"method": "evaluate_hard_constraints", "input_count": candidate_count}) + self._emit( + graph_state, + type="thinking", + agent="planner", + step="feasibility_filter", + message="Filtering candidate actions through hard constraints", + data={ + "method": "evaluate_hard_constraints", + "input_count": candidate_count, + }, + ) # Step 2: Strategy generation - self._emit(graph_state, type="thinking", agent="planner", step="strategy_generation", - message="Generating 3 strategies: cost_first, balanced, resilience_first", - data={"strategies": ["cost_first", "balanced", "resilience_first"]}) + self._emit( + graph_state, + type="thinking", + agent="planner", + step="strategy_generation", + message="Generating 3 strategies: cost_first, balanced, resilience_first", + data={"strategies": ["cost_first", "balanced", "resilience_first"]}, + ) # Step 3: LLM or deterministic - self._emit(graph_state, type="action", agent="planner", step="candidate_drafting", - message="Calling LLM to draft candidate plans (falling back to deterministic if unavailable)", - data={"capability": "planner", "fallback": "deterministic_sort"}) + self._emit( + graph_state, + type="action", + agent="planner", + step="candidate_drafting", + message="Calling LLM to draft candidate plans (falling back to deterministic if unavailable)", + data={"capability": "planner", "fallback": "deterministic_sort"}, + ) # Step 4: Simulation - self._emit(graph_state, type="thinking", agent="planner", step="simulation", - message="Running multi-horizon simulations for each strategy to project KPIs", - data={"method": "evaluate_candidate_plan", "metrics": ["service_level", "cost", "disruption_risk", "recovery_speed"]}) + self._emit( + graph_state, + type="thinking", + agent="planner", + step="simulation", + message="Running multi-horizon simulations for each strategy to project KPIs", + data={ + "method": "evaluate_candidate_plan", + "metrics": [ + "service_level", + "cost", + "disruption_risk", + "recovery_speed", + ], + }, + ) output = self.planner_agent.run(state, event) self._record_output(state, output) @@ -756,7 +1051,8 @@ def planner_node(self, graph_state: OrchestrationState) -> OrchestrationState: "summary": step.summary, } for item in latest_decision.candidate_evaluations - if state.latest_plan and item.strategy_label == state.latest_plan.strategy_label + if state.latest_plan + and item.strategy_label == state.latest_plan.strategy_label for step in item.projection_steps ] if latest_decision and state.latest_plan @@ -768,21 +1064,34 @@ def planner_node(self, graph_state: OrchestrationState) -> OrchestrationState: # Step 5: Report results if state.latest_plan is not None: generated_by = state.latest_plan.generated_by or "unknown" - self._emit(graph_state, type="observation", agent="planner", step="generation_method", - message=f"Plan generated by: {generated_by}", - data={"generated_by": generated_by, "llm_used": output.llm_used, - "llm_error": output.llm_error}) - - self._emit(graph_state, type="decision", agent="planner", step="plan_selected", - message=f"Selected strategy '{state.latest_plan.strategy_label or 'unknown'}' " - f"(score={state.latest_plan.score:.3f})", - data={ - "plan_id": state.latest_plan.plan_id, - "strategy": state.latest_plan.strategy_label, - "score": round(state.latest_plan.score, 4), - "approval_required": state.latest_plan.approval_required, - "action_count": len(state.latest_plan.actions), - }) + self._emit( + graph_state, + type="observation", + agent="planner", + step="generation_method", + message=f"Plan generated by: {generated_by}", + data={ + "generated_by": generated_by, + "llm_used": output.llm_used, + "llm_error": output.llm_error, + }, + ) + + self._emit( + graph_state, + type="decision", + agent="planner", + step="plan_selected", + message=f"Selected strategy '{state.latest_plan.strategy_label or 'unknown'}' " + f"(score={state.latest_plan.score:.3f})", + data={ + "plan_id": state.latest_plan.plan_id, + "strategy": state.latest_plan.strategy_label, + "score": round(state.latest_plan.score, 4), + "approval_required": state.latest_plan.approval_required, + "action_count": len(state.latest_plan.actions), + }, + ) self._record_route( state, "planner", @@ -797,23 +1106,37 @@ def critic_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - - plan_label = state.latest_plan.strategy_label if state.latest_plan else "unknown" - self._emit(graph_state, type="start", agent="critic", step="init", - message=f"Critic starting review of plan '{plan_label}'", - data={ - "plan_id": state.latest_plan.plan_id if state.latest_plan else None, - "strategy": plan_label, - "action_count": len(state.latest_plan.actions) if state.latest_plan else 0, - }) - + + plan_label = ( + state.latest_plan.strategy_label if state.latest_plan else "unknown" + ) + self._emit( + graph_state, + type="start", + agent="critic", + step="init", + message=f"Critic starting review of plan '{plan_label}'", + data={ + "plan_id": state.latest_plan.plan_id if state.latest_plan else None, + "strategy": plan_label, + "action_count": len(state.latest_plan.actions) + if state.latest_plan + else 0, + }, + ) + self._start_step(state, "critic", "agent", event) self._notify_trace_update(state, trace_updater) # Step 1: LLM critique - self._emit(graph_state, type="action", agent="critic", step="llm_critique", - message="Calling LLM to critique the selected plan against alternatives", - data={"capability": "critic", "purpose": "plan_review_and_findings"}) + self._emit( + graph_state, + type="action", + agent="critic", + step="llm_critique", + message="Calling LLM to critique the selected plan against alternatives", + data={"capability": "critic", "purpose": "plan_review_and_findings"}, + ) output = self.critic_agent.run(state, event) state.agent_outputs[output.agent] = output @@ -846,31 +1169,56 @@ def critic_node(self, graph_state: OrchestrationState) -> OrchestrationState: # Step 2: Report results critic_route = route_after_critic({"state": state}) finding_count = ( - len(state.decision_logs[-1].critic_findings) - if state.decision_logs else 0 + len(state.decision_logs[-1].critic_findings) if state.decision_logs else 0 ) critic_summary = ( state.decision_logs[-1].critic_summary - if state.decision_logs else "no findings" + if state.decision_logs + else "no findings" ) if output.llm_used: - self._emit(graph_state, type="observation", agent="critic", step="llm_result", - message=f"LLM critique completed — {finding_count} finding(s)", - data={"llm_used": True, "finding_count": finding_count, - "critic_summary": (critic_summary or "")[:200]}) + self._emit( + graph_state, + type="observation", + agent="critic", + step="llm_result", + message=f"LLM critique completed — {finding_count} finding(s)", + data={ + "llm_used": True, + "finding_count": finding_count, + "critic_summary": (critic_summary or "")[:200], + }, + ) else: - self._emit(graph_state, type="observation", agent="critic", step="deterministic_result", - message=f"Deterministic review completed — {finding_count} finding(s)", - data={"llm_used": False, "llm_error": output.llm_error, - "finding_count": finding_count, - "critic_summary": (critic_summary or "")[:200]}) - - self._emit(graph_state, type="decision", agent="critic", step="routing", - message=f"Critic routing → {critic_route}", - data={"next_node": critic_route, - "reason": self._critic_route_reason(state, critic_route), - "approval_required": state.latest_plan.approval_required if state.latest_plan else False}) + self._emit( + graph_state, + type="observation", + agent="critic", + step="deterministic_result", + message=f"Deterministic review completed — {finding_count} finding(s)", + data={ + "llm_used": False, + "llm_error": output.llm_error, + "finding_count": finding_count, + "critic_summary": (critic_summary or "")[:200], + }, + ) + + self._emit( + graph_state, + type="decision", + agent="critic", + step="routing", + message=f"Critic routing → {critic_route}", + data={ + "next_node": critic_route, + "reason": self._critic_route_reason(state, critic_route), + "approval_required": state.latest_plan.approval_required + if state.latest_plan + else False, + }, + ) self._record_route( state, "critic", @@ -885,24 +1233,35 @@ def approval_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - - approval_reason = state.latest_plan.approval_reason if state.latest_plan else "approval required" - decision_id = state.decision_logs[-1].decision_id if state.decision_logs else None - - self._emit(graph_state, type="decision", agent="system", step="approval_gate", - message=f"Plan awaiting manual approval: {approval_reason}", - data={"reason": approval_reason, "decision_id": decision_id}) - + + approval_reason = ( + state.latest_plan.approval_reason + if state.latest_plan + else "approval required" + ) + decision_id = ( + state.decision_logs[-1].decision_id if state.decision_logs else None + ) + + self._emit( + graph_state, + type="decision", + agent="system", + step="approval_gate", + message=f"Plan awaiting manual approval: {approval_reason}", + data={"reason": approval_reason, "decision_id": decision_id}, + ) + self._start_step(state, "approval", "gate", event) self._notify_trace_update(state, trace_updater) - + state.mode = Mode.APPROVAL self._update_trace_from_plan(state) if state.latest_trace is not None: state.latest_trace.terminal_stage = "approval" state.latest_trace.execution_status = "pending_approval" state.latest_trace.approval_pending = True - + self._complete_step( state, node_key="approval", @@ -917,14 +1276,19 @@ def approval_node(self, graph_state: OrchestrationState) -> OrchestrationState: }, ) state = self._finalize(state, graph_state["started_at"]) - - self._emit(graph_state, type="final", agent="system", step="complete", - message="Orchestration complete — plan awaiting approval", - data={ - "execution_status": "pending_approval", - "plan_id": state.latest_plan.plan_id if state.latest_plan else None, - "mode": state.mode.value, - }) + + self._emit( + graph_state, + type="final", + agent="system", + step="complete", + message="Orchestration complete — plan awaiting approval", + data={ + "execution_status": "pending_approval", + "plan_id": state.latest_plan.plan_id if state.latest_plan else None, + "mode": state.mode.value, + }, + ) self._notify_trace_update(state, trace_updater) return { "state": state, @@ -938,16 +1302,25 @@ def execution_node(self, graph_state: OrchestrationState) -> OrchestrationState: state = graph_state["state"] event = graph_state.get("event") trace_updater = graph_state.get("trace_updater") - + plan_id = state.latest_plan.plan_id if state.latest_plan else None - self._emit(graph_state, type="action", agent="system", step="execute", - message=f"Executing plan '{plan_id or 'N/A'}'", - data={"plan_id": plan_id, - "action_count": len(state.latest_plan.actions) if state.latest_plan else 0}) - + self._emit( + graph_state, + type="action", + agent="system", + step="execute", + message=f"Executing plan '{plan_id or 'N/A'}'", + data={ + "plan_id": plan_id, + "action_count": len(state.latest_plan.actions) + if state.latest_plan + else 0, + }, + ) + self._start_step(state, "execution", "execution", event) self._notify_trace_update(state, trace_updater) - + execution_summary = "no plan executed" execution_status = "no_op" if state.latest_plan and not state.latest_plan.approval_required: @@ -961,13 +1334,13 @@ def execution_node(self, graph_state: OrchestrationState) -> OrchestrationState: f"{state.latest_plan.strategy_label or 'unlabeled'} strategy" ) execution_status = "auto_applied" - + self._update_trace_from_plan(state) if state.latest_trace is not None: state.latest_trace.terminal_stage = "execution" state.latest_trace.execution_status = execution_status state.latest_trace.approval_pending = False - + self._complete_step( state, node_key="execution", @@ -980,17 +1353,27 @@ def execution_node(self, graph_state: OrchestrationState) -> OrchestrationState: }, ) state = self._finalize(state, graph_state["started_at"]) - - self._emit(graph_state, type="final", agent="system", step="complete", - message=f"Orchestration complete — {execution_status}", - data={ - "execution_status": execution_status, - "plan_id": state.latest_plan_id, - "mode": state.mode.value, - }) + + self._emit( + graph_state, + type="final", + agent="system", + step="complete", + message=f"Orchestration complete — {execution_status}", + data={ + "execution_status": execution_status, + "plan_id": state.latest_plan_id, + "mode": state.mode.value, + }, + ) self._notify_trace_update(state, trace_updater) - return {"state": state, "event": event, "started_at": graph_state["started_at"], - "run_id": graph_state.get("run_id"), "trace_updater": trace_updater} + return { + "state": state, + "event": event, + "started_at": graph_state["started_at"], + "run_id": graph_state.get("run_id"), + "trace_updater": trace_updater, + } def _compile(self): graph = StateGraph(OrchestrationState) @@ -1020,6 +1403,7 @@ def _compile(self): "logistics", route_after_logistics, { + "inventory": "inventory", "supplier": "supplier", "planner": "planner", }, @@ -1028,6 +1412,8 @@ def _compile(self): "supplier", route_after_supplier, { + "demand": "demand", + "inventory": "inventory", "logistics": "logistics", "planner": "planner", }, diff --git a/orchestrator/router.py b/orchestrator/router.py index 918637d..055068e 100644 --- a/orchestrator/router.py +++ b/orchestrator/router.py @@ -2,57 +2,61 @@ from typing import Mapping, cast -from core.enums import Mode, EventType -from core.models import SystemState +from core.enums import Mode, EventType +from core.models import SystemState def _state_from_graph(graph_state: Mapping[str, object]) -> SystemState: return cast(SystemState, graph_state["state"]) -def route_after_risk(graph_state: Mapping[str, object]) -> str: - state = _state_from_graph(graph_state) - if state.pending_plan is not None or state.mode == Mode.APPROVAL: - return "approval" - - event_type = state.active_events[-1].type if state.active_events else None - - if event_type in [EventType.ROUTE_BLOCKAGE, EventType.COMPOUND]: - return "logistics" - elif event_type == EventType.SUPPLIER_DELAY: - return "supplier" - elif event_type in {EventType.DEMAND_SPIKE, None}: - return "demand" - - return "demand" - - +def route_after_risk(graph_state: Mapping[str, object]) -> str: + state = _state_from_graph(graph_state) + if state.pending_plan is not None or state.mode == Mode.APPROVAL: + return "approval" + + event_type = state.active_events[-1].type if state.active_events else None + + if event_type in [EventType.ROUTE_BLOCKAGE, EventType.COMPOUND]: + return "logistics" + elif event_type == EventType.SUPPLIER_DELAY: + return "supplier" + elif event_type in {EventType.DEMAND_SPIKE, None}: + return "demand" + + return "demand" + + def route_after_logistics(graph_state: Mapping[str, object]) -> str: state = _state_from_graph(graph_state) event_type = state.active_events[-1].type if state.active_events else None - if event_type in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: + if event_type == EventType.ROUTE_BLOCKAGE: + return "inventory" + if event_type == EventType.COMPOUND: return "supplier" return "planner" - - + + def route_after_supplier(graph_state: Mapping[str, object]) -> str: state = _state_from_graph(graph_state) event_type = state.active_events[-1].type if state.active_events else None if event_type == EventType.SUPPLIER_DELAY: - return "logistics" - return "planner" - - -def route_after_demand(graph_state: Mapping[str, object]) -> str: - return "inventory" - - -def route_after_inventory(graph_state: Mapping[str, object]) -> str: + return "inventory" + if event_type == EventType.COMPOUND: + return "demand" return "planner" - - -def route_after_planner(graph_state: Mapping[str, object]) -> str: - return "critic" + + +def route_after_demand(graph_state: Mapping[str, object]) -> str: + return "inventory" + + +def route_after_inventory(graph_state: Mapping[str, object]) -> str: + return "planner" + + +def route_after_planner(graph_state: Mapping[str, object]) -> str: + return "critic" def route_after_critic(graph_state: Mapping[str, object]) -> str: diff --git a/orchestrator/service.py b/orchestrator/service.py index 78147df..0d48d94 100644 --- a/orchestrator/service.py +++ b/orchestrator/service.py @@ -207,17 +207,19 @@ def _build_safer_plan(state: SystemState, decision_log: DecisionLog) -> Plan: score_breakdown=breakdown, strategy_label="safer_alternative", generated_by="operator_safer_request", - planner_reasoning=build_plan_summary( - decision_log.before_kpis, - projection.projected_kpis, - breakdown, - selected_actions=selected_actions, - strategy_label="safer_alternative", - mode=target_mode, - projection_summary=projection.projection_summary, - simulation_horizon_days=projection.simulation_horizon_days, - worst_case_kpis=projection.worst_case_kpis, - ), + planner_reasoning=build_plan_summary( + decision_log.before_kpis, + projection.projected_kpis, + breakdown, + selected_actions=selected_actions, + strategy_label="safer_alternative", + mode=target_mode, + projection_summary=projection.projection_summary, + simulation_horizon_days=projection.simulation_horizon_days, + worst_case_kpis=projection.worst_case_kpis, + state=state, + event=_latest_event(state), + ), status=PlanStatus.PROPOSED, ) soft_violations = evaluate_soft_constraints(plan, state) @@ -277,18 +279,20 @@ def _build_alternative_plan( strategy_label=evaluation.strategy_label, generated_by="operator_selected_alternative", planner_reasoning=evaluation.rationale - or build_plan_summary( - decision_log.before_kpis, - projection.projected_kpis, - breakdown, - selected_actions=actions, - strategy_label=evaluation.strategy_label, - mode=target_mode, - mode_rationale=evaluation.mode_rationale, - projection_summary=evaluation.projection_summary, - simulation_horizon_days=evaluation.simulation_horizon_days, - worst_case_kpis=evaluation.worst_case_kpis, - ), + or build_plan_summary( + decision_log.before_kpis, + projection.projected_kpis, + breakdown, + selected_actions=actions, + strategy_label=evaluation.strategy_label, + mode=target_mode, + mode_rationale=evaluation.mode_rationale, + projection_summary=evaluation.projection_summary, + simulation_horizon_days=evaluation.simulation_horizon_days, + worst_case_kpis=evaluation.worst_case_kpis, + state=state, + event=_latest_event(state), + ), status=PlanStatus.PROPOSED, feasible=evaluation.feasible, violations=list(evaluation.violations), diff --git a/policies/constraints.py b/policies/constraints.py index 2b7e4d0..d08eab4 100644 --- a/policies/constraints.py +++ b/policies/constraints.py @@ -1,9 +1,10 @@ -from __future__ import annotations - -from abc import ABC, abstractmethod - -from core.enums import ActionType, ConstraintViolationCode, EventType, Mode -from core.models import Event, Plan, SystemState, SupplierRecord, ConstraintViolation as Violation +from __future__ import annotations + +from abc import ABC, abstractmethod + +from core.scenario_scope import payload_route_ids, payload_supplier_ids +from core.enums import ActionType, ConstraintViolationCode, EventType, Mode +from core.models import Event, Plan, SystemState, SupplierRecord, ConstraintViolation as Violation # --------------------------------------------------------------------------- @@ -36,26 +37,26 @@ def mode_rationale(state: SystemState, event: Event | None) -> str: # Context helpers # --------------------------------------------------------------------------- -def _blocked_routes(state: SystemState, event: Event | None) -> set[str]: - blocked = { - item.payload.get("route_id") - for item in state.active_events - if item.type in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND} - } - if event is not None and event.type in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: - blocked.add(event.payload.get("route_id")) - return {item for item in blocked if item} - - -def _delayed_suppliers(state: SystemState, event: Event | None) -> set[str]: - delayed = { - item.payload.get("supplier_id") - for item in state.active_events - if item.type in {EventType.SUPPLIER_DELAY, EventType.COMPOUND} - } - if event is not None and event.type in {EventType.SUPPLIER_DELAY, EventType.COMPOUND}: - delayed.add(event.payload.get("supplier_id")) - return {item for item in delayed if item} +def _blocked_routes(state: SystemState, event: Event | None) -> set[str]: + blocked: set[str] = set() + for item in state.active_events: + if item.type not in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: + continue + blocked.update(payload_route_ids(item)) + if event is not None and event.type in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: + blocked.update(payload_route_ids(event)) + return {item for item in blocked if item} + + +def _delayed_suppliers(state: SystemState, event: Event | None) -> set[str]: + delayed: set[str] = set() + for item in state.active_events: + if item.type not in {EventType.SUPPLIER_DELAY, EventType.COMPOUND}: + continue + delayed.update(payload_supplier_ids(item)) + if event is not None and event.type in {EventType.SUPPLIER_DELAY, EventType.COMPOUND}: + delayed.update(payload_supplier_ids(event)) + return {item for item in delayed if item} def _warehouse_stock(state: SystemState, warehouse_id: str) -> int: diff --git a/policies/explainability.py b/policies/explainability.py index f10b51f..177b306 100644 --- a/policies/explainability.py +++ b/policies/explainability.py @@ -1,15 +1,48 @@ from __future__ import annotations +from core.scenario_scope import direct_scope_summary, resolve_scenario_scope from core.enums import Mode from core.models import ( Action, CandidatePlanEvaluation, ConstraintViolation as Violation, + Event, KPIState, Plan, + SystemState, ) +def _scenario_scope_text( + selected_actions: list[Action], + *, + state: SystemState | None = None, + event: Event | None = None, +) -> str: + if state is not None and event is not None: + return direct_scope_summary(resolve_scenario_scope(state, event)) + if not selected_actions: + return "" + targets = [] + for action in selected_actions: + if action.target_id not in targets: + targets.append(action.target_id) + if not targets: + return "" + if len(targets) == 1: + return f"The direct scenario response is concentrated on {targets[0]}." + preview = ", ".join(targets[:4]) + return f"The direct scenario response is concentrated on {len(targets)} SKU(s): {preview}." + + +def _scenario_scope_targets(selected_actions: list[Action]) -> list[str]: + targets: list[str] = [] + for action in selected_actions: + if action.target_id not in targets: + targets.append(action.target_id) + return targets + + def _pct(value: float) -> str: return f"{value:.1%}" @@ -30,8 +63,8 @@ def _action_label(action: Action) -> str: def _mode_label(mode: Mode | str) -> str: value = mode.value if isinstance(mode, Mode) else mode return "crisis mode" if value == "crisis" else "normal mode" - - + + def build_plan_summary( before_kpis: KPIState, after_kpis: KPIState, @@ -45,6 +78,8 @@ def build_plan_summary( projection_summary: str = "", simulation_horizon_days: int | None = None, worst_case_kpis: KPIState | None = None, + state: SystemState | None = None, + event: Event | None = None, ) -> str: selected_actions = selected_actions or [] action_text = ( @@ -68,20 +103,22 @@ def build_plan_summary( worst_case_text = "" if worst_case_kpis is not None: worst_case_text = ( - f"Worst projected service reaches {_pct(worst_case_kpis.service_level)} and " + f"Network-wide worst projected service reaches {_pct(worst_case_kpis.service_level)} and " f"worst disruption risk reaches {_pct(worst_case_kpis.disruption_risk)}." ) projection_text = f" {projection_summary}" if projection_summary else "" + scope_text = _scenario_scope_text(selected_actions, state=state, event=event) return ( f"Selected {strategy_label or 'current'} strategy using {action_text}. " + f"{scope_text} " f"Projected service level moves from {_pct(before_kpis.service_level)} to {_pct(after_kpis.service_level)}, " f"total cost changes by {_delta_num(before_kpis.total_cost, after_kpis.total_cost)}, " f"disruption risk changes by {_delta_pct(before_kpis.disruption_risk, after_kpis.disruption_risk)}, " f"and recovery speed changes by {_delta_pct(before_kpis.recovery_speed, after_kpis.recovery_speed)}." f"{comparison}{horizon_text}{worst_case_text}{projection_text} {mode_text}{rationale_text}" ).strip() - - + + def build_winning_factors( selected_actions: list[Action], before_kpis: KPIState, @@ -123,38 +160,45 @@ def build_winning_factors( ] ) return winning_factors - - -def explain_rejected_actions( - candidate_actions: list[Action], - selected_actions: list[Action], - action_limit: int, - infeasible_reasons: dict[str, list[Violation]] | None = None, -) -> list[dict[str, str]]: - infeasible_reasons = infeasible_reasons or {} - selected_by_key = { - (action.action_type.value, action.target_id): action for action in selected_actions - } - selected_ids = {action.action_id for action in selected_actions} - selected_priorities = [action.priority for action in selected_actions] or [0.0] - selected_service = [action.estimated_service_delta for action in selected_actions] or [0.0] - selected_risk = [action.estimated_risk_delta for action in selected_actions] or [0.0] - threshold_priority = min(selected_priorities) - avg_service = sum(selected_service) / len(selected_service) - avg_risk = sum(selected_risk) / len(selected_risk) - - rejected_actions: list[dict[str, str]] = [] - for action in candidate_actions: - if action.action_id in selected_ids: - continue - if action.action_id in infeasible_reasons: - reason = "Hard constraint violation: " + "; ".join(v.message for v in infeasible_reasons[action.action_id]) - else: - duplicate = selected_by_key.get((action.action_type.value, action.target_id)) - if duplicate is not None: - reason = ( - f"superseded by {duplicate.action_id}, which addressed the same target with a stronger priority profile" - ) + + +def explain_rejected_actions( + candidate_actions: list[Action], + selected_actions: list[Action], + action_limit: int, + infeasible_reasons: dict[str, list[Violation]] | None = None, +) -> list[dict[str, str]]: + infeasible_reasons = infeasible_reasons or {} + selected_by_key = { + (action.action_type.value, action.target_id): action + for action in selected_actions + } + selected_ids = {action.action_id for action in selected_actions} + selected_priorities = [action.priority for action in selected_actions] or [0.0] + selected_service = [ + action.estimated_service_delta for action in selected_actions + ] or [0.0] + selected_risk = [action.estimated_risk_delta for action in selected_actions] or [ + 0.0 + ] + threshold_priority = min(selected_priorities) + avg_service = sum(selected_service) / len(selected_service) + avg_risk = sum(selected_risk) / len(selected_risk) + + rejected_actions: list[dict[str, str]] = [] + for action in candidate_actions: + if action.action_id in selected_ids: + continue + if action.action_id in infeasible_reasons: + reason = "Hard constraint violation: " + "; ".join( + v.message for v in infeasible_reasons[action.action_id] + ) + else: + duplicate = selected_by_key.get( + (action.action_type.value, action.target_id) + ) + if duplicate is not None: + reason = f"superseded by {duplicate.action_id}, which addressed the same target with a stronger priority profile" elif action.priority < threshold_priority: reason = ( f"lower operational priority ({action.priority:.2f}) than the chosen set threshold " @@ -169,8 +213,8 @@ def explain_rejected_actions( reason = ( f"lower projected service delta ({action.estimated_service_delta:+.2f}) than the " f"chosen set average ({avg_service:+.2f})" - ) - else: + ) + else: reason = "weaker overall trade-off than the chosen actions on priority and projected impact" rejected_actions.append({"action_id": action.action_id, "reason": reason}) return rejected_actions @@ -179,9 +223,16 @@ def explain_rejected_actions( def build_critic_review( selected_plan: Plan, evaluations: list[CandidatePlanEvaluation], + *, + state: SystemState | None = None, + event: Event | None = None, ) -> tuple[str, list[str]]: selected = next( - (item for item in evaluations if item.strategy_label == selected_plan.strategy_label), + ( + item + for item in evaluations + if item.strategy_label == selected_plan.strategy_label + ), evaluations[0] if evaluations else None, ) runner_up = None @@ -196,11 +247,22 @@ def build_critic_review( runner_up = item break - action_labels = ", ".join(_action_label(action) for action in selected_plan.actions[:3]) or "no-op coverage" + action_labels = ( + ", ".join(_action_label(action) for action in selected_plan.actions[:3]) + or "no-op coverage" + ) + scope_text = _scenario_scope_text(selected_plan.actions, state=state, event=event) + scope_targets = ( + resolve_scenario_scope(state, event).affected_skus + if state is not None and event is not None + else _scenario_scope_targets(selected_plan.actions) + ) summary = ( f"Reviewer assessment: the {selected_plan.strategy_label or 'selected'} plan is reasonable because it uses " f"{action_labels} to support the current operating objective." ) + if scope_text: + summary += f" {scope_text}" if selected is not None and runner_up is not None: summary += ( f" It scored ahead of {runner_up.strategy_label} by {selected.score - runner_up.score:+.4f}, " @@ -212,6 +274,12 @@ def build_critic_review( summary += " Human approval is prudent because the expected impact crosses the approval guardrail." findings: list[str] = [] + if scope_targets: + findings.append( + "Direct scenario monitoring should stay focused on " + + ", ".join(scope_targets[:4]) + + "." + ) longest_recovery = max( (action.estimated_recovery_hours for action in selected_plan.actions), default=0.0, @@ -221,9 +289,17 @@ def build_critic_review( f"The plan depends on at least one action with a recovery window of about {longest_recovery:.0f} hours, so benefits may arrive later than the first operating cycle." ) if runner_up is not None and selected is not None: - cost_gap = selected.projected_kpis.total_cost - runner_up.projected_kpis.total_cost - service_gap = selected.projected_kpis.service_level - runner_up.projected_kpis.service_level - risk_gap = selected.projected_kpis.disruption_risk - runner_up.projected_kpis.disruption_risk + cost_gap = ( + selected.projected_kpis.total_cost - runner_up.projected_kpis.total_cost + ) + service_gap = ( + selected.projected_kpis.service_level + - runner_up.projected_kpis.service_level + ) + risk_gap = ( + selected.projected_kpis.disruption_risk + - runner_up.projected_kpis.disruption_risk + ) if cost_gap > 0: findings.append( f"Compared with {runner_up.strategy_label}, this plan carries {cost_gap:,.0f} more projected cost. Monitor whether the added spend is justified by the operating benefit." @@ -238,7 +314,7 @@ def build_critic_review( ) if selected is not None and selected.projected_state_summary is not None: findings.append( - "Projected end-state: " + "Network-wide projected end-state: " f"{selected.projected_state_summary.summary}" ) if selected_plan.approval_required and not findings: diff --git a/simulation/domain_rules.py b/simulation/domain_rules.py index fe45433..874e0cf 100644 --- a/simulation/domain_rules.py +++ b/simulation/domain_rules.py @@ -2,6 +2,12 @@ from dataclasses import dataclass, field +from core.scenario_scope import ( + payload_demand_changes, + payload_route_ids, + payload_supplier_ids, + resolve_scenario_scope, +) from core.enums import ActionType, EventType from core.models import Action, Event, InventoryItem, SystemState from core.state import recompute_kpis @@ -22,9 +28,9 @@ class SimulationContext: def _supplier_for_item(state: SystemState, item: InventoryItem): - return state.suppliers.get(f"{item.preferred_supplier_id}_{item.sku}") or state.suppliers.get( - item.preferred_supplier_id - ) + return state.suppliers.get( + f"{item.preferred_supplier_id}_{item.sku}" + ) or state.suppliers.get(item.preferred_supplier_id) def _route_for_item(state: SystemState, item: InventoryItem): @@ -35,6 +41,29 @@ def _event_type(event: Event | None) -> EventType | None: return event.type if event is not None else None +def _affected_skus(event: Event | None) -> set[str]: + if event is None: + return set() + return { + str(item["sku"]) + for item in payload_demand_changes(event) + if str(item.get("sku") or "").strip() + } | { + str(sku) + for sku in event.payload.get("affected_skus", []) + if str(sku).strip() + } | ({str(event.payload.get("sku"))} if event.payload.get("sku") else set()) + + +def _demand_change_map(event: Event | None) -> dict[str, float]: + if event is None or event.type not in {EventType.DEMAND_SPIKE, EventType.COMPOUND}: + return {} + return { + str(item["sku"]): float(item.get("multiplier", 1.0)) + for item in payload_demand_changes(event) + } + + def _event_penalty_days( *, event: Event | None, @@ -43,36 +72,40 @@ def _event_penalty_days( ) -> int: if event is None: return 0 - severity_penalty = 2 if event.severity >= 0.85 else 1 if event.severity >= 0.55 else 0 + severity_penalty = ( + 2 if event.severity >= 0.85 else 1 if event.severity >= 0.55 else 0 + ) if event.type == EventType.SUPPLIER_DELAY: - affected_supplier = str(event.payload.get("supplier_id") or "") - if supplier_id and supplier_id == affected_supplier: + affected_suppliers = set(payload_supplier_ids(event)) + if supplier_id and supplier_id in affected_suppliers: return max(1, severity_penalty) if event.type == EventType.ROUTE_BLOCKAGE: - affected_route = str(event.payload.get("route_id") or "") - if route_id and route_id == affected_route: + affected_routes = set(payload_route_ids(event)) + if route_id and route_id in affected_routes: return max(1, severity_penalty) if event.type == EventType.COMPOUND: - supplier_match = supplier_id and supplier_id in { - str(value) for value in event.payload.get("supplier_ids", []) - } - route_match = route_id and route_id in { - str(value) for value in event.payload.get("route_ids", []) - } + supplier_match = supplier_id and supplier_id in set(payload_supplier_ids(event)) + route_match = route_id and route_id in set(payload_route_ids(event)) if supplier_match or route_match: return max(1, severity_penalty) return 0 -def _route_delay_days(state: SystemState, item: InventoryItem, event: Event | None) -> int: +def _route_delay_days( + state: SystemState, item: InventoryItem, event: Event | None +) -> int: route = _route_for_item(state, item) if route is None: return 0 blocked_penalty = 2 if route.status == "blocked" else 0 risk_penalty = 1 if route.risk_score >= 0.65 else 0 - return blocked_penalty + risk_penalty + _event_penalty_days( - event=event, - route_id=route.route_id, + return ( + blocked_penalty + + risk_penalty + + _event_penalty_days( + event=event, + route_id=route.route_id, + ) ) @@ -141,9 +174,9 @@ def apply_candidate_actions( if item is None: continue supplier_id = str(action.parameters.get("supplier_id") or "") - supplier = state.suppliers.get(f"{supplier_id}_{item.sku}") or state.suppliers.get( - supplier_id - ) + supplier = state.suppliers.get( + f"{supplier_id}_{item.sku}" + ) or state.suppliers.get(supplier_id) if supplier is None: continue item.preferred_supplier_id = supplier_id @@ -162,7 +195,9 @@ def apply_candidate_actions( if quantity <= 0: continue supplier = _supplier_for_item(state, item) - capacity = max(int(supplier.capacity), 0) if supplier is not None else quantity + capacity = ( + max(int(supplier.capacity), 0) if supplier is not None else quantity + ) scheduled_qty = min(quantity, capacity) if capacity > 0 else quantity arrival_step = _arrival_step_for_item(state, item, event=event) context.inbound_schedule.setdefault(arrival_step, {}) @@ -187,34 +222,44 @@ def apply_candidate_actions( state.kpis = recompute_kpis(state) -def advance_demand(context: SimulationContext, *, step_index: int, event: Event | None) -> None: +def advance_demand( + context: SimulationContext, *, step_index: int, event: Event | None +) -> None: state = context.working_state event_type = _event_type(event) - spike_multiplier = float(event.payload.get("multiplier", 1.0)) if event is not None else 1.0 - spike_sku = str(event.payload.get("sku") or "") if event is not None else "" - if event_type != EventType.DEMAND_SPIKE or not spike_sku: + demand_changes = _demand_change_map(event) + if event_type not in {EventType.DEMAND_SPIKE, EventType.COMPOUND} or not demand_changes: for sku, item in state.inventory.items(): - item.forecast_qty = context.baseline_forecast_by_sku.get(sku, item.forecast_qty) + item.forecast_qty = context.baseline_forecast_by_sku.get( + sku, item.forecast_qty + ) return decay_ratio = 0.55 ** max(step_index - 1, 0) - active_multiplier = 1.0 + max(spike_multiplier - 1.0, 0.0) * decay_ratio for sku, item in state.inventory.items(): base = context.baseline_forecast_by_sku.get(sku, item.forecast_qty) - if sku == spike_sku: + if sku in demand_changes: + spike_multiplier = demand_changes[sku] + active_multiplier = 1.0 + max(spike_multiplier - 1.0, 0.0) * decay_ratio item.forecast_qty = max(int(round(base * active_multiplier)), 0) else: item.forecast_qty = base -def advance_supplier(context: SimulationContext, *, step_index: int, event: Event | None) -> list[str]: +def advance_supplier( + context: SimulationContext, *, step_index: int, event: Event | None +) -> list[str]: state = context.working_state observations: list[str] = [] if event is None: return observations if event.type not in {EventType.SUPPLIER_DELAY, EventType.COMPOUND}: return observations + scope = resolve_scenario_scope(state, event) + affected_skus = set(scope.supplier_affected_skus or scope.affected_skus) for item in state.inventory.values(): + if affected_skus and item.sku not in affected_skus: + continue supplier = _supplier_for_item(state, item) if supplier is None: continue @@ -225,19 +270,28 @@ def advance_supplier(context: SimulationContext, *, step_index: int, event: Even return observations -def advance_logistics(context: SimulationContext, *, step_index: int, event: Event | None) -> list[str]: +def advance_logistics( + context: SimulationContext, *, step_index: int, event: Event | None +) -> list[str]: state = context.working_state observations: list[str] = [] if event is None: return observations if event.type not in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: return observations + scope = resolve_scenario_scope(state, event) + affected_skus = set(scope.route_affected_skus or scope.affected_skus) + blocked_routes = set(scope.route_ids) for item in state.inventory.values(): + if affected_skus and item.sku not in affected_skus: + continue route = _route_for_item(state, item) if route is None: continue - if route.status == "blocked": - observations.append(f"{item.sku} remains exposed to blocked route {route.route_id}") + if route.route_id in blocked_routes or route.status == "blocked": + observations.append( + f"{item.sku} remains exposed to blocked route {route.route_id}" + ) elif route.risk_score >= 0.6 and step_index == 1: observations.append( f"{item.sku} still uses a high-risk lane ({route.route_id}, risk {route.risk_score:.0%})" @@ -298,7 +352,9 @@ def advance_risk( unresolved_pressure = 0.0 if state.inventory: - unresolved_pressure = step_metrics["inventory_out_of_stock"] / max(len(state.inventory), 1) + unresolved_pressure = step_metrics["inventory_out_of_stock"] / max( + len(state.inventory), 1 + ) improvement = min(context.mitigation_count * 0.03, 0.15) decay = 0.08 + improvement escalation = 0.05 if unresolved_pressure >= 0.08 else 0.0 @@ -331,19 +387,37 @@ def _projected_recovery_speed( return max(0.05, min(0.99, base + improvement - penalty)) -def dominant_constraint(context: SimulationContext, step_metrics: dict[str, int]) -> str: +def dominant_constraint( + context: SimulationContext, step_metrics: dict[str, int] +) -> str: state = context.working_state + active_event = state.active_events[-1] if state.active_events else None + scope = resolve_scenario_scope(state, active_event) + blocked_routes = set(scope.route_ids) + delayed_suppliers = set(scope.supplier_ids) if step_metrics["inventory_out_of_stock"] > 0: return "inventory shortfall" if step_metrics["backlog_units"] > 0: return "order backlog" if any( - (_route_for_item(state, item) is not None and _route_for_item(state, item).status == "blocked") + ( + _route_for_item(state, item) is not None + and ( + _route_for_item(state, item).status == "blocked" + or _route_for_item(state, item).route_id in blocked_routes + ) + ) for item in state.inventory.values() ): return "route disruption" if any( - (_supplier_for_item(state, item) is not None and _supplier_for_item(state, item).reliability < 0.85) + ( + _supplier_for_item(state, item) is not None + and ( + _supplier_for_item(state, item).reliability < 0.85 + or _supplier_for_item(state, item).supplier_id in delayed_suppliers + ) + ) for item in state.inventory.values() ): return "supplier reliability" @@ -362,7 +436,9 @@ def step_summary( f"risk at {context.working_state.kpis.disruption_risk:.0%}", ] if step_metrics["inventory_out_of_stock"] > 0: - parts.append(f"{step_metrics['inventory_out_of_stock']} SKU(s) remain out of stock") + parts.append( + f"{step_metrics['inventory_out_of_stock']} SKU(s) remain out of stock" + ) elif step_metrics["inventory_at_risk"] > 0: parts.append(f"{step_metrics['inventory_at_risk']} SKU(s) remain at risk") if step_metrics["inbound_units_due"] > 0: diff --git a/simulation/evaluator.py b/simulation/evaluator.py index 3feb321..64f752b 100644 --- a/simulation/evaluator.py +++ b/simulation/evaluator.py @@ -2,6 +2,7 @@ from dataclasses import dataclass +from core.scenario_scope import direct_scope_summary, resolve_scenario_scope from core.enums import EventType from core.models import ( Action, @@ -27,6 +28,12 @@ ) +def _event_scope_label(state: SystemState, event: Event | None) -> str: + if event is None: + return "" + return direct_scope_summary(resolve_scenario_scope(state, event)) + + @dataclass class CandidateProjection: projected_state: SystemState @@ -47,6 +54,7 @@ def evaluate_candidate_plan( ) -> CandidateProjection: projected_state = clone_state(state) projected_event = event.model_copy(deep=True) if event is not None else None + baseline_scope_summary = _event_scope_label(state, event) if projected_event is not None and projected_state.active_events: projected_state.active_events[-1] = projected_event context = initialize_context(projected_state, projected_event) @@ -66,11 +74,17 @@ def evaluate_candidate_plan( for step_index in range(1, max(horizon_days, 1) + 1): advance_demand(context, step_index=step_index, event=projected_event) - supplier_notes = advance_supplier(context, step_index=step_index, event=projected_event) - route_notes = advance_logistics(context, step_index=step_index, event=projected_event) + supplier_notes = advance_supplier( + context, step_index=step_index, event=projected_event + ) + route_notes = advance_logistics( + context, step_index=step_index, event=projected_event + ) step_metrics = advance_inventory(context, step_index=step_index) final_metrics = step_metrics - severity = advance_risk(context, event=projected_event, step_metrics=step_metrics) + severity = advance_risk( + context, event=projected_event, step_metrics=step_metrics + ) step_kpis = projected_state.kpis.model_copy(deep=True) service_values.append(step_kpis.service_level) cost_values.append(step_kpis.total_cost) @@ -78,10 +92,16 @@ def evaluate_candidate_plan( stockout_values.append(step_kpis.stockout_risk) key_changes = supplier_notes[:2] + route_notes[:2] + if step_index == 1 and baseline_scope_summary: + key_changes.insert(0, baseline_scope_summary) if step_metrics["inbound_units_due"] > 0: - key_changes.append(f"{step_metrics['inbound_units_due']} units landed into the network") + key_changes.append( + f"{step_metrics['inbound_units_due']} units landed into the network" + ) if step_metrics["backlog_units"] > 0: - key_changes.append(f"backlog stands at {step_metrics['backlog_units']} units") + key_changes.append( + f"backlog stands at {step_metrics['backlog_units']} units" + ) projection_steps.append( ProjectionStep( @@ -118,6 +138,7 @@ def evaluate_candidate_plan( projected_event=projected_event, ) projection_summary = _projection_summary( + state=state, projected_steps=projection_steps, projected_kpis=projected_kpis, worst_case_kpis=worst_case_kpis, @@ -168,6 +189,7 @@ def _projected_state_summary( def _projection_summary( *, + state: SystemState, projected_steps: list[ProjectionStep], projected_kpis: KPIState, worst_case_kpis: KPIState, @@ -188,14 +210,20 @@ def _projection_summary( if event is not None: if event.type == EventType.DEMAND_SPIKE: event_text = " Demand pressure is expected to normalize gradually." - elif event.type in {EventType.SUPPLIER_DELAY, EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: + elif event.type in { + EventType.SUPPLIER_DELAY, + EventType.ROUTE_BLOCKAGE, + EventType.COMPOUND, + }: event_text = " Mitigation quality determines whether disruption severity decays fast enough." + scope_text = "" + if event is not None: + scope_text = _event_scope_label(state, event) return ( - f"Over the next {len(projected_steps)} day(s), service bottoms at " + f"{scope_text} Over the next {len(projected_steps)} day(s), projected network-wide service bottoms at " f"{worst_case_kpis.service_level:.0%} on {worst_service_step.label} while disruption risk peaks at " f"{worst_case_kpis.disruption_risk:.0%} on {highest_risk_step.label}. " f"By {projected_steps[-1].label}, the plan projects service at {projected_kpis.service_level:.0%}, " f"risk at {projected_kpis.disruption_risk:.0%}, and recovery speed at {projected_kpis.recovery_speed:.0%}. " f"{summary.summary}{event_text}" ) - diff --git a/simulation/learning.py b/simulation/learning.py index 9b76449..85964e6 100644 --- a/simulation/learning.py +++ b/simulation/learning.py @@ -1,9 +1,10 @@ -from __future__ import annotations - -from uuid import uuid4 - -from core.enums import ApprovalStatus, EventType -from core.models import DecisionLog, MemorySnapshot, ReflectionNote, ScenarioRun, SystemState +from __future__ import annotations + +from uuid import uuid4 + +from core.scenario_scope import payload_route_ids, payload_supplier_ids +from core.enums import ApprovalStatus, EventType +from core.models import DecisionLog, MemorySnapshot, ReflectionNote, ScenarioRun, SystemState from core.state import default_memory, utc_now from llm.service import generate_reflection_note @@ -52,21 +53,21 @@ def _update_route_learning(state: SystemState, route_id: str, severity: float) - return updated -def _update_numeric_learning(state: SystemState, run: ScenarioRun) -> tuple[dict[str, float], dict[str, float]]: - supplier_updates: dict[str, float] = {} - route_updates: dict[str, float] = {} - for event in run.events: - supplier_id = event.payload.get("supplier_id") - route_id = event.payload.get("route_id") - if supplier_id and event.type in {EventType.SUPPLIER_DELAY, EventType.COMPOUND}: - updated_supplier = _update_supplier_learning(state, supplier_id, event.severity) - if updated_supplier is not None: - supplier_updates[supplier_id] = updated_supplier - if route_id and event.type in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: - updated_route = _update_route_learning(state, route_id, event.severity) - if updated_route is not None: - route_updates[route_id] = updated_route - return supplier_updates, route_updates +def _update_numeric_learning(state: SystemState, run: ScenarioRun) -> tuple[dict[str, float], dict[str, float]]: + supplier_updates: dict[str, float] = {} + route_updates: dict[str, float] = {} + for event in run.events: + if event.type in {EventType.SUPPLIER_DELAY, EventType.COMPOUND}: + for supplier_id in payload_supplier_ids(event): + updated_supplier = _update_supplier_learning(state, supplier_id, event.severity) + if updated_supplier is not None: + supplier_updates[supplier_id] = updated_supplier + if event.type in {EventType.ROUTE_BLOCKAGE, EventType.COMPOUND}: + for route_id in payload_route_ids(event): + updated_route = _update_route_learning(state, route_id, event.severity) + if updated_route is not None: + route_updates[route_id] = updated_route + return supplier_updates, route_updates def _build_outcome_summary( diff --git a/simulation/scenarios.py b/simulation/scenarios.py index 34bce34..7cf97e9 100644 --- a/simulation/scenarios.py +++ b/simulation/scenarios.py @@ -37,8 +37,12 @@ def get_scenario_events(name: str) -> list[Event]: event_id="evt_supplier_delay", event_type=EventType.SUPPLIER_DELAY, severity=0.80, - payload={"supplier_id": "SUP_BN", "sku": "SKU_001", "delay_hours": 48}, - entity_ids=["SUP_BN", "SKU_001"], + payload={ + "supplier_id": "SUP_BN", + "affected_skus": ["SKU_001", "SKU_004", "SKU_007", "SKU_010"], + "delay_hours": 48, + }, + entity_ids=["SUP_BN", "SKU_001", "SKU_004", "SKU_007", "SKU_010"], ) ], "demand_spike": [ @@ -46,40 +50,45 @@ def get_scenario_events(name: str) -> list[Event]: event_id="evt_demand_spike", event_type=EventType.DEMAND_SPIKE, severity=0.70, - payload={"sku": "SKU_024", "multiplier": 2.2}, - entity_ids=["SKU_024"], - ) - ], - "route_blockage": [ - build_event( - event_id="evt_route_blockage", - event_type=EventType.ROUTE_BLOCKAGE, - severity=0.78, - payload={"route_id": "R_BN_HN_MAIN", "reason": "flooding"}, - entity_ids=["R_BN_HN_MAIN"], - ) - ], - "compound_disruption": [ - build_event( - event_id="evt_compound_delay", - event_type=EventType.COMPOUND, - severity=0.92, payload={ - "supplier_id": "SUP_BN", - "route_id": "R_BN_HN_MAIN", - "sku": "SKU_001", + "demand_changes": [ + {"sku": "SKU_024", "multiplier": 2.2}, + {"sku": "SKU_013", "multiplier": 1.6}, + {"sku": "SKU_036", "multiplier": 1.4}, + ] }, - entity_ids=["SUP_BN", "R_BN_HN_MAIN", "SKU_001"], - ), - build_event( - event_id="evt_compound_spike", - event_type=EventType.DEMAND_SPIKE, - severity=0.75, - payload={"sku": "SKU_024", "multiplier": 1.9}, - entity_ids=["SKU_024"], - ), + entity_ids=["SKU_024", "SKU_013", "SKU_036"], + ) ], - } + "route_blockage": [ + build_event( + event_id="evt_route_blockage", + event_type=EventType.ROUTE_BLOCKAGE, + severity=0.78, + payload={ + "route_ids": ["R_BN_HN_MAIN"], + "reason": "flooding", + }, + entity_ids=["R_BN_HN_MAIN"], + ) + ], + "compound_disruption": [ + build_event( + event_id="evt_compound_delay", + event_type=EventType.COMPOUND, + severity=0.92, + payload={ + "supplier_ids": ["SUP_BN"], + "route_ids": ["R_BN_HN_MAIN"], + "demand_changes": [ + {"sku": "SKU_024", "multiplier": 1.9}, + ], + "reason": "supplier delay plus lane disruption during demand pressure", + }, + entity_ids=["SUP_BN", "R_BN_HN_MAIN", "SKU_024"], + ) + ], + } return mapping[name] diff --git a/tests/test_agents.py b/tests/test_agents.py index 0911262..508ba13 100644 --- a/tests/test_agents.py +++ b/tests/test_agents.py @@ -35,3 +35,34 @@ def test_agents_emit_actions_for_supplier_delay() -> None: assert "reorder point" in inventory.domain_summary assert supplier.proposals assert isinstance(logistics.proposals, list) + + +def test_route_blockage_actions_stay_within_blocked_lane_scope() -> None: + state = load_initial_state() + event = Event( + event_id="evt_route_scope_test", + type=EventType.ROUTE_BLOCKAGE, + source="test", + severity=0.78, + entity_ids=["R_BN_HN_MAIN"], + occurred_at=utc_now(), + detected_at=utc_now(), + payload={"route_ids": ["R_BN_HN_MAIN"], "reason": "flooding"}, + dedupe_key="evt_route_scope_test", + ) + + logistics = LogisticsAgent().run(state, event) + inventory = InventoryAgent().run(state, event) + + logistics_targets = {action.target_id for action in logistics.proposals} + inventory_targets = {action.target_id for action in inventory.proposals} + + assert logistics_targets + assert all( + state.inventory[sku].preferred_route_id == "R_BN_HN_MAIN" + for sku in logistics_targets + ) + assert all( + state.inventory[sku].preferred_route_id == "R_BN_HN_MAIN" + for sku in inventory_targets + ) diff --git a/tests/test_seed_data_alignment.py b/tests/test_seed_data_alignment.py index f8f5e37..39e6815 100644 --- a/tests/test_seed_data_alignment.py +++ b/tests/test_seed_data_alignment.py @@ -1,6 +1,7 @@ from agents.demand import DemandAgent from agents.inventory import InventoryAgent from app_api.services import inventory_rows +from core.scenario_scope import resolve_scenario_scope from core.state import load_initial_state from orchestrator.graph import build_graph from simulation.scenarios import get_scenario_events @@ -30,7 +31,7 @@ def test_seed_scenarios_target_existing_entities() -> None: state = load_initial_state() inventory_skus = set(state.inventory) route_ids = set(state.routes) - supplier_ids = {record.supplier_id for record in state.suppliers.values()} + supplier_catalog_ids = {record.supplier_id for record in state.suppliers.values()} for scenario_name in [ "supplier_delay", @@ -39,16 +40,47 @@ def test_seed_scenarios_target_existing_entities() -> None: "compound_disruption", ]: for event in get_scenario_events(scenario_name): + payload_supplier_ids = event.payload.get("supplier_ids") or [] + route_ids_payload = event.payload.get("route_ids") or [] supplier_id = event.payload.get("supplier_id") sku = event.payload.get("sku") + affected_skus = event.payload.get("affected_skus") or [] + demand_changes = event.payload.get("demand_changes") or [] route_id = event.payload.get("route_id") if supplier_id is not None: - assert supplier_id in supplier_ids + assert supplier_id in supplier_catalog_ids + for supplier_item in payload_supplier_ids: + assert supplier_item in supplier_catalog_ids if sku is not None: assert sku in inventory_skus + for affected_sku in affected_skus: + assert affected_sku in inventory_skus + for change in demand_changes: + assert change["sku"] in inventory_skus if route_id is not None: assert route_id in route_ids + for route_item in route_ids_payload: + assert route_item in route_ids + + +def test_route_and_compound_scenarios_resolve_multi_entity_scope() -> None: + state = load_initial_state() + + route_event = get_scenario_events("route_blockage")[0] + route_scope = resolve_scenario_scope(state, route_event) + assert route_scope.route_ids == ["R_BN_HN_MAIN"] + assert "SKU_001" in route_scope.route_affected_skus + assert route_scope.warehouse_ids == ["WH_HN"] + assert set(route_scope.node_ids) == {"Bac Ninh", "Hanoi"} + + compound_event = get_scenario_events("compound_disruption")[0] + compound_scope = resolve_scenario_scope(state, compound_event) + assert compound_scope.route_ids == ["R_BN_HN_MAIN"] + assert compound_scope.supplier_ids == ["SUP_BN"] + assert compound_scope.demand_affected_skus == ["SKU_024"] + assert "SKU_001" in compound_scope.supplier_affected_skus + assert "SKU_001" in compound_scope.route_affected_skus def test_seed_inventory_baseline_has_managed_low_stock_pressure() -> None: