feat(services): XOpt Optimizer Service infrastructure

src/osprey/approval/__init__.py

@@ -56,7 +56,9 @@
     create_code_approval_interrupt,
     create_memory_approval_interrupt,
     create_plan_approval_interrupt,
+    create_question_interrupt,
     create_step_approval_interrupt,
+    create_xopt_approval_interrupt,
     get_approval_resume_data,
     get_approved_payload_from_state,
     handle_service_with_interrupts,
@@ -77,6 +79,8 @@
     "create_code_approval_interrupt",
     "create_memory_approval_interrupt",
     "create_channel_write_approval_interrupt",
+    "create_question_interrupt",
+    "create_xopt_approval_interrupt",
     "get_approved_payload_from_state",
     "get_approval_resume_data",
     "clear_approval_state",

src/osprey/approval/approval_system.py

@@ -525,6 +525,138 @@ def create_code_approval_interrupt(
     }
 
 
+def create_question_interrupt(
+    question: str,
+    options: list[str] | None = None,
+) -> dict[str, Any]:
+    """Create a question interrupt payload for non-approval user interactions.
+
+    Unlike approval interrupts (which expect yes/no), question interrupts pass
+    the user's free-text answer back via ``Command(resume=answer)``.  The
+    gateway detects the ``"type": "question"`` key and skips approval
+    classification.
+
+    :param question: The question to present to the user
+    :type question: str
+    :param options: Optional list of valid option strings for display
+    :type options: list[str] | None
+    :return: Interrupt payload dict with type, user_message, and options
+    :rtype: dict[str, Any]
+    """
+    return {
+        "type": "question",
+        "user_message": question,
+        "options": options or [],
+    }
+
+
+def create_xopt_approval_interrupt(
+    optimization_config: dict[str, Any],
+    strategy: str,
+    objective: str,
+    machine_state_details: dict[str, Any] | None = None,
+    step_objective: str = "Execute XOpt optimization",
+) -> dict[str, Any]:
+    """Create structured interrupt data for XOpt optimization approval.
+
+    Generates LangGraph-compatible interrupt data for XOpt configurations that
+    require human approval before execution. The interrupt provides comprehensive
+    context including the optimization config, strategy, and machine state details.
+
+    :param optimization_config: Generated optimization config dict
+    :type optimization_config: Dict[str, Any]
+    :param strategy: Selected optimization strategy (exploration/optimization)
+    :type strategy: str
+    :param objective: Optimization objective description
+    :type objective: str
+    :param machine_state_details: Optional machine state assessment details
+    :type machine_state_details: Dict[str, Any], optional
+    :param step_objective: High-level objective for user context
+    :type step_objective: str
+    :return: Dictionary containing user_message and resume_payload for LangGraph
+    :rtype: Dict[str, Any]
+
+    Examples:
+        Basic XOpt approval::
+
+            >>> interrupt_data = create_xopt_approval_interrupt(
+            ...     optimization_config={"algorithm": "upper_confidence_bound"},
+            ...     strategy="exploration",
+            ...     objective="Maximize injection efficiency",
+            ...     step_objective="Execute XOpt optimization"
+            ... )
+            >>> 'yes' in interrupt_data['user_message']
+            True
+
+    .. warning::
+       This function is used for security-critical approval decisions for
+       optimization operations that may affect machine parameters.
+    """
+    import yaml as _yaml
+
+    # Format config as YAML for human readability
+    config_display = _yaml.dump(optimization_config, default_flow_style=False, sort_keys=False)
+
+    # Format machine state if available
+    machine_state_section = ""
+    if machine_state_details:
+        machine_state_section = f"""
+**Machine State Assessment:**
+```
+{_format_machine_state(machine_state_details)}
+```
+"""
+
+    user_message = f"""
+⚠️ **HUMAN APPROVAL REQUIRED** ⚠️
+
+**Task:** {step_objective}
+**Optimization Objective:** {objective}
+**Strategy:** {strategy.upper()}
+{machine_state_section}
+**Optimization Configuration:**
+```yaml
+{config_display}```
+
+**Review the configuration above carefully.**
+
+**To proceed, respond with:**
+- **`yes`** to approve and execute the optimization
+- **`no`** to cancel this operation
+""".strip()
+
+    return {
+        "user_message": user_message,
+        "resume_payload": {
+            "approval_type": create_approval_type("xopt_optimizer"),
+            "step_objective": step_objective,
+            "optimization_config": optimization_config,
+            "strategy": strategy,
+            "objective": objective,
+            "machine_state_details": machine_state_details,
+        },
+    }
+
+
+def _format_machine_state(details: dict[str, Any]) -> str:
+    """Format machine state details for display.
+
+    :param details: Machine state details dictionary
+    :type details: Dict[str, Any]
+    :return: Formatted string for display
+    :rtype: str
+    """
+    lines = []
+    for key, value in details.items():
+        if isinstance(value, dict):
+            lines.append(f"{key}:")
+            for k, v in value.items():
+                lines.append(f"  {k}: {v}")
+        else:
+            lines.append(f"{key}: {value}")
+    return "\n".join(lines)
+
+
 # =============================================================================
 # STREAMLINED APPROVAL HELPERS
 # =============================================================================