diff --git a/datadog_sync/model/downtime_schedules.py b/datadog_sync/model/downtime_schedules.py
index 5b2beaf4..b2ee07cb 100644
--- a/datadog_sync/model/downtime_schedules.py
+++ b/datadog_sync/model/downtime_schedules.py
@@ -3,17 +3,35 @@
 # This product includes software developed at Datadog (https://www.datadoghq.com/).
 # Copyright 2019 Datadog, Inc.
 from __future__ import annotations
+import json
+import logging
+import re
 from typing import TYPE_CHECKING, Optional, List, Dict, Tuple
 from datetime import datetime, timedelta, timezone
 from dateutil.parser import parse
 
+from datadog_sync.constants import LOGGER_NAME
 from datadog_sync.utils.base_resource import BaseResource, ResourceConfig
 from datadog_sync.utils.custom_client import PaginationConfig
-from datadog_sync.utils.resource_utils import DowntimeSchedulesDateOperator, SkipResource
+from datadog_sync.utils.resource_utils import (
+    CustomClientHTTPError,
+    DowntimeSchedulesDateOperator,
+    SkipResource,
+)
 
 if TYPE_CHECKING:
     from datadog_sync.utils.custom_client import CustomClient
 
+log = logging.getLogger(LOGGER_NAME)
+
+# Substring of the destination API's 400 body when a create collides with an
+# equivalent existing downtime, e.g.
+#   {"errors":["The downtime being created is a duplicate of one or more
+#    existing downtimes: ['<id>']"]}
+_DUPLICATE_DOWNTIME_MARKER = "duplicate of one or more existing downtimes"
+_DOWNTIME_NOT_FOUND_MARKER = "Downtime not found"
+_SINGLE_DUPLICATE_ID_RE = re.compile(r"existing downtimes:\s*\[\s*(['\"])(?P<id>[^'\"]+)\1\s*\]\s*$")
+
 
 class DowntimeSchedules(BaseResource):
     resource_type = "downtime_schedules"
@@ -87,35 +105,66 @@ def _parse_utc(value):
     def _iso_utc(dt) -> str:
         return dt.isoformat().replace("+00:00", "Z")
 
+    @staticmethod
+    def _http_error_messages(error: CustomClientHTTPError) -> List[str]:
+        """Return string messages from a JSON API error response."""
+        body = error.response_body
+        if not isinstance(body, str):
+            return []
+        try:
+            parsed = json.loads(body)
+        except (json.JSONDecodeError, TypeError):
+            return []
+        errors = parsed.get("errors", []) if isinstance(parsed, dict) else []
+        return [message for message in errors if isinstance(message, str)]
+
+    @classmethod
+    def _single_duplicate_id(cls, error: CustomClientHTTPError) -> Optional[str]:
+        for message in cls._http_error_messages(error):
+            if _DUPLICATE_DOWNTIME_MARKER not in message:
+                continue
+            match = _SINGLE_DUPLICATE_ID_RE.search(message)
+            if match:
+                return match.group("id")
+        return None
+
+    @classmethod
+    def _is_downtime_not_found(cls, error: CustomClientHTTPError) -> bool:
+        return any(_DOWNTIME_NOT_FOUND_MARKER in message for message in cls._http_error_messages(error))
+
+    def _normalize_create_schedule(self, _id: str, resource: Dict) -> None:
+        schedule = resource["attributes"].get("schedule")
+        if not schedule:
+            return
+        now = datetime.now(timezone.utc)
+
+        # Past `end` means the maintenance window has already closed on the
+        # source. Replicating it to the destination would either invent a
+        # new customer-visible maintenance (if we shifted `end` forward) or
+        # 400 with "Downtime cannot be scheduled in the past". Skip: an
+        # ended downtime has nothing left to silence.
+        end_raw = schedule.get("end")
+        if end_raw:
+            end_dt = self._parse_utc(end_raw)
+            if end_dt <= now:
+                raise SkipResource(
+                    str(_id),
+                    self.resource_type,
+                    "Downtime end is in the past.",
+                )
+
+        # Rewrite past `start` forward to now+60s. `end` (if present) is
+        # left as-is per customer intent — the window may shrink but its
+        # original end time is preserved.
+        start_raw = schedule.get("start")
+        if start_raw:
+            start_dt = self._parse_utc(start_raw)
+            if start_dt <= now:
+                schedule["start"] = self._iso_utc(now + timedelta(seconds=60))
+
     async def pre_resource_action_hook(self, _id, resource: Dict) -> None:
         if _id not in self.config.state.destination[self.resource_type]:
