diff --git a/python/sglang/srt/disaggregation/base/conn.py b/python/sglang/srt/disaggregation/base/conn.py
index a7bf9904a2..a273fc0bbf 100644
--- a/python/sglang/srt/disaggregation/base/conn.py
+++ b/python/sglang/srt/disaggregation/base/conn.py
@@ -95,7 +95,6 @@ class BaseKVManager(ABC):


 class BaseKVSender(ABC):
-
     @abstractmethod
     def __init__(
         self,
@@ -151,7 +150,6 @@ class BaseKVSender(ABC):


 class BaseKVReceiver(ABC):
-
     @abstractmethod
     def __init__(
         self,
@@ -209,6 +207,13 @@ class BaseKVReceiver(ABC):
         """
         pass

+    def notify_decode_abort(self):
+        """
+        Notify the decode side that this request was aborted on prefill side.
+        The decode side should also abort the request.
+        """
+        pass
+

 class BaseKVBootstrapServer(ABC):
     @abstractmethod
diff --git a/python/sglang/srt/disaggregation/common/conn.py b/python/sglang/srt/disaggregation/common/conn.py
index f025bbd7aa..124398e706 100644
--- a/python/sglang/srt/disaggregation/common/conn.py
+++ b/python/sglang/srt/disaggregation/common/conn.py
@@ -834,6 +834,16 @@ class CommonKVSender(BaseKVSender):
     ):
         pass

+    def notify_decode_abort(self):
+        if self.conclude_state is not None:
+            return
+        self.kv_mgr.record_failure(
+            self.bootstrap_room,
+            "Aborted on prefill side (input validation failed).",
+        )
+        self.kv_mgr.update_status(self.bootstrap_room, KVPoll.Failed)
+        self.conclude_state = KVPoll.Failed
+
     def _check_bootstrap_timeout(self) -> Optional[KVPoll]:
         if self.init_time is None:
             return None
diff --git a/python/sglang/srt/disaggregation/fake/conn.py b/python/sglang/srt/disaggregation/fake/conn.py
index 09c5f80ce4..9933878b6a 100644
--- a/python/sglang/srt/disaggregation/fake/conn.py
+++ b/python/sglang/srt/disaggregation/fake/conn.py
@@ -89,6 +89,9 @@ class FakeKVSender(BaseKVSender):
     def abort(self):
         self.conclude_state = KVPoll.Failed

+    def notify_decode_abort(self):
+        self.conclude_state = KVPoll.Failed
+

 class FakeKVReceiver(BaseKVReceiver):
     def __init__(
diff --git a/python/sglang/srt/disaggregation/mooncake/conn.py b/python/sglang/srt/disaggregation/mooncake/conn.py
index b21aee9f7c..4771763ff3 100644
--- a/python/sglang/srt/disaggregation/mooncake/conn.py
+++ b/python/sglang/srt/disaggregation/mooncake/conn.py
@@ -1669,7 +1669,6 @@ class MooncakeKVManager(CommonKVManager):


 class MooncakeKVSender(CommonKVSender):
-
     def __init__(
         self,
         mgr: MooncakeKVManager,
@@ -1766,6 +1765,30 @@ class MooncakeKVSender(CommonKVSender):
         self.trace_ctx.abort(abort_info={"reason": "Aborted"})
         self.trace_ctx.trace_req_finish()

+    def notify_decode_abort(self):
+        if self.conclude_state is not None:
+            return
+        self.kv_mgr.update_status(self.bootstrap_room, KVPoll.Failed)
+        self.conclude_state = KVPoll.Failed
+        prefill_unique_rank = (
+            self.kv_mgr.attn_tp_rank * (self.kv_mgr.pp_size * self.kv_mgr.attn_cp_size)
+            + self.kv_mgr.pp_rank * self.kv_mgr.attn_cp_size
+            + self.kv_mgr.attn_cp_rank
+        )
+        transfer_info_dict = self.kv_mgr.transfer_infos.get(self.bootstrap_room)
+        if transfer_info_dict is not None:
+            for transfer_info in transfer_info_dict.values():
+                if not transfer_info.is_dummy:
+                    self.kv_mgr.sync_status_to_decode_endpoint(
+                        transfer_info.endpoint,
+                        transfer_info.dst_port,
+                        self.bootstrap_room,
+                        KVPoll.Failed,
+                        prefill_unique_rank,
+                    )
+        self.trace_ctx.abort(abort_info={"reason": "Aborted on prefill side"})
+        self.trace_ctx.trace_req_finish()
+

 class MooncakeKVReceiver(CommonKVReceiver):
     def __init__(
diff --git a/python/sglang/srt/disaggregation/mori/conn.py b/python/sglang/srt/disaggregation/mori/conn.py
index 81bb7e960b..539dba3cba 100644
--- a/python/sglang/srt/disaggregation/mori/conn.py
+++ b/python/sglang/srt/disaggregation/mori/conn.py
@@ -1551,9 +1551,12 @@ class MoriKVSender(CommonKVSender):
     def abort(self):
         self._finalize_failure("Aborted by AbortReq.")

+    def notify_decode_abort(self):
+        if self.conclude_state is not None:
+            return
+        self._finalize_failure("Aborted on prefill side (input validation failed).")

 class MoriKVReceiver(CommonKVReceiver):
-
     def __init__(
         self,
         mgr: MoriKVManager,
diff --git a/python/sglang/srt/disaggregation/prefill.py b/python/sglang/srt/disaggregation/prefill.py
index ce1afdac3a..bd51587dfe 100644
--- a/python/sglang/srt/disaggregation/prefill.py
+++ b/python/sglang/srt/disaggregation/prefill.py
@@ -578,6 +578,22 @@ class SchedulerDisaggregationPrefillMixin:
                         advance_logprob_pt(i, req)
                         continue

+                # Aborted requests (e.g. input too long): skip KV transfer
+                # and notify decode side to abort as well.
+                if is_aborted(req):
+                    logger.warning(
+                        f"[ABORT_DEBUG] is_aborted=True, calling notify_decode_abort "
+                        f"rid={req.rid} room={req.bootstrap_room}"
+                    )
+                    req.disagg_kv_sender.notify_decode_abort()
+                    release_kv_cache(req, self.tree_cache)
+                    req.finished_reason = req.to_finish
+                    req.to_finish = None
+                    req.time_stats.set_completion_time()
+                    self.output_streamer.stream_output([req], req.return_logprob, None)
+                    advance_logprob_pt(i, req)
+                    continue
+
                 req.output_ids.append(next_token_id)
                 maybe_cache_unfinished_req(req, self.tree_cache)
                 self.disagg_prefill_inflight_queue.append(req)
@@ -688,7 +704,6 @@ class SchedulerDisaggregationPrefillMixin:
         undone_reqs: List[Req] = []
         # Check .poll() for the reqs in disagg_prefill_inflight_queue. If Success, respond to the client and remove it from the queue
         for req, poll in zip(self.disagg_prefill_inflight_queue, polls):
-
             if rids_to_check is not None:
                 if req.rid not in rids_to_check:
                     undone_reqs.append(req)
@@ -713,7 +728,9 @@ class SchedulerDisaggregationPrefillMixin:
                 undone_reqs.append(req)
             elif poll == KVPoll.Success:  # transfer done
                 release_kv_cache(req, self.tree_cache)  # unlock the tree
-                req.finished_reason = FINISH_LENGTH(length=0)
+                # Do not overwrite abort status with FINISH_LENGTH
+                if not isinstance(req.finished_reason, FINISH_ABORT):
+                    req.finished_reason = FINISH_LENGTH(length=0)
                 # FIXME: clean up req's data in transfer engine
                 if hasattr(req.disagg_kv_sender, "clear"):
                     req.disagg_kv_sender.clear()