feat(clmm): orphaned-position lifecycle, bin_count passthrough, Raydium via Gateway

bots/controllers/generic/lp_rebalancer/lp_rebalancer.py

@@ -58,7 +58,8 @@ class LPRebalancerConfig(ControllerConfigBase):
     position_offset_pct: Decimal = Field(
         default=Decimal("0.01"),
         json_schema_extra={"is_updatable": True},
-        description="Offset from current price. Positive = out-of-range (single-sided). Negative = in-range (needs both tokens, autoswap will convert |offset|%)"
+        description="Offset from current price. Positive = out-of-range (single-sided). "
+                    "Negative = in-range (needs both tokens, autoswap will convert |offset|%)"
     )
 
     # Rebalance threshold - used to set LP executor's limit prices
@@ -184,6 +185,9 @@ def __init__(self, config: LPRebalancerConfig, *args, **kwargs):
 
         # Track the executor we created
         self._current_executor_id: Optional[str] = None
+        # Set when a FAILED LP executor still reports a live on-chain position; the
+        # controller halts new position creation until it is recovered manually
+        self._orphaned_position_address: Optional[str] = None
 
         # Track amounts from last closed position (for autoswap sizing)
         self._last_closed_base_amount: Optional[Decimal] = None
@@ -416,6 +420,16 @@ def determine_executor_actions(self) -> List[ExecutorAction]:
 
         actions = []
 
+        # An orphaned on-chain position (FAILED close) must be recovered before any new
+        # position is opened - creating a fresh executor here would stack live exposure
+        # on top of the stranded one.
+        if self._orphaned_position_address:
+            self.logger().debug(
+                f"Halted: position {self._orphaned_position_address} from a FAILED executor is "
+                "still open on-chain and requires manual recovery"
+            )
+            return actions
+
         # Handle order executor tracking and completion (for autoswap)
         if self._pending_swap_side is not None:
             if not self._swap_executor_id:
@@ -509,10 +523,15 @@ def determine_executor_actions(self) -> List[ExecutorAction]:
             # Previous executor terminated - capture final amounts and update position_hold
             terminated_executor = self.get_tracked_executor()
             if terminated_executor:
-                # Skip position_hold update if executor failed (no tokens were actually deposited/returned)
-                if terminated_executor.close_type == CloseType.FAILED:
+                # Skip position_hold update if the executor failed (nothing deposited or
+                # returned) or ended as an involuntary hold (close exhausted): in the
+                # latter case base/quote amounts are pool balances of a still-open
+                # position, and booking them as returned tokens would corrupt the hold.
+                if (terminated_executor.close_type == CloseType.FAILED
+                        or terminated_executor.custom_info.get("hold_reason")):
                     self.logger().warning(
-                        f"Executor {terminated_executor.id} FAILED - skipping position_hold update"
+                        f"Executor {terminated_executor.id} ended {terminated_executor.close_type} "
+                        "without returning tokens - skipping position_hold update"
                     )
                 else:
                     self._last_closed_base_amount = Decimal(str(terminated_executor.custom_info.get("base_amount", 0)))
@@ -539,16 +558,32 @@ def determine_executor_actions(self) -> List[ExecutorAction]:
                         f"Position hold total: base={self._position_hold_base}, quote={self._position_hold_quote}"
                     )
 
-            # Check if executor FAILED - retry with same side from executor's config
+            # Check if the executor went terminal abnormally - FAILED (nothing on-chain)
+            # or an involuntary POSITION_HOLD (close retries exhausted, hold_reason set)
             executor_failed = terminated_executor and terminated_executor.close_type == CloseType.FAILED
+            involuntary_hold = bool(terminated_executor and terminated_executor.custom_info.get("hold_reason"))
             failed_executor_side = None
-            if executor_failed:
+            if executor_failed or involuntary_hold:
                 failed_executor_side = terminated_executor.custom_info.get("side")
+                # A terminal executor that still reports a position address went down on
+                # the CLOSE side: its deposit is still on-chain (involuntary hold, or a
+                # legacy FAILED-with-position from a force-stop). Re-opening would stack
+                # a second position on top of the stranded one.
+                orphaned_position = terminated_executor.custom_info.get("position_address")
+                if orphaned_position:
+                    self._orphaned_position_address = orphaned_position
+                    self._current_executor_id = None
+                    self.logger().error(
+                        f"Executor {terminated_executor.id} ended {terminated_executor.close_type} "
+                        f"with position {orphaned_position} still open on-chain. Halting new "
+                        "position creation until the position is closed or recovered manually."
+                    )
+                    return actions
 
             # Capture closed position bounds for side determination (only for successful closes)
             closed_lower_price = None
             closed_upper_price = None
-            if terminated_executor and not executor_failed:
+            if terminated_executor and not executor_failed and not involuntary_hold:
                 closed_lower_price = Decimal(str(terminated_executor.custom_info.get("lower_price", 0)))
                 closed_upper_price = Decimal(str(terminated_executor.custom_info.get("upper_price", 0)))
 