-            schedule = resource["attributes"].get("schedule")
-            if not schedule:
-                return
-            now = datetime.now(timezone.utc)
-
-            # Past `end` means the maintenance window has already closed on the
-            # source. Replicating it to the destination would either invent a
-            # new customer-visible maintenance (if we shifted `end` forward) or
-            # 400 with "Downtime cannot be scheduled in the past". Skip: an
-            # ended downtime has nothing left to silence.
-            end_raw = schedule.get("end")
-            if end_raw:
-                end_dt = self._parse_utc(end_raw)
-                if end_dt <= now:
-                    raise SkipResource(
-                        str(_id), self.resource_type,
-                        "Downtime end is in the past.",
-                    )
-
-            # Rewrite past `start` forward to now+60s. `end` (if present) is
-            # left as-is per customer intent — the window may shrink but its
-            # original end time is preserved.
-            start_raw = schedule.get("start")
-            if start_raw:
-                start_dt = self._parse_utc(start_raw)
-                if start_dt <= now:
-                    schedule["start"] = self._iso_utc(now + timedelta(seconds=60))
+            self._normalize_create_schedule(_id, resource)
         else:
             # If start or end times of the resource are in the past, we set to the current destination `start` and `end`
             # this is to avoid unnecessary diff outputs
@@ -139,7 +188,19 @@ async def pre_apply_hook(self) -> None:
     async def create_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
         payload = {"data": resource}
-        resp = await destination_client.post(self.resource_config.base_path, payload)
+        try:
+            resp = await destination_client.post(self.resource_config.base_path, payload)
+        except CustomClientHTTPError as e:
+            duplicate_id = self._single_duplicate_id(e) if e.status_code == 400 else None
+            if duplicate_id is not None:
+                # The API identified exactly one equivalent destination
+                # downtime. Fetch and return it so BaseResource persists the
+                # recovered source-to-destination mapping. Ambiguous or
+                # malformed duplicate responses still propagate as failures.
+                existing = await destination_client.get(self.resource_config.base_path + f"/{duplicate_id}")
+                log.info(f"[downtime_schedules - {_id}] reconciled duplicate with existing destination downtime")
+                return _id, existing["data"]
+            raise
 
         return _id, resp["data"]
 
@@ -147,18 +208,40 @@ async def update_resource(self, _id: str, resource: Dict) -> Tuple[str, Dict]:
         destination_client = self.config.destination_client
         resource["id"] = self.config.state.destination[self.resource_type][_id]["id"]
         payload = {"data": resource}
-        resp = await destination_client.patch(
-            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
-            payload,
-        )
+        try:
+            resp = await destination_client.patch(
+                self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}",
+                payload,
+            )
+        except CustomClientHTTPError as e:
+            if e.status_code == 404 and self._is_downtime_not_found(e):
+                # The mapped destination downtime was removed out-of-band, so
+                # the PATCH target no longer exists ("Downtime not found").
+                # Recreate it now and return the new destination object so the
+                # BaseResource wrapper replaces the stale persisted mapping.
+                # Re-run create-only schedule normalization because the first
+                # pre-action hook took the update branch.
+                resource.pop("id", None)
+                self._normalize_create_schedule(_id, resource)
+                log.info(f"[downtime_schedules - {_id}] recreating missing mapped downtime on destination")
+                return await self.create_resource(_id, resource)
+            raise
 
         return _id, resp["data"]
 
     async def delete_resource(self, _id: str) -> None:
         destination_client = self.config.destination_client