@@ -576,7 +611,10 @@ def determine_executor_actions(self) -> List[ExecutorAction]:
                 else:
                     # Price is within old bounds (shouldn't happen with limit-price auto-close)
                     side = self._determine_side_from_price(self._pool_price)
-                    self.logger().info(f"Price {self._pool_price} in range [{closed_lower_price}, {closed_upper_price}] → side={side} from limits")
+                    self.logger().info(
+                        f"Price {self._pool_price} in range [{closed_lower_price}, {closed_upper_price}] "
+                        f"→ side={side} from limits"
+                    )
             else:
                 # Fallback to price limits
                 if not self._pool_price:
@@ -941,7 +979,8 @@ def to_format_status(self) -> List[str]:
         width = self.config.position_width_pct
         offset = self.config.position_offset_pct
         threshold = self.config.rebalance_threshold_pct
-        line = f"| Config: side={side_str}, amount={amt} {self._quote_token}, width={width}%, offset={offset}%, threshold={threshold}%"
+        line = (f"| Config: side={side_str}, amount={amt} {self._quote_token}, "
+                f"width={width}%, offset={offset}%, threshold={threshold}%")
         status.append(line + " " * (box_width - len(line) + 1) + "|")
 
         status.append("|" + " " * box_width + "|")
@@ -985,7 +1024,8 @@ def to_format_status(self) -> List[str]:
                 lower_limit = Decimal(str(lower_price)) * (Decimal("1") - threshold_pct)
                 upper_limit = Decimal(str(upper_price)) * (Decimal("1") + threshold_pct)
 
-                line = f"| Price: {float(self._pool_price):.{price_decimals}f}  |  Auto-close if: <{float(lower_limit):.{price_decimals}f} or >{float(upper_limit):.{price_decimals}f}"
+                line = (f"| Price: {float(self._pool_price):.{price_decimals}f}  |  Auto-close if: "
+                        f"<{float(lower_limit):.{price_decimals}f} or >{float(upper_limit):.{price_decimals}f}")
                 status.append(line + " " * (box_width - len(line) + 1) + "|")
 
                 state = custom.get("state", "UNKNOWN")
@@ -1070,7 +1110,8 @@ def to_format_status(self) -> List[str]:
             line = f"| Swaps Executed: {len(closed_swaps)}"
             status.append(line + " " * (box_width - len(line) + 1) + "|")
 
-        line = f"| Fees Collected: {float(total_fees_base):.6f} {self._base_token} + {float(total_fees_quote):.6f} {self._quote_token} = {float(total_fees_value):.6f} {self._quote_token}"
+        line = (f"| Fees Collected: {float(total_fees_base):.6f} {self._base_token} + "
+                f"{float(total_fees_quote):.6f} {self._quote_token} = {float(total_fees_value):.6f} {self._quote_token}")
         status.append(line + " " * (box_width - len(line) + 1) + "|")
 
         status.append("+" + "-" * box_width + "+")

database/repositories/executor_repository.py

@@ -134,6 +134,28 @@ async def get_executors(
         result = await self.session.execute(stmt)
         return list(result.scalars().all())
 
+    async def get_executors_by_close_types(
+            self,
+            close_types: List[str],
+            executor_type: Optional[str] = None,
+            limit: Optional[int] = 500,
+    ) -> List[ExecutorRecord]:
+        """Get executors whose close_type is one of the given values, newest first.
+
+        Filter by executor_type in SQL where possible: applying `limit` to a broad
+        candidate set and filtering in Python can silently drop the very rows the
+        caller is looking for once the newest N candidates are all irrelevant.
+        """
+        stmt = select(ExecutorRecord).where(ExecutorRecord.close_type.in_(close_types))
+        if executor_type:
+            stmt = stmt.where(ExecutorRecord.executor_type == executor_type)
+        stmt = stmt.order_by(desc(ExecutorRecord.created_at))
+        if limit is not None:
+            stmt = stmt.limit(limit)
+
+        result = await self.session.execute(stmt)
+        return list(result.scalars().all())
+
     async def get_active_executors(
             self,
             account_name: Optional[str] = None,

models/__init__.py

@@ -70,6 +70,8 @@
     ExecutorFilterRequest,
     ExecutorResponse,
     ExecutorsSummaryResponse,
+    OrphanedPositionRecord,
+    OrphanedPositionsResponse,
     StopExecutorRequest,
     StopExecutorResponse,
 )
@@ -384,4 +386,6 @@
     "ExecutorResponse",
     "ExecutorDetailResponse",
     "ExecutorsSummaryResponse",
+    "OrphanedPositionRecord",
+    "OrphanedPositionsResponse",
 ]

models/executors.py

@@ -421,8 +421,52 @@ class CreateExecutorResponse(BaseModel):
 class StopExecutorResponse(BaseModel):
     """Response after stopping an executor."""
     executor_id: str = Field(description="Executor identifier")
-    status: str = Field(description="New status (usually 'stopping')")
+    status: str = Field(description="New status: 'stopping', or 'already_terminated' when the stop was a no-op")
     keep_position: bool = Field(description="Whether position was kept open")