-        await destination_client.delete(
-            self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
-        )
+        try:
+            await destination_client.delete(
+                self.resource_config.base_path + f"/{self.config.state.destination[self.resource_type][_id]['id']}"
+            )
+        except CustomClientHTTPError as e:
+            if e.status_code == 404:
+                # Already gone on the destination: deleting a non-existent
+                # downtime is a successful no-op, not a failure.
+                log.info(f"[downtime_schedules - {_id}] already deleted on destination")
+                return
+            raise
 
     def connect_id(self, key: str, r_obj: Dict, resource_to_connect: str) -> Optional[List[str]]:
         return super(DowntimeSchedules, self).connect_id(key, r_obj, resource_to_connect)
diff --git a/datadog_sync/utils/resource_utils.py b/datadog_sync/utils/resource_utils.py
index aa550b78..0dae7813 100644
--- a/datadog_sync/utils/resource_utils.py
+++ b/datadog_sync/utils/resource_utils.py
@@ -91,6 +91,7 @@ class CustomClientHTTPError(Exception):
     def __init__(self, response, message=None):
         super().__init__(f"{response.status} {response.message} - {message}")
         self.status_code = response.status
+        self.response_body = message
 
 
 class LogsPipelinesOrderIdsComparator(BaseOperator):
diff --git a/tests/unit/test_downtime_schedules_conflict_skip.py b/tests/unit/test_downtime_schedules_conflict_skip.py
new file mode 100644
index 00000000..d3450646
--- /dev/null
+++ b/tests/unit/test_downtime_schedules_conflict_skip.py
@@ -0,0 +1,248 @@
+# Unless explicitly stated otherwise all files in this repository are licensed
+# under the 3-clause BSD style license (see LICENSE).
+# This product includes software developed at Datadog (https://www.datadoghq.com/).
+# Copyright 2019 Datadog, Inc.
+
+"""Unit tests for downtime_schedules apply-time conflict/gone handling.
+
+Two deterministic destination-API rejections must not fail the whole
+downtime_schedules resource for a batch:
+
+1. create path -> 400 "... duplicate of one or more existing downtimes ...":
+   reconcile an unambiguous existing downtime into destination state; reject
+   missing or multiple candidate IDs instead of accepting an unmanaged object.
+2. update/delete path -> 404 "Downtime not found": the mapped destination
+   downtime was removed out-of-band. Recreate it on update so the stale mapping
+   is replaced, and treat delete as an idempotent no-op.
+
+Non-matching 4xx/5xx must still propagate so the retry layer and failure
+accounting engage.
+"""
+
+import asyncio
+import logging
+from types import SimpleNamespace
+
+import pytest
+
+from datadog_sync.constants import LOGGER_NAME
+from datadog_sync.model.downtime_schedules import DowntimeSchedules
+from datadog_sync.utils.resource_utils import CustomClientHTTPError
+
+DUPLICATE_BODY = (
+    '{"errors":["The downtime being created is a duplicate of one or more '
+    "existing downtimes: ['downtime-existing']\"]}"
+)
+NOT_FOUND_BODY = '{"errors":["Downtime not found"]}'
+
+
+def _run(coro):
+    loop = asyncio.new_event_loop()
+    try:
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+
+
+def _http_error(status, message):
+    return CustomClientHTTPError(SimpleNamespace(status=status, message="err"), message=message)
+
+
+def _make_resource():
+    return {"attributes": {"schedule": {"start": "2999-01-01T00:00:00Z"}}}
+
+
+@pytest.fixture
+def downtime(mock_config):
+    return DowntimeSchedules(mock_config)
+
+
+# --- create path: duplicate 400 --------------------------------------------
+
+
+def test_create_duplicate_400_reconciles_single_existing_downtime(downtime):
+    downtime.config.destination_client.post = _http_error_raiser(400, DUPLICATE_BODY)
+
+    async def _get(path):
+        assert path == "/api/v2/downtime/downtime-existing"
+        return {"data": {"id": "downtime-existing", "type": "downtime"}}
+
+    downtime.config.destination_client.get = _get
+    _run(downtime._create_resource("src-id", _make_resource()))
+
+    assert downtime.config.state.destination["downtime_schedules"]["src-id"] == {
+        "id": "downtime-existing",
+        "type": "downtime",
+    }
+
+
+def test_create_duplicate_400_logs_reconciliation_at_info(downtime, caplog):
+    downtime.config.destination_client.post = _http_error_raiser(400, DUPLICATE_BODY)
+
+    async def _get(_path):
+        return {"data": {"id": "downtime-existing", "type": "downtime"}}
+
+    downtime.config.destination_client.get = _get
+    with caplog.at_level(logging.INFO, logger=LOGGER_NAME):
+        _run(downtime.create_resource("src-id", _make_resource()))
+    recs = [r for r in caplog.records if r.name == LOGGER_NAME and "src-id" in r.getMessage()]
+    assert recs, "expected an INFO reconciliation log identifying the downtime"
+    assert all(r.levelno == logging.INFO for r in recs)
+
+
+@pytest.mark.parametrize(
+    "body",
+    [
+        '{"errors":["duplicate of one or more existing downtimes"]}',
+        '{"errors":["The downtime being created is a duplicate of one or more existing downtimes: '
+        "['dest-one', 'dest-two']\"]}",
+    ],
+)
+def test_create_duplicate_without_one_unambiguous_id_propagates(downtime, body):
+    downtime.config.destination_client.post = _http_error_raiser(400, body)
+    with pytest.raises(CustomClientHTTPError) as exc:
+        _run(downtime.create_resource("src-id", _make_resource()))
+    assert exc.value.status_code == 400
+
+
+def test_create_non_duplicate_400_propagates(downtime):
+    downtime.config.destination_client.post = _http_error_raiser(
+        400, '{"errors":["Downtime cannot be scheduled in the past"]}'
+    )
+    with pytest.raises(CustomClientHTTPError) as exc:
+        _run(downtime.create_resource("src-id", _make_resource()))
+    assert exc.value.status_code == 400
+
+
+def test_create_500_propagates(downtime):
+    downtime.config.destination_client.post = _http_error_raiser(500, "Internal Server Error")
+    with pytest.raises(CustomClientHTTPError) as exc:
+        _run(downtime.create_resource("src-id", _make_resource()))
+    assert exc.value.status_code == 500
+
+
+def test_create_200_unchanged(downtime):
+    async def _ok(_path, _payload):
+        return {"data": {"id": "dest-id", "type": "downtime"}}
+
+    downtime.config.destination_client.post = _ok
+    _id, data = _run(downtime.create_resource("src-id", _make_resource()))
+    assert _id == "src-id"
+    assert data == {"id": "dest-id", "type": "downtime"}
+
+
+# --- update path: not-found 404 --------------------------------------------
+
+
+def _seed_dest(downtime, _id="src-id", dest_id="dest-id"):
+    downtime.config.state.destination["downtime_schedules"][_id] = {"id": dest_id}
+
+
+def test_update_not_found_404_recreates_and_replaces_stale_mapping(downtime):
+    _seed_dest(downtime)
+    downtime.config.destination_client.patch = _http_error_raiser(404, NOT_FOUND_BODY)
+    posted = []
+
+    async def _post(_path, payload):
+        posted.append(payload)
+        return {"data": {"id": "replacement-id", "type": "downtime"}}
+
+    downtime.config.destination_client.post = _post
+    _run(downtime._update_resource("src-id", _make_resource()))
+
+    assert posted[0]["data"].get("id") is None
+    assert downtime.config.state.destination["downtime_schedules"]["src-id"] == {
+        "id": "replacement-id",
+        "type": "downtime",
+    }
+
+
+def test_update_not_found_404_logs_recreation_at_info(downtime, caplog):
+    _seed_dest(downtime)
+    downtime.config.destination_client.patch = _http_error_raiser(404, NOT_FOUND_BODY)
+
+    async def _post(_path, _payload):
+        return {"data": {"id": "replacement-id", "type": "downtime"}}
+
+    downtime.config.destination_client.post = _post
+    with caplog.at_level(logging.INFO, logger=LOGGER_NAME):
+        _run(downtime.update_resource("src-id", _make_resource()))
+    recs = [r for r in caplog.records if r.name == LOGGER_NAME and "src-id" in r.getMessage()]
+    assert recs and all(r.levelno == logging.INFO for r in recs)
+
+
+def test_update_unrelated_404_propagates(downtime):
+    _seed_dest(downtime)
+    downtime.config.destination_client.patch = _http_error_raiser(404, '{"errors":["Route not found"]}')
+    with pytest.raises(CustomClientHTTPError) as exc:
+        _run(downtime.update_resource("src-id", _make_resource()))
+    assert exc.value.status_code == 404
+
+
+def test_update_400_propagates(downtime):
+    _seed_dest(downtime)
+    downtime.config.destination_client.patch = _http_error_raiser(400, "Bad Request")
+    with pytest.raises(CustomClientHTTPError) as exc:
+        _run(downtime.update_resource("src-id", _make_resource()))
+    assert exc.value.status_code == 400
+
+
+def test_update_200_unchanged(downtime):
+    _seed_dest(downtime)
+
+    async def _ok(_path, _payload):
+        return {"data": {"id": "dest-id", "type": "downtime"}}
+
+    downtime.config.destination_client.patch = _ok
+    _id, data = _run(downtime.update_resource("src-id", _make_resource()))
+    assert _id == "src-id"
+    assert data == {"id": "dest-id", "type": "downtime"}
+
+
+# --- delete path: not-found 404 is a no-op ---------------------------------
+
+
+def test_delete_404_is_noop(downtime):
+    _seed_dest(downtime)
+    downtime.config.destination_client.delete = _http_error_raiser(404, NOT_FOUND_BODY)
+    # Already gone == delete succeeded; must not raise.
+    assert _run(downtime.delete_resource("src-id")) is None
+
+
+def test_delete_500_propagates(downtime):
+    _seed_dest(downtime)
+    downtime.config.destination_client.delete = _http_error_raiser(500, "Internal Server Error")
+    with pytest.raises(CustomClientHTTPError) as exc:
+        _run(downtime.delete_resource("src-id"))
+    assert exc.value.status_code == 500
+
+
+# --- batch resilience: one duplicate doesn't stop siblings ------------------
+
+
+def test_duplicate_does_not_block_siblings(downtime):
+    calls = {"n": 0}
+
+    async def _post(_path, _payload):
+        calls["n"] += 1
+        if calls["n"] == 2:
+            raise _http_error(400, DUPLICATE_BODY)
+        return {"data": {"id": f"dest-{calls['n']}"}}
+
+    async def _get(_path):
+        return {"data": {"id": "downtime-existing"}}
+
+    downtime.config.destination_client.post = _post
+    downtime.config.destination_client.get = _get
+    synced = []
+    for src in ("a", "b", "c"):
+        _id, _ = _run(downtime.create_resource(src, _make_resource()))
+        synced.append(_id)
+    assert synced == ["a", "b", "c"]
+
+
+def _http_error_raiser(status, message):
+    async def _raise(*_args, **_kwargs):
+        raise _http_error(status, message)
+
+    return _raise