+    close_type: Optional[str] = Field(default=None, description="Final close_type when already terminated")
+    position_address: Optional[str] = Field(
+        default=None, description="On-chain position address from the executor's final state, if any"
+    )
+    orphaned_position: bool = Field(
+        default=False,
+        description="True when the executor terminated with a live on-chain position that needs recovery"
+    )
+    hold_reason: Optional[str] = Field(
+        default=None,
+        description="Why a POSITION_HOLD terminal was involuntary (e.g. close_retries_exhausted); None for voluntary holds"
+    )
+
+
+class OrphanedPositionRecord(BaseModel):
+    """A terminated executor that may still own an on-chain position."""
+    executor_id: str = Field(description="Executor identifier")
+    executor_type: str = Field(description="Executor type (e.g. lp_executor)")
+    account_name: Optional[str] = Field(default=None, description="Account name")
+    connector_name: Optional[str] = Field(default=None, description="Connector name")
+    trading_pair: Optional[str] = Field(default=None, description="Trading pair")
+    controller_id: str = Field(default="main", description="Controller/agent grouping label")
+    close_type: Optional[str] = Field(
+        default=None, description="POSITION_HOLD (involuntary hold), FAILED, or SYSTEM_CLEANUP"
+    )
+    closed_at: Optional[str] = Field(default=None, description="Termination timestamp (ISO format)")
+    position_address: Optional[str] = Field(
+        default=None, description="On-chain position address (None for restart cleanups, which never persisted state)"
+    )
+    state: Optional[str] = Field(default=None, description="Executor state at termination (e.g. FAILED, CLOSING)")
+    hold_reason: Optional[str] = Field(
+        default=None,
+        description="Why the hold was involuntary (e.g. close_retries_exhausted); None for legacy FAILED/SYSTEM_CLEANUP records"
+    )
+    needs_onchain_reconciliation: bool = Field(
+        default=False,
+        description="True when the position address is unknown and on-chain state must be checked externally"
+    )
+
+
+class OrphanedPositionsResponse(BaseModel):
+    """Terminated executors that may have stranded on-chain positions."""
+    count: int = Field(description="Number of orphan candidates")
+    orphans: List[OrphanedPositionRecord] = Field(description="Orphan candidate records")
 
 
 class ExecutorsSummaryResponse(BaseModel):

routers/executors.py

@@ -18,6 +18,7 @@
     ExecutorFilterRequest,
     ExecutorLogsResponse,
     ExecutorsSummaryResponse,
+    OrphanedPositionsResponse,
     PerformanceReportResponse,
     PositionHoldResponse,
     PositionsSummaryResponse,
@@ -357,6 +358,57 @@ async def stop_executor(
 # Position Hold Endpoints
 # ========================================
 
+@router.get("/positions/orphaned", response_model=OrphanedPositionsResponse)
+async def get_orphaned_positions(
+    executor_service: ExecutorService = Depends(get_executor_service)
+):
+    """
+    List terminated executors that may still own an on-chain position.
+
+    Covers three orphan classes:
+    - Involuntary holds: close_type POSITION_HOLD with hold_reason set (an LP close
+      that exhausted its retries): the position is live on-chain with no automated
+      owner.
+    - Legacy FAILED records whose final state still reported a position_address
+      (force-stop stragglers, records persisted by older executors).
+    - Executors terminated by SYSTEM_CLEANUP after an API restart: their on-chain
+      state was never persisted, so they need external reconciliation.
+
+    This is a DB-side listing. Before recovering, cross-check candidates against
+    on-chain reality via the gateway positions-owned endpoints
+    (/trading/clmm/positions-owned, /trading/amm/positions-owned).
+    """
+    try:
+        orphans = await executor_service.get_orphaned_positions()
+        return OrphanedPositionsResponse(count=len(orphans), orphans=orphans)
+    except Exception as e:
+        logger.error(f"Error listing orphaned positions: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Error listing orphaned positions: {str(e)}")
+
+
+@router.post("/{executor_id}/resolve-orphan")
+async def resolve_orphaned_position(
+    executor_id: str,
+    executor_service: ExecutorService = Depends(get_executor_service)
+):
+    """
+    Mark an orphaned position as recovered.
+
+    Call after the stranded on-chain position has been closed (or adopted)
+    externally. Removes the executor from /executors/positions/orphaned and from
+    agent-facing orphan warnings. Only valid for terminated executors that are
+    orphan candidates: an involuntary hold (POSITION_HOLD with hold_reason or the
+    orphaned_position flag), FAILED, or SYSTEM_CLEANUP.
+    """
+    try:
+        return await executor_service.resolve_orphaned_position(executor_id)
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error(f"Error resolving orphaned position for {executor_id}: {e}", exc_info=True)
+        raise HTTPException(status_code=500, detail=f"Error resolving orphaned position: {str(e)}")
+
+
 @router.get("/positions/summary", response_model=PositionsSummaryResponse)
 async def get_positions_summary(
     controller_id: Optional[str] = None,