diff --git a/docs/architecture/inspect-app/decisions/003-e2e-testing.md b/docs/architecture/inspect-app/decisions/003-e2e-testing.md
index 967d381..b01aff0 100644
--- a/docs/architecture/inspect-app/decisions/003-e2e-testing.md
+++ b/docs/architecture/inspect-app/decisions/003-e2e-testing.md
@@ -15,6 +15,46 @@ cassettes, no fake VideoIPath server.
 These tests are **not** required on every CI push; they are run locally when a
 developer has an instance available (`poetry run test-e2e`).
 
+## Maintenance lifecycle suite
+
+Run the focused maintenance suite with:
+
+```bash
+poetry run test-e2e tests/e2e/apps/test_inspect_maintenance.py
+```
+
+The suite targets the 2026.2 API and has been verified against **2026.2.0**.
+
+Its eight cases form three groups:
+
+- **Lifecycle — four cases:** one-time, daily, weekly, and monthly bookings.
+  Checks include persisted metadata and dates, revision conflicts, lock
+  preservation, start-now, filtering, and independent recurring occurrences.
+- **Resource targets — three cases:** modules, ports, and edges. Checks include
+  resource relationships and full snapshot reads.
+- **Immediate and open-ended — one case:** nullable timestamps, schedule updates,
+  and preservation of staged topology edits.
+
+Tests assert persisted server values and returned booking IDs, not just operation
+success.
+
+Tests use exclusively their own mock-driver devices, `E2E-` labels, and the
+`vipat-e2e` tag. Each test also has a unique tag for finding its bookings after a
+rename or a lost response. This module disables the shared session sweep and
+cleans only its own resources.
+
+Cleanup also runs on test failure, in this order:
+
+1. Unlock any remaining test bookings.
+2. Delete the bookings and verify their absence.
+3. Remove the mock devices and verify their absence from Inspect and Inventory.
+
+Impact preview and current-impact calls assert empty service impacts on these
+isolated mocks; populated impact parsing is covered offline. Start-now assertions
+use server state and resolved windows, allowing server/runner clock skew.
+
+The normal E2E gate applies, so these tests do not run in the default offline suite.
+
 ## Consequences
 
 - Highest confidence: tests exercise the real API, auth, and payload shapes.
diff --git a/docs/architecture/inspect-app/decisions/006-collector-only-endpoints.md b/docs/architecture/inspect-app/decisions/006-collector-only-endpoints.md
index 1068016..e7c1cab 100644
--- a/docs/architecture/inspect-app/decisions/006-collector-only-endpoints.md
+++ b/docs/architecture/inspect-app/decisions/006-collector-only-endpoints.md
@@ -12,6 +12,7 @@
 | Collector actions | `POST /rest/v2/actions/status/collector/*` (`updateTopology`, lookups, …) |
 | Network actions | `POST /rest/v2/actions/status/network/{addDevices,syncDevices,updateVirtualInstances,updateVirtualTemplates,addVirtualTopology}` |
 | Tag actions | `POST /rest/v2/actions/status/tags/{assignTag,unassignTag}` |
+| Maintenance actions | `POST /rest/v2/actions/status/pathman/{updateMaintenance,validateMaintenanceImpactDetailed,fetchMaintenanceImpact}` (exact allow-list entries) |
 | Alarm reads | `GET /rest/v2/data/status/alarms/current/…` |
 | Virtual reads | `GET /rest/v2/data/status/network/{virtualDevices,virtualTemplates}/**` |
 | System probes | `GET /rest/v2/data/status/system/about/…` (version gating) |
@@ -28,9 +29,14 @@ only. `app.topology` remains the escape hatch for raw, revisioned
 
 ## Consequences
 
-- **No `_rev` is available** to the Inspect package, and the write path
-  enforces none (last-writer-wins). Write consistency is solved client-side —
+- **Topology collector reads have no `_rev`**, and topology writes enforce
+  none (last-writer-wins). Topology write consistency is solved client-side —
   [ADR-007](./007-write-consistency.md).
+- Maintenance collector reads expose `rev`. Maintenance updates use a fresh read
+  and submit that revision; `expected_rev` optionally checks the caller's baseline.
+  These immediate actions run independently of topology transactions. The three
+  pathman actions above were verified on **2026.2.0**; no general pathman prefix
+  is allowed. Compatibility with older versions is untested.
 - Persisted forms for `replace*` payloads come from collector-namespace lookups
   (`lookupInspectDevice`, `lookupInspectVertexByIds`,
   `lookupInspectEdgesByIds`), not from `nGraphElements` reads.
diff --git a/docs/architecture/inspect-app/endpoints.md b/docs/architecture/inspect-app/endpoints.md
index 0cd939f..0868b56 100644
--- a/docs/architecture/inspect-app/endpoints.md
+++ b/docs/architecture/inspect-app/endpoints.md
@@ -608,8 +608,10 @@ document and the known Inspect UI payload shape.
 ## `GET /rest/v2/data/status/collector/maintenanceBookings/**`
 
 Purpose: maintenance booking status relevant to Inspect service/link display.
+Loaded lazily as the maintenance snapshot section; a full collector snapshot can
+provide the same collection. Empty responses contain `maintenanceBookings._items=[]`.
 
-Observed response on this instance:
+Populated shape verified against **VideoIPath 2026.2.0**, anonymized:
 
 ```json
 {
@@ -617,7 +619,47 @@ Observed response on this instance:
     "status": {
       "collector": {
         "maintenanceBookings": {
-          "_items": []
+          "_items": [
+            {
+              "_id": "maintenance-a",
+              "_vid": "maintenance-a-view",
+              "rev": "revision-a",
+              "generic": {
+                "descriptor": {
+                  "label": "maintenance-a",
+                  "desc": "Example window"
+                },
+                "locked": false,
+                "state": 0
+              },
+              "scheduleInfo": {
+                "startTimestamp": 1893456000000,
+                "endTimestamp": 1893459600000,
+                "infinite": false
+              },
+              "action": "nothing",
+              "trigger": "create",
+              "allowOverlap": false,
+              "switchFormatStateOnReroute": false,
+              "tags": [
+                "maintenance-example"
+              ],
+              "devices": [
+                {
+                  "label": "device-a",
+                  "context": {
+                    "devicePid": "device-a"
+                  },
+                  "pid": [
+                    "device-a"
+                  ]
+                }
+              ],
+              "modules": [],
+              "ports": [],
+              "edges": []
+            }
+          ]
         }
       }
     }
@@ -625,6 +667,236 @@ Observed response on this instance:
 }
 ```
 
+The record has three main groups:
+
+- **Identity and metadata:** `_id`, `rev`, `generic`, and `tags` identify the booking
+  and describe its label, lock, and state.
+- **Timing and behavior:** `scheduleInfo` describes the resolved window; `action`,
+  `trigger`, and the two boolean settings describe its behavior.
+- **Selected resources:** `devices`, `modules`, `ports`, and `edges` identify what
+  the booking affects.
+
+State `0` is Scheduled, `1` Active; preserve unknown values.
+
+Module and port resources include `modulePid` and `portPid` in their contexts. Edges carry `id`,
+`from`, and `to` resources. Resolved collector schedules are distinct from writable
+schedules; `infinite=true` may accompany a large integer end sentinel.
+
+## Maintenance actions (`status/pathman`)
+
+These three exact endpoints, including live writes, were verified on **2026.2.0**.
+Older-version compatibility is untested. Requests use `header: {"id": 0}`.
+
+### `POST /rest/v2/actions/status/pathman/updateMaintenance`
+
+Use the three lists in `data` to select the operation. Keep unused lists empty.
+
+| Operation | Resource field | Other required fields |
+| --- | --- | --- |
+| Create | `serviceDefinition.devicePids` and `edgeIds` | `scheduleInfo` and the complete service definition |
+| Update | `updater.pids` and `edgeIds` | `id`, fresh `rev`, `inheritable`, and the complete updater |
+| Delete | Explicit booking IDs in `delete` | No resource definition |
+
+Device, module, and port targets all use canonical PIDs.
+
+#### Create a booking
+
+`scheduleInfo` defines the window. `serviceDefinition` groups the metadata,
+selected resources, and action settings.
+
+```json
+{
+  "header": {
+    "id": 0
+  },
+  "data": {
+    "create": [
+      {
+        "scheduleInfo": {
+          "type": "once",
+          "startTimestamp": 1893456000000,
+          "endTimestamp": 1893459600000
+        },
+        "serviceDefinition": {
+          "descriptor": {
+            "label": "maintenance-a",
+            "desc": "Example window"
+          },
+          "devicePids": [
+            "device-a"
+          ],
+          "edgeIds": [],
+          "tags": [],
+          "action": "nothing",
+          "trigger": "create",
+          "allowOverlap": false,
+          "switchFormatStateOnReroute": false
+        }
+      }
+    ],
+    "update": [],
+    "delete": []
+  }
+}
+```
+
+#### Update a booking
+
+Place the following entry inside `data.update`, with `create=[]` and `delete=[]`.
+
+The top-level `id` and `rev` select the booking and its revision. `inheritable`
+contains the intended schedule and lock state. `updater` carries the complete
+metadata, resources, and action settings.
+
+```json
+{
+  "id": "maintenance-a",
+  "rev": "revision-a",
+  "inheritable": {
+    "scheduleInfo": {
+      "type": "once",
+      "startTimestamp": null,
+      "endTimestamp": null
+    },
+    "locked": false
+  },
+  "updater": {
+    "descriptor": {
+      "label": "maintenance-a",
+      "desc": "Example window"
+    },
+    "pids": [
+      "device-a"
+    ],
+    "edgeIds": [],
+    "tags": [],
+    "action": "nothing",
+    "trigger": "create",
+    "allowOverlap": false,
+    "switchFormatStateOnReroute": false
+  }
+}
+```
+
+Schedule timestamps are epoch milliseconds:
+
+- `startTimestamp: null` means immediate start.
+- `endTimestamp: null` means open-ended duration.
+
+Valid actions are `nothing`, `invalidate`, `reroute`, and `rerouteSA`. Valid
+triggers are `create` and `active`. The SDK preserves the fresh lock value when
+the public `locked` argument is omitted.
+
+#### Delete bookings
+
+Send this object as `data`. Every entry in `delete` is an explicit server booking ID.
+
+```json
+{
+  "create": [],
+  "update": [],
+  "delete": [
+    "maintenance-a"
+  ]
+}
+```
+
+#### Interpret the result
+
+Success requires both the REST header and `data.result.ok`. The response `data`
+contains an operation result and details keyed by booking ID:
+
+```json
+{
+  "result": {
+    "ok": true,
+    "msg": [
+      "Example success message"
+    ]
+  },
+  "details": {
+    "maintenance-a": {
+      "rev": "revision-b",
+      "isCancel": false,
+      "status": 0,
+      "type": "generic",
+      "resolvable": false
+    }
+  }
+}
+```
+
+The keys of `details` are authoritative booking IDs; never parse messages or infer
+IDs from labels. On deletion `isCancel` is true. The SDK exposes typed result and
+detail objects, and does not retry writes after uncertain transport failures.
+
+### `POST /rest/v2/actions/status/pathman/validateMaintenanceImpactDetailed`
+
+Read-only preview. Data contains `pids`, `edgeIds`, `label`, `scheduleInfo`, and
+`id` (null for creation; the booking ID when editing). Unlike creation it uses
+`pids`. The response data maps service IDs to impact reports; an empty map is valid.
+
+Reports contain `label`, `start`, `end`, `protection`, `maxSaBefore`, `maxSaAfter`,
+and `entries` with time windows, `saBefore`, `saAfter`, and contributors keyed by
+booking ID (`{"label": "maintenance-a"}`). SA codes remain raw values rather than
+being interpreted using Inspect status severity codes.
+
+### `POST /rest/v2/actions/status/pathman/fetchMaintenanceImpact`
+
+Read-only current impact. Data is `{"ids": ["maintenance-a"]}`. Response data is
+keyed first by booking ID, then by service ID, using the same impact report shape.
+
+### Recurrence encoding and the verified server limitation
+
+The UI's native recurrence DTO is preserved in transport models:
+
+```json
+{
+  "type": "recurring",
+  "pattern": {
+    "pattern": 1,
+    "startTime": 1893456000000,
+    "endTime": 1896134400000,
+    "timeZoneId": "Europe/Berlin",
+    "weekDays": [
+      1,
+      5
+    ],
+    "iterationFilter": []
+  },
+  "instance": {
+    "localStartTime": 14400,
+    "localEndTime": 18000,
+    "iterationFilter": []
+  }
+}
+```
+
+The two nested objects describe different parts of the rule:
+
+- **`pattern`:** frequency, recurrence bounds, timezone, and selected weekdays.
+  Frequency `0/1/2` means daily/weekly/monthly; weekdays use ISO numbering 1–7.
+- **`instance`:** the local start and end times for each occurrence, expressed
+  as seconds since midnight.
+
+**Do not use this DTO to assume working native recurrence
+on 2026.2.0**: live creation yielded a broad resolved booking whose updates were
+rejected. The public SDK expands finite rules into dated `once` creates in one
+request and previews each window separately. Updates always address one ID and
+require a one-time schedule. There is no series-management API.
+
+Expansion uses the IANA zone and explicit recurrence bounds, with at most 1,000
+complete windows. Monthly rules use the local start date's day and skip missing
+days.
+
+Overnight windows are supported; DST gaps are rejected and ambiguous times
+use the first occurrence. Transport models preserve iteration filters; high-level
+expansion rejects nonempty filters whose server semantics remain unverified.
+
+Live acceptance covered one-time, daily, weekly, and monthly creation on test-owned
+mock resources, preview/current impact, updates, locking/unlocking, start-now, and
+deletion. Test bookings were removed before their resources.
+
 ## `GET /rest/v2/data/status/collector/superProfiles/**`
 
 Purpose: routing/super-profile status data used by the Inspect aggregate.
diff --git a/docs/architecture/inspect-app/models.md b/docs/architecture/inspect-app/models.md
index 87ba262..b9aca17 100644
--- a/docs/architecture/inspect-app/models.md
+++ b/docs/architecture/inspect-app/models.md
@@ -325,6 +325,44 @@ service = app.inspect.get_service_by_booking_id("booking-1001")
 all_services = app.inspect.services
 ```
 
+### Maintenance bookings (`InspectMaintenanceBooking`)
+
+The lazy maintenance snapshot section reads `collector.maintenanceBookings` and
+reuses it when present in a full response. `InspectMaintenanceBooking` exposes
+identity/revision, descriptors, tags, state, lock status, resolved timestamps,
+action settings, and selected topology resources.
+
+`.raw` preserves unresolved IDs.
+Devices, modules, ports, and edges expose related `.maintenance_bookings`.
+
+Use these models to describe a write:
+
+| Model | Purpose |
+| --- | --- |
+| `MaintenanceTargets` | Select devices, modules, ports, and edges by canonical ID or Inspect object. |
+| `MaintenanceOnceSchedule` | Describe one immediate or scheduled window. |
+| `MaintenanceRecurringSchedule` | Describe a finite daily, weekly, or monthly rule. |
+| `MaintenanceBookingSpec` | Combine metadata, targets, a schedule, and action settings. |
+
+Writable schedules are separate from `InspectApiMaintenanceWindow`, which
+represents resolved collector timestamps.
+
+Recurring rules expand into dated one-time bookings because
+native recurrence could not be updated on the verified **2026.2.0** server. Each
+server ID is managed independently with an explicit one-time schedule.
+
+Results and errors have separate roles:
+
+- `MaintenanceResult.details` maps server IDs to typed `MaintenanceChange` objects.
+- `MaintenanceImpact` models service impacts, keeping raw SA codes and timestamp
+  sentinels.
+- `InspectMaintenanceError` carries operation, booking IDs, and response details.
+- `InspectMaintenanceConflictError` adds expected and actual revisions.
+
+The [usage guide](../../getting-started-guide/03_B_Inspect.md#5-maintenance-bookings)
+and [endpoint reference](./endpoints.md#maintenance-actions-statuspathman) describe
+schedule limits, payloads, and the version-specific behavior.
+
 ## Transport DTO Examples
 
 A path item links one service or booking to the devices and ports used to carry
@@ -829,6 +867,7 @@ Transport DTOs are split by payload area under `apps/inspect/model/`:
 - `alarms.py` — current-alarm wire models (`status/alarms/current`)
 - `tags.py` — `assignTag` / `unassignTag` request DTOs
 - `virtual.py` — virtual-device and port-template wire models
+- `maintenance.py` — maintenance collector, schedule, action, result, and impact DTOs
 
 User-facing domain models and the snapshot:
 
@@ -841,6 +880,8 @@ User-facing domain models and the snapshot:
 - `domain/edge.py` — `InspectEdge`
 - `domain/service.py` — `InspectService`
 - `domain/alarm.py` — `InspectAlarm`
+- `domain/maintenance.py` — `InspectMaintenanceBooking`
+- `maintenance.py` — public booking specifications, targets, and schedule expansion
 
 When adding transport fields, prefer extending the nearest existing `InspectApi*`
 DTO. When adding user-facing behaviour, extend the domain layer and snapshot
diff --git a/docs/examples/04_inspect/04_maintenance_bookings.py b/docs/examples/04_inspect/04_maintenance_bookings.py
new file mode 100644
index 0000000..119cb2a
--- /dev/null
+++ b/docs/examples/04_inspect/04_maintenance_bookings.py
@@ -0,0 +1,143 @@
+"""Maintenance examples for VideoIPath 2026.2.0; use a dedicated test device.
+
+Configuration comes from the usual VIPAT_* environment variables. Running this
+script writes maintenance bookings and deletes them afterward. Adapt DEVICE_ID.
+No topology transaction is required. Older server versions have not been tested.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, time, timedelta
+
+from videoipath_automation_tool.apps.inspect import (
+    MaintenanceBookingSpec,
+    MaintenanceOnceSchedule,
+    MaintenanceRecurringSchedule,
+    MaintenanceTargets,
+)
+from videoipath_automation_tool.apps.videoipath_app import VideoIPathApp
+
+DEVICE_ID = "device-a"
+
+
+def main() -> None:
+    # --- Connect and select a dedicated test device ---
+
+    app = VideoIPathApp()
+    device = app.inspect.get_device(DEVICE_ID)
+
+    if device is None:
+        raise ValueError("Set DEVICE_ID to a dedicated test device in Inspect.")
+
+    # --- Describe a one-time window ---
+    # Keep the target, schedule, and booking description separate so each is easy
+    # to adapt. All datetime values must include a timezone.
+
+    now = datetime.now(UTC)
+    targets = MaintenanceTargets(devices=[device])
+
+    schedule = MaintenanceOnceSchedule(
+        start=now + timedelta(days=1),
+        end=now + timedelta(days=1, hours=2),
+    )
+
+    spec = MaintenanceBookingSpec(
+        label="E2E-maintenance-example",
+        description="Example maintenance window",
+        tags=["vipat-e2e"],
+        targets=targets,
+        schedule=schedule,
+    )
+
+    # --- Preview, then create ---
+    # Preview is read-only. Creation is a separate, immediate server action.
+
+    preview = app.inspect.validate_maintenance_booking(spec)
+    print("Preview:", preview)
+
+    created = app.inspect.create_maintenance_booking(spec)
+
+    # Only the server's result identifies the new bookings. Do not infer IDs
+    # from the label or from a human-readable success message.
+    booking_ids = list(created.details)
+
+    try:
+        # --- Inspect, lock, and start the booking ---
+
+        for booking_id in booking_ids:
+            booking = app.inspect.get_maintenance_booking(booking_id)
+
+            if booking is None:
+                raise RuntimeError("Created booking is not visible; inspect server state before retrying.")
+
+            print(booking.label, booking.state, booking.devices)
+
+            # Updates always need the complete specification. The expected
+            # revision prevents overwriting a booking that changed since reading.
+            app.inspect.update_maintenance_booking(
+                booking_id,
+                spec,
+                locked=True,
+                expected_rev=booking.rev,
+            )
+
+            # An omitted start means "now" on the server. Keep the other fields
+            # by copying the spec, and explicitly unlock the booking here.
+            immediate_schedule = MaintenanceOnceSchedule(end=now + timedelta(hours=1))
+            start_now = spec.model_copy(update={"schedule": immediate_schedule})
+
+            app.inspect.update_maintenance_booking(booking_id, start_now, locked=False)
+
+        if booking_ids:
+            impact = app.inspect.get_maintenance_impact(booking_ids)
+            print("Impact:", impact)
+
+    finally:
+        # --- Remove the example bookings, even if an earlier step fails ---
+        # Explicitly unlock before deleting; omitted locked preserves its value.
+        # This example reuses the original complete spec when unlocking.
+
+        for booking_id in booking_ids:
+            app.inspect.update_maintenance_booking(booking_id, spec, locked=False)
+
+        if booking_ids:
+            app.inspect.delete_maintenance_bookings(booking_ids)
+
+    # --- Describe a recurring rule ---
+    # Daily rules need no weekdays. Monthly rules use the local start date's day.
+
+    rule = MaintenanceRecurringSchedule(
+        frequency="weekly",
+        # Finite bounds limit which dated windows will be created.
+        start=now + timedelta(days=1),
+        end=now + timedelta(days=32),
+        # The local window repeats in this timezone on Monday and Friday.
+        timezone="Europe/Berlin",
+        local_start=time(4),
+        local_end=time(5),
+        weekdays=[1, 5],  # ISO numbering: Monday=1, Sunday=7.
+    )
+
+    recurring = spec.model_copy(update={"schedule": rule})
+
+    # --- Inspect the expanded dates, preview their impact, then create ---
+
+    windows = rule.expand()
+    print("Dated windows:", windows)
+
+    recurring_preview = app.inspect.validate_maintenance_booking(recurring)
+    print("Recurring preview:", recurring_preview)
+
+    result = app.inspect.create_maintenance_booking(recurring)
+
+    # Each dated booking has its own ID. Manage IDs individually with one-time
+    # schedules; collector windows cannot reconstruct the original recurrence rule.
+    recurring_ids = list(result.details)
+
+    # Remove every dated booking created by this example.
+    if recurring_ids:
+        app.inspect.delete_maintenance_bookings(recurring_ids)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/docs/examples/README.md b/docs/examples/README.md
index 751875d..2b61cfe 100644
--- a/docs/examples/README.md
+++ b/docs/examples/README.md
@@ -76,6 +76,9 @@ Each scenario is implemented twice — once with the modern **Inspect** app, onc
   changes; detect concurrent edits and `rebase` + retry.
 - [03_services_and_paths.py](04_inspect/03_services_and_paths.py) — inspect services, their paths, and
   the service-impact guard.
+- [04_maintenance_bookings.py](04_inspect/04_maintenance_bookings.py) — preview, create, update,
+  lock, start, and delete maintenance; daily/weekly/monthly rules expand into dated bookings.
+  Maintenance lifecycle verified on **2026.2.0** only.
 
 ### 05 — Administration
 
diff --git a/docs/getting-started-guide/03_B_Inspect.md b/docs/getting-started-guide/03_B_Inspect.md
index 43bd148..6c5935e 100644
--- a/docs/getting-started-guide/03_B_Inspect.md
+++ b/docs/getting-started-guide/03_B_Inspect.md
@@ -206,9 +206,237 @@ info = app.inspect.get_sync_info(["device12"])
 app.inspect.sync_devices(["device12"], add_only=True, conflict_strategy=ConflictStrategy.STRICT)
 ```
 
-## 5. Notes
+## 5. Maintenance bookings
 
-- Inspect uses **only** the collector API surface at runtime; it never calls the
+Maintenance methods execute immediately, independently of `transaction()` and
+`commit()`. These examples use an existing `app` instance and a dedicated test
+device named `device-a`.
+
+### Prepare a one-time booking
+
+Describe the selected resources, the time window, and the booking metadata.
+Creating these Python objects does not change anything on the server.
+
+```python
+from datetime import UTC, datetime, timedelta
+
+from videoipath_automation_tool.apps.inspect import (
+    MaintenanceBookingSpec,
+    MaintenanceOnceSchedule,
+    MaintenanceTargets,
+)
+
+# Select the resources that the maintenance will affect.
+device = app.inspect.get_device("device-a")
+
+if device is None:
+    raise ValueError("Create or select a test device before running this example.")
+
+targets = MaintenanceTargets(devices=[device])
+
+
+# Define one future window using timezone-aware datetimes.
+now = datetime.now(UTC)
+
+schedule = MaintenanceOnceSchedule(
+    start=now + timedelta(days=1),
+    end=now + timedelta(days=1, hours=2),
+)
+
+
+# Combine the metadata, resources, and schedule into a complete specification.
+spec = MaintenanceBookingSpec(
+    label="maintenance-a",
+    tags=["maintenance-example"],
+    targets=targets,
+    schedule=schedule,
+)
+```
+
+Targets accept canonical resource IDs or matching Inspect domain objects. String
+module/port IDs must be globally qualified PIDs, not display labels or local keys.
+
+Defaults match the UI: no overlap, `action="nothing"`, `trigger="create"`, and no
+format switching. Supported actions are `nothing`, `invalidate`, `reroute`, and
+`rerouteSA`; triggers are `create` and `active`.
+
+### Preview and create
+
+Previewing is read-only. Call the creation method separately when ready to write.
+
+```python
+# Preview groups reports by affected service ID, with one report per affected window.
+preview = app.inspect.validate_maintenance_booking(spec)
+print("Preview:", preview)
+
+
+# Create immediately and keep the IDs returned by the server.
+created = app.inspect.create_maintenance_booking(spec)
+booking_ids = list(created.details)
+
+# Do not infer booking IDs from labels or success messages.
+if not booking_ids:
+    raise RuntimeError("No booking IDs returned; inspect server state before retrying.")
+```
+
+### Read and filter
+
+```python
+# Fetch one booking by its server ID.
+booking = app.inspect.get_maintenance_booking(booking_ids[0])
+
+if booking is None:
+    raise RuntimeError("The booking is not visible; refresh before continuing.")
+
+print(booking.label, booking.state, booking.starts_at, booking.ends_at)
+
+
+# Search labels and tags, optionally limiting the state.
+scheduled = app.inspect.find_maintenance_bookings(
+    state="scheduled",
+    search="maintenance-example",
+)
+
+
+# Current impact is grouped by booking ID, then by affected service ID.
+impact = app.inspect.get_maintenance_impact(booking_ids)
+print("Current impact:", impact)
+```
+
+`maintenance_bookings` reads all bookings. `find_maintenance_bookings(state="all",
+search="")` supports `all`, `active`, and `scheduled`, with case-insensitive label/tag
+search. `get_maintenance_booking(id)` returns `None` when absent. The collection is
+loaded once per snapshot without device hydration; full snapshots reuse their
+existing maintenance data. Call `refresh()` to observe changes from other clients.
+
+Bookings expose `rev`, `label`, `description`, `tags`, `state`, `locked`, action
+settings, `starts_at`/`ends_at`, and selected `devices`, `modules`, `ports`, and `edges`.
+Unknown numeric states remain available as integers. `raw` retains unresolved
+resource contexts and PID segments, even if the corresponding topology object is
+missing.
+
+Resource objects expose `.maintenance_bookings`; device/module relationships
+also include explicitly selected descendants and edge endpoints. Ancestor bookings
+are available on that ancestor rather than repeated on every descendant.
+
+### Update, lock, start, and delete
+
+Updates require the complete specification and an explicit one-time schedule for
+the booking ID. They fetch the current booking revision before posting.
+
+`locked=None` preserves its fresh lock state; booleans change it. `expected_rev`
+raises `InspectMaintenanceConflictError` before writing if the revision differs.
+
+```python
+# Lock the booking only if its revision still matches the one we read.
+app.inspect.update_maintenance_booking(
+    booking.id,
+    spec,
+    locked=True,
+    expected_rev=booking.rev,
+)
+
+
+# Reuse the complete specification, changing only the intended schedule.
+# An omitted start means "now" on the server; end is an explicit timestamp.
+immediate_schedule = MaintenanceOnceSchedule(
+    end=datetime.now(UTC) + timedelta(hours=1),
+)
+
+start_now = spec.model_copy(update={"schedule": immediate_schedule})
+
+app.inspect.update_maintenance_booking(
+    booking.id,
+    start_now,
+    locked=False,
+)
+
+
+# Delete the explicit IDs created by this example.
+app.inspect.delete_maintenance_bookings(booking_ids)
+```
+
+`MaintenanceOnceSchedule(start=None, end=...)` starts now; `end=None` is open-ended.
+Supply timezone-aware datetimes; the SDK converts them to epoch milliseconds.
+
+### Recurring schedules
+
+```python
+from datetime import time
+
+from videoipath_automation_tool.apps.inspect import MaintenanceRecurringSchedule
+
+# Define a finite rule. Other supported frequencies are "daily" and "monthly".
+rule = MaintenanceRecurringSchedule(
+    frequency="weekly",
+
+    # Only complete windows within these bounds will be included.
+    start=datetime(2030, 1, 1, tzinfo=UTC),
+    end=datetime(2030, 2, 1, tzinfo=UTC),
+
+    # Repeat the local window every Monday and Friday in this timezone.
+    timezone="Europe/Berlin",
+    local_start=time(4),
+    local_end=time(5),
+    weekdays=[1, 5],  # ISO Monday=1; use weekdays only for weekly rules.
+)
+
+recurring_spec = spec.model_copy(update={"schedule": rule})
+
+
+# Inspect the individual dates before contacting the server.
+windows = rule.expand()
+
+for window in windows:
+    print(window.start, window.end)
+
+
+# Preview each window, then create all dated bookings in one write action.
+preview = app.inspect.validate_maintenance_booking(recurring_spec)
+print("Recurring preview:", preview)
+
+created = app.inspect.create_maintenance_booking(recurring_spec)
+recurring_ids = list(created.details)
+
+
+# Each occurrence has its own ID. Remove the example occurrences when finished.
+if recurring_ids:
+    app.inspect.delete_maintenance_bookings(recurring_ids)
+```
+
+On the verified server, native recurring maintenance requests created a broad window
+that could not subsequently be updated. The SDK therefore expands rules into dated
+one-time bookings and creates them in one action. Each has an independent server ID
+in `created.details`. Preview makes one read-only request per window. Keep the rule
+in your application if needed: collector windows do not preserve recurrence intent.
+
+Update, lock, start, and delete individual IDs; there is no series-management API.
+
+The expansion rules are:
+
+- **Bounds:** include only complete windows within the finite range, up to 1,000.
+- **Monthly dates:** use the local start date's day and skip months without that day.
+- **Overnight windows:** an end time earlier than the start time crosses midnight.
+- **Daylight saving time:** reject missing local times; use the first occurrence
+  when a local time is ambiguous.
+- **Iteration filters:** raw transport models preserve `iterationFilter`, but
+  expansion rejects nonempty filters because their server-specific semantics
+  are unverified.
+
+### Errors and cached reads
+
+`InspectMaintenanceError` includes `operation`, `booking_ids`, `detail`, and
+`response`. Both envelope and operation failures are checked. Writes are never
+automatically retried after an uncertain transport failure.
+
+A successful write invalidates maintenance/status/service caches without clearing staged topology
+edits or making follow-up reads; a later refresh failure cannot hide that success.
+
+See the [complete maintenance example](../examples/04_inspect/04_maintenance_bookings.py).
+
+## 6. Notes
+
+- Inspect uses the Inspect API surface, including its maintenance actions; it never calls the
   legacy `nGraphElements` / `edgesByDevice` endpoints.
 - The topology view is loaded lazily and kept internal to `app.inspect`; a
   pure-write workflow never triggers a read. Reads and hydration are internally
diff --git a/src/videoipath_automation_tool/apps/inspect/__init__.py b/src/videoipath_automation_tool/apps/inspect/__init__.py
index e591354..6bf4b9c 100644
--- a/src/videoipath_automation_tool/apps/inspect/__init__.py
+++ b/src/videoipath_automation_tool/apps/inspect/__init__.py
@@ -1,9 +1,9 @@
 from __future__ import annotations
 
+from videoipath_automation_tool.apps.inspect import maintenance as _maintenance
 from videoipath_automation_tool.apps.inspect import model as _model
+from videoipath_automation_tool.apps.inspect.app import InspectApp as InspectApp
 from videoipath_automation_tool.apps.inspect.app.actions import ConflictStrategy as ConflictStrategy
-from videoipath_automation_tool.apps.inspect.transaction import CommitResult as CommitResult
-from videoipath_automation_tool.apps.inspect.transaction import InspectTransaction as InspectTransaction
 from videoipath_automation_tool.apps.inspect.domain import InspectAlarm as InspectAlarm
 from videoipath_automation_tool.apps.inspect.domain import InspectDevice as InspectDevice
 from videoipath_automation_tool.apps.inspect.domain import InspectEdge as InspectEdge
@@ -14,38 +14,53 @@
 from videoipath_automation_tool.apps.inspect.domain import PortFromTemplate as PortFromTemplate
 from videoipath_automation_tool.apps.inspect.domain import VirtualDeviceSpec as VirtualDeviceSpec
 from videoipath_automation_tool.apps.inspect.domain import VirtualModuleSpec as VirtualModuleSpec
+from videoipath_automation_tool.apps.inspect.domain.maintenance import (
+    InspectMaintenanceBooking as InspectMaintenanceBooking,
+)
 from videoipath_automation_tool.apps.inspect.errors import InspectCommitConflictError as InspectCommitConflictError
 from videoipath_automation_tool.apps.inspect.errors import InspectCommitError as InspectCommitError
 from videoipath_automation_tool.apps.inspect.errors import InspectConflict as InspectConflict
 from videoipath_automation_tool.apps.inspect.errors import InspectEntityNotFoundError as InspectEntityNotFoundError
 from videoipath_automation_tool.apps.inspect.errors import InspectError as InspectError
+from videoipath_automation_tool.apps.inspect.errors import (
+    InspectMaintenanceConflictError as InspectMaintenanceConflictError,
+)
+from videoipath_automation_tool.apps.inspect.errors import InspectMaintenanceError as InspectMaintenanceError
 from videoipath_automation_tool.apps.inspect.errors import InspectQueryTooLongError as InspectQueryTooLongError
-from videoipath_automation_tool.apps.inspect.app import InspectApp as InspectApp
+from videoipath_automation_tool.apps.inspect.maintenance import *
 from videoipath_automation_tool.apps.inspect.model import *
+from videoipath_automation_tool.apps.inspect.transaction import CommitResult as CommitResult
+from videoipath_automation_tool.apps.inspect.transaction import InspectTransaction as InspectTransaction
 
 # InspectSnapshot is an internal implementation detail of InspectApp; it is not part of
 # the public API. Interact with the topology entirely through ``app.inspect``.
 
-__all__ = [
-    "CommitResult",
-    "ConflictStrategy",
-    "InspectAlarm",
-    "InspectApp",
-    "InspectCommitConflictError",
-    "InspectCommitError",
-    "InspectConflict",
-    "InspectDevice",
-    "InspectEdge",
-    "InspectEntityNotFoundError",
-    "InspectError",
-    "InspectModule",
-    "InspectPort",
-    "InspectPortTemplate",
-    "InspectQueryTooLongError",
-    "InspectService",
-    "InspectTransaction",
-    "PortFromTemplate",
-    "VirtualDeviceSpec",
-    "VirtualModuleSpec",
-    *_model.__all__,
-]
+__all__ = (
+    [
+        "CommitResult",
+        "InspectMaintenanceBooking",
+        "InspectMaintenanceError",
+        "InspectMaintenanceConflictError",
+        "ConflictStrategy",
+        "InspectAlarm",
+        "InspectApp",
+        "InspectCommitConflictError",
+        "InspectCommitError",
+        "InspectConflict",
+        "InspectDevice",
+        "InspectEdge",
+        "InspectEntityNotFoundError",
+        "InspectError",
+        "InspectModule",
+        "InspectPort",
+        "InspectPortTemplate",
+        "InspectQueryTooLongError",
+        "InspectService",
+        "InspectTransaction",
+        "PortFromTemplate",
+        "VirtualDeviceSpec",
+        "VirtualModuleSpec",
+    ]
+    + _maintenance.__all__
+    + _model.__all__
+)
diff --git a/src/videoipath_automation_tool/apps/inspect/api/inspect_api.py b/src/videoipath_automation_tool/apps/inspect/api/inspect_api.py
index 22b9ee1..c127326 100644
--- a/src/videoipath_automation_tool/apps/inspect/api/inspect_api.py
+++ b/src/videoipath_automation_tool/apps/inspect/api/inspect_api.py
@@ -59,9 +59,10 @@
 from videoipath_automation_tool.utils.cross_app_utils import create_fallback_logger
 
 from . import queries
+from .maintenance import InspectMaintenanceAPI
 
 
-class InspectAPI:
+class InspectAPI(InspectMaintenanceAPI):
     def __init__(self, vip_connector: VideoIPathConnector, logger: logging.Logger | None = None) -> None:
         self._logger = logger or create_fallback_logger("videoipath_automation_tool_inspect_api")
         self.vip_connector = vip_connector
diff --git a/src/videoipath_automation_tool/apps/inspect/api/maintenance.py b/src/videoipath_automation_tool/apps/inspect/api/maintenance.py
new file mode 100644
index 0000000..5abdfee
--- /dev/null
+++ b/src/videoipath_automation_tool/apps/inspect/api/maintenance.py
@@ -0,0 +1,101 @@
+"""Verified maintenance endpoints, kept separate from topology transactions."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Any
+
+from pydantic import TypeAdapter
+
+from ..errors import InspectMaintenanceError
+from ..model.common import InspectApiBaseModel
+from ..model.maintenance import (
+    InspectApiFetchMaintenanceImpactData,
+    InspectApiFetchMaintenanceImpactRequest,
+    InspectApiMaintenanceBookingItem,
+    InspectApiUpdateMaintenanceData,
+    InspectApiUpdateMaintenanceRequest,
+    InspectApiUpdateMaintenanceResponse,
+    InspectApiValidateMaintenanceData,
+    InspectApiValidateMaintenanceRequest,
+    MaintenanceImpact,
+    MaintenanceResult,
+)
+from . import queries
+
+if TYPE_CHECKING:
+    from videoipath_automation_tool.connector.vip_connector import VideoIPathConnector
+
+
+class InspectMaintenanceAPI:
+    vip_connector: VideoIPathConnector
+
+    def get_maintenance_section(self) -> list[InspectApiMaintenanceBookingItem]:
+        response = None
+        try:
+            response = self.vip_connector.rest.get(queries.maintenance_section(), allow_projection=True)
+            _check_header(response, "read", [])
+            collection = response.data["status"]["collector"]["maintenanceBookings"]
+            return [InspectApiMaintenanceBookingItem.model_validate(item) for item in collection.get("_items", [])]
+        except InspectMaintenanceError:
+            raise
+        except Exception as exc:  # The connector also raises untyped transport/envelope errors.
+            raise InspectMaintenanceError("read", [], str(exc), response) from exc
+
+    def update_maintenance(self, data: InspectApiUpdateMaintenanceData, *, operation: str) -> MaintenanceResult:
+        ids = [item.id for item in data.update] + data.delete
+        response = self._maintenance_post(
+            "updateMaintenance", InspectApiUpdateMaintenanceRequest(data=data), operation, ids
+        )
+        try:
+            parsed = InspectApiUpdateMaintenanceResponse.model_validate(
+                {"data": response.data, "header": response.header.model_dump(mode="json")}
+            )
+        except ValueError as exc:
+            raise InspectMaintenanceError(operation, ids, f"Invalid response: {exc}", response) from exc
+        if not parsed.data.result.ok:
+            raise InspectMaintenanceError(operation, ids, "; ".join(parsed.data.result.msg), parsed)
+        return parsed.data.result.model_copy(update={"details": parsed.data.details})
+
+    def validate_maintenance(self, data: InspectApiValidateMaintenanceData) -> dict[str, MaintenanceImpact]:
+        ids = [data.id] if data.id else []
+        response = self._maintenance_post(
+            "validateMaintenanceImpactDetailed", InspectApiValidateMaintenanceRequest(data=data), "validate", ids
+        )
+        return _impact_map(response, "validate", ids, _IMPACT)
+
+    def fetch_maintenance_impact(self, ids: list[str]) -> dict[str, dict[str, MaintenanceImpact]]:
+        response = self._maintenance_post(
+            "fetchMaintenanceImpact",
+            InspectApiFetchMaintenanceImpactRequest(data=InspectApiFetchMaintenanceImpactData(ids=ids)),
+            "impact",
+            ids,
+        )
+        return _impact_map(response, "impact", ids, _IMPACTS)
+
+    def _maintenance_post(self, action: str, body: InspectApiBaseModel, operation: str, ids: list[str]) -> Any:
+        response = None
+        try:
+            response = self.vip_connector.rest.post(f"/rest/v2/actions/status/pathman/{action}", body)
+            _check_header(response, operation, ids)
+            return response
+        except InspectMaintenanceError:
+            raise
+        except Exception as exc:
+            raise InspectMaintenanceError(operation, ids, str(exc), response) from exc
+
+
+def _check_header(response: Any, operation: str, ids: list[str]) -> None:
+    header = response.header.model_dump(mode="json")
+    if not header.get("ok") or not header.get("auth") or header.get("code") != "OK":
+        raise InspectMaintenanceError(operation, ids, "; ".join(header.get("msg", [])) or "Envelope rejected", response)
+
+
+def _impact_map(response: Any, operation: str, ids: list[str], adapter: TypeAdapter) -> Any:
+    try:
+        return adapter.validate_python(response.data)
+    except ValueError as exc:
+        raise InspectMaintenanceError(operation, ids, f"Invalid impact response: {exc}", response) from exc
+
+
+_IMPACT = TypeAdapter(dict[str, MaintenanceImpact])
+_IMPACTS = TypeAdapter(dict[str, dict[str, MaintenanceImpact]])
diff --git a/src/videoipath_automation_tool/apps/inspect/api/queries.py b/src/videoipath_automation_tool/apps/inspect/api/queries.py
index 544c595..107856b 100644
--- a/src/videoipath_automation_tool/apps/inspect/api/queries.py
+++ b/src/videoipath_automation_tool/apps/inspect/api/queries.py
@@ -65,6 +65,11 @@ def alarms_section() -> str:
     return _build(_ALARMS_SECTION)
 
 
+def maintenance_section() -> str:
+    """Maintenance bookings with their resolved resource contexts (verified 2026.2)."""
+    return _build("/status/collector/maintenanceBookings/**")
+
+
 def collector_full() -> str:
     """GET path for the full collector aggregate (eager / fallback mode)."""
     return _build(_COLLECTOR_FULL)
diff --git a/src/videoipath_automation_tool/apps/inspect/app/app.py b/src/videoipath_automation_tool/apps/inspect/app/app.py
index 77e9688..a19551a 100644
--- a/src/videoipath_automation_tool/apps/inspect/app/app.py
+++ b/src/videoipath_automation_tool/apps/inspect/app/app.py
@@ -1,6 +1,6 @@
 """InspectApp — the user-facing entry point for the VideoIPath Inspect surface.
 
-Read-only monitoring plus commit-style topology writes, built entirely on the collector API.
+Monitoring, commit-style topology writes, and immediate maintenance operations on the Inspect API surface.
 Composed from focused mixins, mirroring the Inventory/Topology app layout.
 
 This app is currently in beta; the API and behaviour may change in future releases.
@@ -10,7 +10,7 @@
 
 import logging
 import warnings
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING
 
 from videoipath_automation_tool.apps.inspect.api import InspectAPI
 from videoipath_automation_tool.connector.vip_connector import VideoIPathConnector
@@ -19,6 +19,7 @@
     from videoipath_automation_tool.apps.inspect.snapshot import InspectSnapshot
 
 from .actions import InspectActionsMixin
+from .maintenance import InspectMaintenanceMixin
 from .read import InspectReadMixin, LoadMode
 from .write import InspectWriteMixin
 
@@ -27,11 +28,11 @@
 )
 
 
-class InspectApp(InspectReadMixin, InspectWriteMixin, InspectActionsMixin):
+class InspectApp(InspectReadMixin, InspectWriteMixin, InspectActionsMixin, InspectMaintenanceMixin):
     def __init__(
         self,
         vip_connector: VideoIPathConnector,
-        logger: Optional[logging.Logger] = None,
+        logger: logging.Logger | None = None,
         load: LoadMode = "skeleton",
     ) -> None:
         """Inspect App (beta): read the topology/status and apply commit-style topology changes.
@@ -53,7 +54,7 @@ def __init__(
         self._inspect_api = InspectAPI(vip_connector=vip_connector, logger=self._logger)
         self._vip_connector = vip_connector
         self._load_mode: LoadMode = load
-        self._snapshot: Optional[InspectSnapshot] = None
+        self._snapshot: InspectSnapshot | None = None
         self._warn_beta()
         self._warn_if_version_unverified()
         self._logger.debug("Inspect APP initialized.")
@@ -76,7 +77,7 @@ def _warn_if_version_unverified(self) -> None:
 _MIN_VERIFIED_VERSION = (2025, 4)
 
 
-def _parse_version(version: str) -> Optional[tuple[int, int]]:
+def _parse_version(version: str) -> tuple[int, int] | None:
     parts = version.split(".")
     if len(parts) < 2:
         return None
diff --git a/src/videoipath_automation_tool/apps/inspect/app/maintenance.py b/src/videoipath_automation_tool/apps/inspect/app/maintenance.py
new file mode 100644
index 0000000..2653d37
--- /dev/null
+++ b/src/videoipath_automation_tool/apps/inspect/app/maintenance.py
@@ -0,0 +1,166 @@
+"""Immediate maintenance lifecycle methods; never part of a topology commit."""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Literal
+
+from ..errors import InspectEntityNotFoundError, InspectMaintenanceConflictError
+from ..maintenance import MaintenanceBookingSpec, MaintenanceOnceSchedule, MaintenanceRecurringSchedule
+from ..model.maintenance import (
+    InspectApiMaintenanceInheritable,
+    InspectApiMaintenanceUpdate,
+    InspectApiUpdateMaintenanceData,
+    InspectApiValidateMaintenanceData,
+    MaintenanceImpact,
+    MaintenanceResult,
+    MaintenanceState,
+)
+
+if TYPE_CHECKING:
+    from ..api import InspectAPI
+    from ..domain.maintenance import InspectMaintenanceBooking
+    from ..snapshot import InspectSnapshot
+
+
+class InspectMaintenanceMixin:
+    _inspect_api: InspectAPI
+    _snapshot: InspectSnapshot | None
+
+    @property
+    def maintenance_bookings(self) -> list[InspectMaintenanceBooking]:
+        """All server bookings (lazy snapshot section). Call refresh() for external changes."""
+        return self._get_snapshot().maintenance_bookings
+
+    def get_maintenance_booking(self, booking_id: str) -> InspectMaintenanceBooking | None:
+        return self._get_snapshot().get_maintenance_booking(booking_id)
+
+    def find_maintenance_bookings(
+        self, state: Literal["all", "active", "scheduled"] = "all", search: str = ""
+    ) -> list[InspectMaintenanceBooking]:
+        """Filter the snapshot by state and case-insensitive label/tag substring."""
+        if state not in ("all", "active", "scheduled"):
+            raise ValueError("state must be all, active, or scheduled.")
+        code = {"active": MaintenanceState.ACTIVE, "scheduled": MaintenanceState.SCHEDULED}.get(state)
+        needle = search.casefold()
+        return [
+            booking
+            for booking in self.maintenance_bookings
+            if (code is None or booking.state == code)
+            and (needle in booking.label.casefold() or any(needle in tag.casefold() for tag in booking.tags))
+        ]
+
+    def validate_maintenance_booking(
+        self, spec: MaintenanceBookingSpec, booking_id: str | None = None
+    ) -> dict[str, list[MaintenanceImpact]]:
+        """Read-only impacts keyed by service ID, with separate reports for each window."""
+        spec = _validated_spec(spec)
+        if booking_id is not None:
+            _ids([booking_id])
+            _require_once_update(spec)
+        result: dict[str, list[MaintenanceImpact]] = {}
+        for schedule in _windows(spec):
+            impacts = self._inspect_api.validate_maintenance(
+                InspectApiValidateMaintenanceData(
+                    pids=spec.targets.pids,
+                    edgeIds=spec.targets.edges,
+                    label=spec.label.strip(),
+                    scheduleInfo=schedule.to_wire(),
+                    id=booking_id,
+                )
+            )
+            for service_id, impact in impacts.items():
+                result.setdefault(service_id, []).append(impact)
+        return result
+
+    def get_maintenance_impact(self, booking_ids: list[str]) -> dict[str, dict[str, MaintenanceImpact]]:
+        """Current impact, keyed first by booking ID, then by affected service ID."""
+        return self._inspect_api.fetch_maintenance_impact(_ids(booking_ids))
+
+    def create_maintenance_booking(self, spec: MaintenanceBookingSpec) -> MaintenanceResult:
+        """Create immediately, expanding recurrence into one batch of dated bookings.
+
+        Preview separately. Result details carry server-returned IDs when available.
+        """
+        spec = _validated_spec(spec)
+        result = self._inspect_api.update_maintenance(
+            InspectApiUpdateMaintenanceData(create=[spec.to_create(window) for window in _windows(spec)]),
+            operation="create",
+        )
+        self._maintenance_written()
+        return result
+
+    def update_maintenance_booking(
+        self,
+        booking_id: str,
+        spec: MaintenanceBookingSpec,
+        *,
+        locked: bool | None = None,
+        expected_rev: str | None = None,
+    ) -> MaintenanceResult:
+        """Replace the specification using a fresh revision, optionally changing the lock.
+
+        Supply a one-time schedule for this explicit booking ID. Recurring creation
+        produces individual bookings; this method does not modify their siblings. For
+        start-now, supply a one-time schedule with start=None and the desired end.
+        Omitted locked preserves the fresh server value. No automatic write retry.
+        """
+        _ids([booking_id])
+        spec = _validated_spec(spec)
+        _require_once_update(spec)
+        current = next((b for b in self._inspect_api.get_maintenance_section() if b.id == booking_id), None)
+        if current is None:
+            raise InspectEntityNotFoundError(booking_id, "maintenance booking")
+        if expected_rev is not None and current.rev != expected_rev:
+            raise InspectMaintenanceConflictError(booking_id, expected_rev, current.rev)
+        result = self._inspect_api.update_maintenance(
+            InspectApiUpdateMaintenanceData(
+                update=[
+                    InspectApiMaintenanceUpdate(
+                        id=booking_id,
+                        rev=current.rev,
+                        inheritable=InspectApiMaintenanceInheritable(
+                            scheduleInfo=spec.schedule.to_wire(),
+                            locked=current.generic.locked if locked is None else locked,
+                        ),
+                        updater=spec.to_updater(),
+                    )
+                ]
+            ),
+            operation="update",
+        )
+        self._maintenance_written()
+        return result
+
+    def delete_maintenance_bookings(self, booking_ids: list[str]) -> MaintenanceResult:
+        """Cancel/delete explicit server IDs. The server decides whether locked IDs may be deleted."""
+        result = self._inspect_api.update_maintenance(
+            InspectApiUpdateMaintenanceData(delete=_ids(booking_ids)), operation="delete"
+        )
+        self._maintenance_written()
+        return result
+
+    def _maintenance_written(self) -> None:
+        if self._snapshot is not None:
+            self._snapshot.invalidate_maintenance()
+
+
+def _validated_spec(spec: MaintenanceBookingSpec) -> MaintenanceBookingSpec:
+    # Recheck nested mutable lists / model_copy(update=...) before issuing I/O.
+    return MaintenanceBookingSpec.model_validate(spec.model_dump())
+
+
+def _windows(spec: MaintenanceBookingSpec) -> list[MaintenanceOnceSchedule]:
+    return spec.schedule.expand() if isinstance(spec.schedule, MaintenanceRecurringSchedule) else [spec.schedule]
+
+
+def _require_once_update(spec: MaintenanceBookingSpec) -> None:
+    if spec.schedule.type != "once":
+        raise ValueError("Updating an explicit booking ID requires a one-time schedule; create recurrence separately.")
+
+
+def _ids(values: list[str]) -> list[str]:
+    if not isinstance(values, (list, tuple)) or not values:
+        raise ValueError("booking_ids must be a nonempty list of IDs.")
+    if any(not isinstance(v, str) or not v.strip() or v != v.strip() for v in values):
+        raise ValueError("booking IDs must be nonempty strings without surrounding whitespace.")
+    return list(dict.fromkeys(values))
diff --git a/src/videoipath_automation_tool/apps/inspect/domain/__init__.py b/src/videoipath_automation_tool/apps/inspect/domain/__init__.py
index 0087641..5404b05 100644
--- a/src/videoipath_automation_tool/apps/inspect/domain/__init__.py
+++ b/src/videoipath_automation_tool/apps/inspect/domain/__init__.py
@@ -3,6 +3,7 @@
 from videoipath_automation_tool.apps.inspect.domain.alarm import InspectAlarm
 from videoipath_automation_tool.apps.inspect.domain.device import InspectDevice, VirtualDeviceSpec
 from videoipath_automation_tool.apps.inspect.domain.edge import InspectEdge
+from videoipath_automation_tool.apps.inspect.domain.maintenance import InspectMaintenanceBooking
 from videoipath_automation_tool.apps.inspect.domain.module import InspectModule, VirtualModuleSpec
 from videoipath_automation_tool.apps.inspect.domain.port import InspectPort, InspectPortTemplate, PortFromTemplate
 from videoipath_automation_tool.apps.inspect.domain.service import InspectService
@@ -21,6 +22,7 @@
     "InspectEdge",
     "InspectGenericVertex",
     "InspectIpVertex",
+    "InspectMaintenanceBooking",
     "InspectModule",
     "InspectPort",
     "InspectPortTemplate",
diff --git a/src/videoipath_automation_tool/apps/inspect/domain/device.py b/src/videoipath_automation_tool/apps/inspect/domain/device.py
index 06a5468..77bf2dc 100644
--- a/src/videoipath_automation_tool/apps/inspect/domain/device.py
+++ b/src/videoipath_automation_tool/apps/inspect/domain/device.py
@@ -26,6 +26,7 @@
 if TYPE_CHECKING:
     from videoipath_automation_tool.apps.inspect.domain.alarm import InspectAlarm
     from videoipath_automation_tool.apps.inspect.domain.edge import InspectEdge
+    from videoipath_automation_tool.apps.inspect.domain.maintenance import InspectMaintenanceBooking
     from videoipath_automation_tool.apps.inspect.domain.module import InspectModule
     from videoipath_automation_tool.apps.inspect.domain.port import InspectPort
     from videoipath_automation_tool.apps.inspect.domain.service import InspectService
@@ -141,6 +142,11 @@ def status(self) -> InspectApiStatusSummary | None:
     def sync_severity(self) -> InspectSeverity | int | str | None:
         return self._record().node.syncSeverity
 
+    @property
+    def maintenance_bookings(self) -> list[InspectMaintenanceBooking]:
+        """Bookings selecting this resource or its descendants; no topology hydration."""
+        return self.snapshot.maintenance_for_resource("device", self.id)
+
     @property
     def alarms(self) -> list[InspectAlarm]:
         """Active alarms correlated to this device (worst severity first)."""
diff --git a/src/videoipath_automation_tool/apps/inspect/domain/edge.py b/src/videoipath_automation_tool/apps/inspect/domain/edge.py
index 6272f5c..0930ed8 100644
--- a/src/videoipath_automation_tool/apps/inspect/domain/edge.py
+++ b/src/videoipath_automation_tool/apps/inspect/domain/edge.py
@@ -16,6 +16,7 @@
 if TYPE_CHECKING:
     from videoipath_automation_tool.apps.inspect.domain.alarm import InspectAlarm
     from videoipath_automation_tool.apps.inspect.domain.device import InspectDevice
+    from videoipath_automation_tool.apps.inspect.domain.maintenance import InspectMaintenanceBooking
     from videoipath_automation_tool.apps.inspect.domain.port import InspectPort
     from videoipath_automation_tool.apps.inspect.domain.service import InspectService
     from videoipath_automation_tool.apps.inspect.model.actions import InspectApiEdgeForm
@@ -81,6 +82,11 @@ def status(self) -> InspectApiExternalEdgeLiveStatus | None:
         the per-edge status (per direction) appears in the full edge shape."""
         return self.indexed.edge.status or self.indexed.pair_status
 
+    @property
+    def maintenance_bookings(self) -> list[InspectMaintenanceBooking]:
+        """Bookings explicitly selecting this edge; no topology hydration."""
+        return self.snapshot.maintenance_for_resource("edge", self.id)
+
     @property
     def alarms(self) -> list[InspectAlarm]:
         """Active alarms correlated to this edge or its pair (worst severity first)."""
@@ -213,7 +219,7 @@ def bandwidth_capacity(self) -> float | int | None:
         return self._staged_or("bandwidth", lambda: self._form_get("bandwidth"))
 
     @bandwidth_capacity.setter
-    def bandwidth_capacity(self, value: float | int) -> None:
+    def bandwidth_capacity(self, value: float) -> None:
         self._stage("bandwidth", value)
 
     @property
diff --git a/src/videoipath_automation_tool/apps/inspect/domain/maintenance.py b/src/videoipath_automation_tool/apps/inspect/domain/maintenance.py
new file mode 100644
index 0000000..12b45c5
--- /dev/null
+++ b/src/videoipath_automation_tool/apps/inspect/domain/maintenance.py
@@ -0,0 +1,155 @@
+"""Snapshot-backed maintenance bookings and their resolved topology resources."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import TYPE_CHECKING
+
+from ..errors import InspectEntityNotFoundError
+from ..model.common import InspectFrozenModel, format_repr
+from ..model.maintenance import InspectApiMaintenanceBookingItem, InspectApiMaintenanceWindow, MaintenanceState
+from ..snapshot import InspectSnapshot
+
+if TYPE_CHECKING:
+    from .device import InspectDevice
+    from .edge import InspectEdge
+    from .module import InspectModule
+    from .port import InspectPort
+
+
+class InspectMaintenanceBooking(InspectFrozenModel):
+    """Read-only view; write through app.inspect's explicit maintenance methods.
+
+    ``raw`` retains all IDs and contexts, even when topology objects are absent.
+    Resource properties resolve explicitly selected resources, not expanded descendants.
+    """
+
+    snapshot: InspectSnapshot
+    id: str
+
+    @property
+    def raw(self) -> InspectApiMaintenanceBookingItem:
+        item = self.snapshot.get_maintenance_record(self.id)
+        if item is None:
+            raise InspectEntityNotFoundError(self.id, "maintenance booking")
+        return item
+
+    @property
+    def rev(self) -> str:
+        return self.raw.rev
+
+    @property
+    def label(self) -> str:
+        return self.raw.generic.descriptor.label
+
+    @property
+    def description(self) -> str:
+        return self.raw.generic.descriptor.desc
+
+    @property
+    def tags(self) -> list[str]:
+        return list(self.raw.tags)
+
+    @property
+    def state(self) -> MaintenanceState | int:
+        return self.raw.generic.state
+
+    @property
+    def locked(self) -> bool:
+        return self.raw.generic.locked
+
+    @property
+    def schedule(self) -> InspectApiMaintenanceWindow:
+        """Resolved window, not the recurrence rule originally submitted."""
+        return self.raw.scheduleInfo
+
+    @property
+    def starts_at(self) -> datetime:
+        return datetime.fromtimestamp(self.schedule.startTimestamp / 1000, UTC)
+
+    @property
+    def ends_at(self) -> datetime | None:
+        return None if self.schedule.infinite else datetime.fromtimestamp(self.schedule.endTimestamp / 1000, UTC)
+
+    @property
+    def action(self) -> str:
+        return self.raw.action
+
+    @property
+    def trigger(self) -> str:
+        return self.raw.trigger
+
+    @property
+    def allow_overlap(self) -> bool:
+        return self.raw.allowOverlap
+
+    @property
+    def switch_format_state_on_reroute(self) -> bool:
+        return self.raw.switchFormatStateOnReroute
+
+    @property
+    def device_ids(self) -> list[str]:
+        return [r.context.devicePid for r in self.raw.devices if r.context.devicePid]
+
+    @property
+    def module_ids(self) -> list[str]:
+        return [r.context.modulePid for r in self.raw.modules if r.context.modulePid]
+
+    @property
+    def port_ids(self) -> list[str]:
+        return [r.context.portPid for r in self.raw.ports if r.context.portPid]
+
+    @property
+    def edge_ids(self) -> list[str]:
+        return [r.id for r in self.raw.edges]
+
+    @property
+    def devices(self) -> list[InspectDevice]:
+        return [
+            d
+            for pid in self.device_ids
+            if (d := self.snapshot.get_device(self.snapshot._resolve_device_id(pid))) is not None
+        ]
+
+    @property
+    def modules(self) -> list[InspectModule]:
+        result = []
+        for resource in self.raw.modules:
+            context = resource.context
+            device_id = self.snapshot._resolve_device_id(context.devicePid)
+            if device_id and context.modulePid:
+                for module in self.snapshot.get_modules_for_device(device_id):
+                    status = module._status()
+                    pid = (
+                        (status.context.modulePid if status and status.context else None)
+                        or (status.pid if status else None)
+                        or module.id
+                    )
+                    if pid == context.modulePid:
+                        result.append(module)
+        return result
+
+    @property
+    def ports(self) -> list[InspectPort]:
+        result = []
+        for resource in self.raw.ports:
+            context = resource.context
+            device_id = self.snapshot._resolve_device_id(context.devicePid)
+            if device_id and context.portPid:
+                port = self.snapshot.get_port(device_id, context.portPid)
+                if port is not None:
+                    result.append(port)
+        return result
+
+    @property
+    def edges(self) -> list[InspectEdge]:
+        ids = set(self.edge_ids)
+        return [edge for edge in self.snapshot.edges if edge.id in ids]
+
+    def __repr__(self) -> str:
+        return format_repr(self, id=self.id, label=lambda: self.label)
+
+    __str__ = __repr__
+
+
+__all__ = ["InspectMaintenanceBooking"]
diff --git a/src/videoipath_automation_tool/apps/inspect/domain/module.py b/src/videoipath_automation_tool/apps/inspect/domain/module.py
index fc9fd09..9336096 100644
--- a/src/videoipath_automation_tool/apps/inspect/domain/module.py
+++ b/src/videoipath_automation_tool/apps/inspect/domain/module.py
@@ -12,11 +12,12 @@
     format_repr,
 )
 from videoipath_automation_tool.apps.inspect.model.virtual import InspectApiVirtualModule
-from videoipath_automation_tool.apps.inspect.snapshot import InspectSnapshot, _STAGED_MISSING
+from videoipath_automation_tool.apps.inspect.snapshot import _STAGED_MISSING, InspectSnapshot
 
 if TYPE_CHECKING:
     from videoipath_automation_tool.apps.inspect.domain.alarm import InspectAlarm
     from videoipath_automation_tool.apps.inspect.domain.device import InspectDevice
+    from videoipath_automation_tool.apps.inspect.domain.maintenance import InspectMaintenanceBooking
     from videoipath_automation_tool.apps.inspect.domain.port import InspectPort
     from videoipath_automation_tool.apps.inspect.domain.vertex import InspectVertex
     from videoipath_automation_tool.apps.inspect.model.collector import InspectApiModuleStatus
@@ -82,6 +83,13 @@ def status(self) -> InspectApiStatusSummary | None:
         status = self._status()
         return status.status if status is not None else None
 
+    @property
+    def maintenance_bookings(self) -> list[InspectMaintenanceBooking]:
+        """Bookings selecting this resource or its descendants; no topology hydration."""
+        status = self._status()
+        pid = (status.context.modulePid if status and status.context else None) or (status.pid if status else None)
+        return self.snapshot.maintenance_for_resource("module", pid or self.id)
+
     @property
     def alarms(self) -> list[InspectAlarm]:
         """Active alarms correlated to this module (worst severity first)."""
diff --git a/src/videoipath_automation_tool/apps/inspect/domain/port.py b/src/videoipath_automation_tool/apps/inspect/domain/port.py
index 09668e9..318ff71 100644
--- a/src/videoipath_automation_tool/apps/inspect/domain/port.py
+++ b/src/videoipath_automation_tool/apps/inspect/domain/port.py
@@ -30,6 +30,7 @@
     from videoipath_automation_tool.apps.inspect.domain.alarm import InspectAlarm
     from videoipath_automation_tool.apps.inspect.domain.device import InspectDevice
     from videoipath_automation_tool.apps.inspect.domain.edge import InspectEdge
+    from videoipath_automation_tool.apps.inspect.domain.maintenance import InspectMaintenanceBooking
     from videoipath_automation_tool.apps.inspect.domain.module import InspectModule
     from videoipath_automation_tool.apps.inspect.domain.vertex import InspectVertex
     from videoipath_automation_tool.apps.inspect.model.common import InspectApiStatusSummary
@@ -140,6 +141,11 @@ def description(self) -> str | None:
     def status(self) -> InspectApiStatusSummary | None:
         return self.indexed.port.status
 
+    @property
+    def maintenance_bookings(self) -> list[InspectMaintenanceBooking]:
+        """Bookings selecting this port, including selected edge endpoints."""
+        return self.snapshot.maintenance_for_resource("port", self.id)
+
     @property
     def alarms(self) -> list[InspectAlarm]:
         """Active alarms correlated to this port (worst severity first)."""
diff --git a/src/videoipath_automation_tool/apps/inspect/errors.py b/src/videoipath_automation_tool/apps/inspect/errors.py
index 53f51a3..060e8ad 100644
--- a/src/videoipath_automation_tool/apps/inspect/errors.py
+++ b/src/videoipath_automation_tool/apps/inspect/errors.py
@@ -30,6 +30,30 @@ def __init__(self, entity_id: str, kind: str = "entity") -> None:
         super().__init__(f"Inspect {kind} '{entity_id}' was not found on the server.")
 
 
+class InspectMaintenanceError(InspectError):
+    """Maintenance transport, protocol, or server failure. Writes are never retried.
+
+    ``response`` retains server details when available. A transport failure may
+    have happened after the write; re-read before deciding whether to retry.
+    """
+
+    def __init__(self, operation: str, booking_ids: list[str], detail: str, response: object = None) -> None:
+        self.operation = operation
+        self.booking_ids = booking_ids
+        self.detail = detail
+        self.response = response
+        super().__init__(f"Inspect maintenance {operation} failed: {detail}")
+
+
+class InspectMaintenanceConflictError(InspectMaintenanceError):
+    """The caller's expected revision differs from the freshly read booking."""
+
+    def __init__(self, booking_id: str, expected_rev: str, actual_rev: str) -> None:
+        self.expected_rev = expected_rev
+        self.actual_rev = actual_rev
+        super().__init__("update", [booking_id], f"revision changed from {expected_rev!r} to {actual_rev!r}")
+
+
 class InspectQueryTooLongError(InspectError):
     """A scoped collector query URL exceeds the server/proxy URI length limit (HTTP 414).
 
@@ -54,7 +78,7 @@ class InspectCommitError(InspectError):
     typed response so callers can inspect ``validation.details`` and ``res.msg``.
     """
 
-    def __init__(self, response: "InspectApiUpdateTopologyResponse") -> None:
+    def __init__(self, response: InspectApiUpdateTopologyResponse) -> None:
         self.response = response
         self.result = response.data.res
         self.validation = response.data.validation
@@ -94,10 +118,12 @@ def __init__(self, conflicts: list[InspectConflict]) -> None:
 
 
 __all__ = [
-    "InspectError",
-    "InspectEntityNotFoundError",
-    "InspectQueryTooLongError",
+    "InspectCommitConflictError",
     "InspectCommitError",
     "InspectConflict",
-    "InspectCommitConflictError",
+    "InspectEntityNotFoundError",
+    "InspectError",
+    "InspectMaintenanceConflictError",
+    "InspectMaintenanceError",
+    "InspectQueryTooLongError",
 ]
diff --git a/src/videoipath_automation_tool/apps/inspect/maintenance.py b/src/videoipath_automation_tool/apps/inspect/maintenance.py
new file mode 100644
index 0000000..5209c4f
--- /dev/null
+++ b/src/videoipath_automation_tool/apps/inspect/maintenance.py
@@ -0,0 +1,259 @@
+"""User-facing maintenance specifications. Writes require an explicit schedule."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, time, timedelta
+from typing import Annotated, Any, Literal, Self
+from zoneinfo import ZoneInfo, ZoneInfoNotFoundError
+
+from pydantic import AwareDatetime, ConfigDict, Field, ValidationInfo, field_validator, model_validator
+
+from .model.common import InspectApiDescriptor, InspectFrozenModel
+from .model.maintenance import (
+    InspectApiMaintenanceCreate,
+    InspectApiMaintenanceCreateDefinition,
+    InspectApiMaintenanceInstance,
+    InspectApiMaintenanceOnce,
+    InspectApiMaintenancePattern,
+    InspectApiMaintenanceRecurring,
+    InspectApiMaintenanceUpdater,
+    MaintenanceAction,
+    MaintenanceTrigger,
+)
+
+
+class MaintenanceOnceSchedule(InspectFrozenModel):
+    """One window: ``start=None`` means now, ``end=None`` means never ends."""
+
+    model_config = ConfigDict(extra="forbid")
+    type: Literal["once"] = "once"
+    start: AwareDatetime | None = None
+    end: AwareDatetime | None = None
+
+    def to_wire(self) -> InspectApiMaintenanceOnce:
+        return InspectApiMaintenanceOnce(startTimestamp=_milliseconds(self.start), endTimestamp=_milliseconds(self.end))
+
+    @model_validator(mode="after")
+    def _ordered(self) -> Self:
+        if self.start is not None and self.end is not None and self.end.astimezone(UTC) <= self.start.astimezone(UTC):
+            raise ValueError("end must be after start.")
+        return self
+
+
+class MaintenanceRecurringSchedule(InspectFrozenModel):
+    """A daily/weekly/monthly rule expanded into dated one-time bookings by the SDK.
+
+    Weekdays use ISO numbering (Monday=1). Monthly rules use the start date's day.
+    Local times use ``timezone``. Nonexistent DST times are rejected; ambiguous
+    times use the first occurrence. Missing monthly days are skipped. Only complete
+    windows inside [start, end] are included, at most 1000 per request. Raw iteration
+    filters can be serialized by to_wire(), but cannot be expanded by this SDK.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+    type: Literal["recurring"] = "recurring"
+    frequency: Literal["daily", "weekly", "monthly"]
+    start: AwareDatetime
+    end: AwareDatetime
+    timezone: str
+    local_start: time
+    local_end: time
+    weekdays: list[int] = Field(default_factory=list)
+    pattern_iteration_filter: list[int] = Field(default_factory=list)
+    instance_iteration_filter: list[int] = Field(default_factory=list)
+
+    def expand(self) -> list[MaintenanceOnceSchedule]:
+        """Expand the finite rule before any I/O; no local scheduler is installed."""
+        if self.pattern_iteration_filter or self.instance_iteration_filter:
+            raise ValueError("Client expansion does not support server-specific iteration filters.")
+        zone = ZoneInfo(self.timezone)
+        date = self.start.astimezone(zone).date()
+        last = self.end.astimezone(zone).date()
+        anchor_day = date.day
+        windows: list[MaintenanceOnceSchedule] = []
+        while date <= last:
+            matches = (
+                self.frequency == "daily"
+                or self.frequency == "weekly"
+                and date.isoweekday() in self.weekdays
+                or self.frequency == "monthly"
+                and date.day == anchor_day
+            )
+            if matches:
+                start = _local_datetime(datetime.combine(date, self.local_start), zone)
+                end_date = date + timedelta(days=self.local_end < self.local_start)
+                end = _local_datetime(datetime.combine(end_date, self.local_end), zone)
+                if start >= self.start and end <= self.end:
+                    windows.append(MaintenanceOnceSchedule(start=start, end=end))
+                    if len(windows) > 1000:
+                        raise ValueError("Recurrence exceeds 1000 windows; use a shorter recurrence range.")
+            if date == last:
+                break
+            date += timedelta(days=1)
+        if not windows:
+            raise ValueError("Recurrence range contains no complete maintenance windows.")
+        return windows
+
+    def to_wire(self) -> InspectApiMaintenanceRecurring:
+        return InspectApiMaintenanceRecurring(
+            pattern=InspectApiMaintenancePattern(
+                pattern={"daily": 0, "weekly": 1, "monthly": 2}[self.frequency],
+                startTime=_milliseconds(self.start),
+                endTime=_milliseconds(self.end),
+                timeZoneId=self.timezone,
+                weekDays=self.weekdays,
+                iterationFilter=self.pattern_iteration_filter,
+            ),
+            instance=InspectApiMaintenanceInstance(
+                localStartTime=_seconds(self.local_start),
+                localEndTime=_seconds(self.local_end),
+                iterationFilter=self.instance_iteration_filter,
+            ),
+        )
+
+    @model_validator(mode="after")
+    def _valid_rule(self) -> Self:
+        if self.end.astimezone(UTC) <= self.start.astimezone(UTC):
+            raise ValueError("end must be after start.")
+        try:
+            ZoneInfo(self.timezone)
+        except (ZoneInfoNotFoundError, ValueError) as exc:
+            raise ValueError("timezone must be a valid IANA timezone.") from exc
+        if any(t.tzinfo is not None or t.microsecond for t in (self.local_start, self.local_end)):
+            raise ValueError("local times must have whole seconds and no timezone; use the timezone field.")
+        if self.local_start == self.local_end:
+            raise ValueError("local_start and local_end must differ.")
+        if any(day not in range(1, 8) for day in self.weekdays) or len(set(self.weekdays)) != len(self.weekdays):
+            raise ValueError("weekdays must contain unique ISO weekday numbers (1–7).")
+        if self.frequency == "weekly" and not self.weekdays:
+            raise ValueError("weekly recurrence requires weekdays.")
+        if self.frequency != "weekly" and self.weekdays:
+            raise ValueError("weekdays apply only to weekly recurrence.")
+        return self
+
+
+MaintenanceSchedule = Annotated[MaintenanceOnceSchedule | MaintenanceRecurringSchedule, Field(discriminator="type")]
+
+
+class MaintenanceTargets(InspectFrozenModel):
+    """Canonical PIDs/edge IDs, or matching Inspect objects, selected for maintenance.
+
+    String module/port IDs must be globally qualified PIDs, not display labels or
+    module-local keys. Device objects handle the virtual device ID/PID distinction.
+    """
+
+    model_config = ConfigDict(extra="forbid")
+    devices: list[str] = Field(default_factory=list)
+    modules: list[str] = Field(default_factory=list)
+    ports: list[str] = Field(default_factory=list)
+    edges: list[str] = Field(default_factory=list)
+
+    @property
+    def pids(self) -> list[str]:
+        return list(dict.fromkeys([*self.devices, *self.modules, *self.ports]))
+
+    @field_validator("devices", "modules", "ports", "edges", mode="before")
+    @classmethod
+    def _normalize(cls, values: Any, info: ValidationInfo) -> list[str]:
+        from .domain.device import InspectDevice
+        from .domain.edge import InspectEdge
+        from .domain.module import InspectModule
+        from .domain.port import InspectPort
+
+        kinds = {"devices": InspectDevice, "modules": InspectModule, "ports": InspectPort, "edges": InspectEdge}
+        if not isinstance(values, (list, tuple)):
+            raise ValueError("targets must be a list or tuple.")  # noqa: TRY004 — Pydantic validation error
+        ids: list[str] = []
+        for value in values:
+            if isinstance(value, str):
+                pid = value
+            elif not isinstance(value, kinds[info.field_name]):
+                raise ValueError(f"{info.field_name} requires IDs or matching Inspect objects.")  # noqa: TRY004
+            elif isinstance(value, InspectDevice):
+                record = value.snapshot.get_device_record(value.id)
+                pid = (
+                    (record.node.context.devicePid if record and record.node.context else None)
+                    or (record.pid if record else None)
+                    or value.id
+                )
+            elif isinstance(value, InspectModule):
+                status = value._status()
+                pid = (
+                    (status.context.modulePid if status and status.context else None)
+                    or (status.pid if status else None)
+                    or value.id
+                )
+            else:
+                pid = value.id
+            if not pid or not pid.strip() or pid != pid.strip():
+                raise ValueError("Every maintenance target must have a nonempty canonical identifier.")
+            ids.append(pid)
+        return list(dict.fromkeys(ids))
+
+
+class MaintenanceBookingSpec(InspectFrozenModel):
+    """Complete create/update intent. Previewing and writing are separate calls."""
+
+    model_config = ConfigDict(extra="forbid")
+    label: str
+    description: str = ""
+    tags: list[str] = Field(default_factory=list)
+    targets: MaintenanceTargets
+    schedule: MaintenanceSchedule
+    action: MaintenanceAction = "nothing"
+    trigger: MaintenanceTrigger = "create"
+    allow_overlap: bool = False
+    switch_format_state_on_reroute: bool = False
+
+    def to_create(self, schedule: MaintenanceOnceSchedule | None = None) -> InspectApiMaintenanceCreate:
+        return InspectApiMaintenanceCreate(
+            scheduleInfo=(schedule or self.schedule).to_wire(),
+            serviceDefinition=InspectApiMaintenanceCreateDefinition(
+                **self.to_updater().model_dump(exclude={"pids"}), devicePids=self.targets.pids
+            ),
+        )
+
+    def to_updater(self) -> InspectApiMaintenanceUpdater:
+        return InspectApiMaintenanceUpdater(
+            descriptor=InspectApiDescriptor(label=self.label.strip(), desc=self.description.strip()),
+            pids=self.targets.pids,
+            edgeIds=self.targets.edges,
+            tags=self.tags,
+            action=self.action,
+            trigger=self.trigger,
+            allowOverlap=self.allow_overlap,
+            switchFormatStateOnReroute=self.switch_format_state_on_reroute,
+        )
+
+    @model_validator(mode="after")
+    def _required(self) -> Self:
+        if not self.label.strip():
+            raise ValueError("label must not be empty.")
+        if not self.targets.pids and not self.targets.edges:
+            raise ValueError("Select at least one maintenance target.")
+        return self
+
+
+def _milliseconds(value: datetime | None) -> int | None:
+    return int(value.timestamp() * 1000) if value is not None else None
+
+
+def _seconds(value: time) -> int:
+    return value.hour * 3600 + value.minute * 60 + value.second
+
+
+def _local_datetime(value: datetime, zone: ZoneInfo) -> datetime:
+    aware = value.replace(tzinfo=zone, fold=0)
+    utc = aware.astimezone(UTC)
+    if utc.astimezone(zone).replace(tzinfo=None) != value:
+        raise ValueError(f"Nonexistent local maintenance time {value.isoformat()} in {zone.key}.")
+    return utc
+
+
+__all__ = [
+    "MaintenanceBookingSpec",
+    "MaintenanceOnceSchedule",
+    "MaintenanceRecurringSchedule",
+    "MaintenanceSchedule",
+    "MaintenanceTargets",
+]
diff --git a/src/videoipath_automation_tool/apps/inspect/model/__init__.py b/src/videoipath_automation_tool/apps/inspect/model/__init__.py
index bed24fe..8eb82a5 100644
--- a/src/videoipath_automation_tool/apps/inspect/model/__init__.py
+++ b/src/videoipath_automation_tool/apps/inspect/model/__init__.py
@@ -5,6 +5,7 @@
     alarms,
     collector,
     common,
+    maintenance,
     ngraph,
     tags,
     update_topology,
@@ -14,18 +15,24 @@
 from videoipath_automation_tool.apps.inspect.model.alarms import *
 from videoipath_automation_tool.apps.inspect.model.collector import *
 from videoipath_automation_tool.apps.inspect.model.common import *
+from videoipath_automation_tool.apps.inspect.model.maintenance import *
 from videoipath_automation_tool.apps.inspect.model.ngraph import *
 from videoipath_automation_tool.apps.inspect.model.tags import *
 from videoipath_automation_tool.apps.inspect.model.update_topology import *
 from videoipath_automation_tool.apps.inspect.model.virtual import *
 
-__all__ = [
-    *actions.__all__,
-    *alarms.__all__,
-    *collector.__all__,
-    *common.__all__,
-    *ngraph.__all__,
-    *tags.__all__,
-    *update_topology.__all__,
-    *virtual.__all__,
-]
+__all__ = list(
+    dict.fromkeys(
+        [
+            *actions.__all__,
+            *alarms.__all__,
+            *collector.__all__,
+            *common.__all__,
+            *maintenance.__all__,
+            *ngraph.__all__,
+            *tags.__all__,
+            *update_topology.__all__,
+            *virtual.__all__,
+        ]
+    )
+)
diff --git a/src/videoipath_automation_tool/apps/inspect/model/collector.py b/src/videoipath_automation_tool/apps/inspect/model/collector.py
index 4a0ac13..82afd55 100644
--- a/src/videoipath_automation_tool/apps/inspect/model/collector.py
+++ b/src/videoipath_automation_tool/apps/inspect/model/collector.py
@@ -21,6 +21,8 @@
     map_severity,
 )
 
+from .maintenance import InspectApiMaintenanceBookingItem
+
 
 class InspectApiGenericServiceFields(InspectApiBaseModel):
     allocationState: int | None = None
@@ -320,11 +322,6 @@ class InspectApiExternalEdgesByDeviceKeyItem(InspectApiBaseModel):
     status: InspectApiExternalEdgeLiveStatus | None = None
 
 
-class InspectApiMaintenanceBookingItem(InspectApiBaseModel):
-    id: str = Field(alias="_id")
-    vid: str | None = Field(default=None, alias="_vid")
-
-
 class InspectApiSuperProfileItem(InspectApiBaseModel):
     id: str = Field(alias="_id")
     vid: str | None = Field(default=None, alias="_vid")
diff --git a/src/videoipath_automation_tool/apps/inspect/model/maintenance.py b/src/videoipath_automation_tool/apps/inspect/model/maintenance.py
new file mode 100644
index 0000000..29c7944
--- /dev/null
+++ b/src/videoipath_automation_tool/apps/inspect/model/maintenance.py
@@ -0,0 +1,265 @@
+"""Maintenance transport models (VideoIPath 2026.2).
+
+Collector schedules describe resolved windows; action schedules describe intent.
+They deliberately use different types so a recurring rule cannot be lost on edit.
+"""
+
+from __future__ import annotations
+
+from enum import IntEnum
+from typing import Annotated, Any, Literal
+
+from pydantic import Field, field_validator
+
+from .common import (
+    InspectApiBaseModel,
+    InspectApiDescriptor,
+    InspectApiPostRequestHeader,
+    InspectApiRestV2Header,
+    InspectApiSimpleActionResult,
+    InspectApiStatusContext,
+)
+
+MaintenanceAction = Literal["nothing", "invalidate", "reroute", "rerouteSA"]
+MaintenanceTrigger = Literal["create", "active"]
+
+
+class MaintenanceState(IntEnum):
+    SCHEDULED = 0
+    ACTIVE = 1
+
+
+class InspectApiMaintenanceWindow(InspectApiBaseModel):
+    startTimestamp: int
+    endTimestamp: int
+    infinite: bool = False
+
+
+class InspectApiMaintenanceResource(InspectApiBaseModel):
+    label: str = ""
+    context: InspectApiStatusContext = Field(default_factory=InspectApiStatusContext)
+    pid: list[str] = Field(default_factory=list)
+    deviceId: str | None = None
+
+
+class InspectApiMaintenanceEdge(InspectApiBaseModel):
+    id: str
+    from_: InspectApiMaintenanceResource = Field(alias="from")
+    to: InspectApiMaintenanceResource
+
+
+class InspectApiMaintenanceGeneric(InspectApiBaseModel):
+    descriptor: InspectApiDescriptor = Field(default_factory=InspectApiDescriptor)
+    locked: bool = False
+    state: MaintenanceState | int
+
+    @field_validator("state", mode="before")
+    @classmethod
+    def _state(cls, value: Any) -> Any:
+        return MaintenanceState(value) if type(value) is int and value in (0, 1) else value
+
+
+class InspectApiMaintenanceBookingItem(InspectApiBaseModel):
+    id: str = Field(alias="_id")
+    vid: str | None = Field(default=None, alias="_vid")
+    rev: str
+    generic: InspectApiMaintenanceGeneric
+    scheduleInfo: InspectApiMaintenanceWindow
+    action: MaintenanceAction | str = "nothing"
+    trigger: MaintenanceTrigger | str = "create"
+    allowOverlap: bool = False
+    switchFormatStateOnReroute: bool = False
+    tags: list[str] = Field(default_factory=list)
+    devices: list[InspectApiMaintenanceResource] = Field(default_factory=list)
+    modules: list[InspectApiMaintenanceResource] = Field(default_factory=list)
+    ports: list[InspectApiMaintenanceResource] = Field(default_factory=list)
+    edges: list[InspectApiMaintenanceEdge] = Field(default_factory=list)
+
+
+class InspectApiMaintenanceOnce(InspectApiBaseModel):
+    type: Literal["once"] = "once"
+    startTimestamp: int | None = None
+    endTimestamp: int | None = None
+
+
+class InspectApiMaintenancePattern(InspectApiBaseModel):
+    pattern: Literal[0, 1, 2]
+    startTime: int
+    endTime: int
+    timeZoneId: str
+    weekDays: list[int] = Field(default_factory=list)
+    iterationFilter: list[int] = Field(default_factory=list)
+
+
+class InspectApiMaintenanceInstance(InspectApiBaseModel):
+    localStartTime: int
+    localEndTime: int
+    iterationFilter: list[int] = Field(default_factory=list)
+
+
+class InspectApiMaintenanceRecurring(InspectApiBaseModel):
+    type: Literal["recurring"] = "recurring"
+    pattern: InspectApiMaintenancePattern
+    instance: InspectApiMaintenanceInstance
+
+
+InspectApiMaintenanceSchedule = Annotated[
+    InspectApiMaintenanceOnce | InspectApiMaintenanceRecurring, Field(discriminator="type")
+]
+
+
+class InspectApiMaintenanceDefinition(InspectApiBaseModel):
+    descriptor: InspectApiDescriptor
+    action: MaintenanceAction = "nothing"
+    trigger: MaintenanceTrigger = "create"
+    allowOverlap: bool = False
+    switchFormatStateOnReroute: bool = False
+    tags: list[str] = Field(default_factory=list)
+    edgeIds: list[str] = Field(default_factory=list)
+
+
+class InspectApiMaintenanceCreateDefinition(InspectApiMaintenanceDefinition):
+    devicePids: list[str] = Field(default_factory=list)
+
+
+class InspectApiMaintenanceUpdater(InspectApiMaintenanceDefinition):
+    pids: list[str] = Field(default_factory=list)
+
+
+class InspectApiMaintenanceCreate(InspectApiBaseModel):
+    scheduleInfo: InspectApiMaintenanceSchedule
+    serviceDefinition: InspectApiMaintenanceCreateDefinition
+
+
+class InspectApiMaintenanceInheritable(InspectApiBaseModel):
+    scheduleInfo: InspectApiMaintenanceSchedule
+    locked: bool
+
+
+class InspectApiMaintenanceUpdate(InspectApiBaseModel):
+    id: str
+    rev: str
+    inheritable: InspectApiMaintenanceInheritable
+    updater: InspectApiMaintenanceUpdater
+
+
+class InspectApiUpdateMaintenanceData(InspectApiBaseModel):
+    create: list[InspectApiMaintenanceCreate] = Field(default_factory=list)
+    update: list[InspectApiMaintenanceUpdate] = Field(default_factory=list)
+    delete: list[str] = Field(default_factory=list)
+
+
+class InspectApiUpdateMaintenanceRequest(InspectApiBaseModel):
+    header: InspectApiPostRequestHeader = Field(default_factory=InspectApiPostRequestHeader)
+    data: InspectApiUpdateMaintenanceData
+
+
+class MaintenanceChange(InspectApiBaseModel):
+    """Server-confirmed booking change; its dictionary key is the authoritative ID."""
+
+    rev: str
+    isCancel: bool
+    status: int
+    type: str
+    resolvable: bool | None = None
+
+
+class MaintenanceResult(InspectApiSimpleActionResult):
+    """Server result and optional per-booking details; never IDs parsed from messages."""
+
+    details: dict[str, MaintenanceChange] = Field(default_factory=dict)
+
+
+class InspectApiUpdateMaintenanceResult(InspectApiBaseModel):
+    result: MaintenanceResult
+    details: dict[str, MaintenanceChange] = Field(default_factory=dict)
+
+
+class InspectApiUpdateMaintenanceResponse(InspectApiBaseModel):
+    header: InspectApiRestV2Header
+    data: InspectApiUpdateMaintenanceResult
+
+
+class InspectApiValidateMaintenanceData(InspectApiBaseModel):
+    edgeIds: list[str] = Field(default_factory=list)
+    pids: list[str] = Field(default_factory=list)
+    scheduleInfo: InspectApiMaintenanceSchedule
+    label: str
+    id: str | None = None
+
+
+class InspectApiValidateMaintenanceRequest(InspectApiBaseModel):
+    header: InspectApiPostRequestHeader = Field(default_factory=InspectApiPostRequestHeader)
+    data: InspectApiValidateMaintenanceData
+
+
+class InspectApiFetchMaintenanceImpactData(InspectApiBaseModel):
+    ids: list[str]
+
+
+class InspectApiFetchMaintenanceImpactRequest(InspectApiBaseModel):
+    header: InspectApiPostRequestHeader = Field(default_factory=InspectApiPostRequestHeader)
+    data: InspectApiFetchMaintenanceImpactData
+
+
+class MaintenanceImpactContributor(InspectApiBaseModel):
+    label: str
+
+
+class MaintenanceImpactEntry(InspectApiBaseModel):
+    start: int
+    end: int
+    saBefore: int | str
+    saAfter: int | str
+    contributors: dict[str, MaintenanceImpactContributor] = Field(default_factory=dict)
+
+
+class MaintenanceImpact(InspectApiBaseModel):
+    """One affected service; maps returned by the API are keyed by service ID.
+
+    Timestamps are raw epoch milliseconds, including the server's infinite sentinel.
+    """
+
+    label: str
+    start: int
+    end: int
+    protection: str
+    maxSaBefore: int | str
+    maxSaAfter: int | str
+    entries: list[MaintenanceImpactEntry] = Field(default_factory=list)
+
+
+__all__ = [
+    "InspectApiFetchMaintenanceImpactData",
+    "InspectApiFetchMaintenanceImpactRequest",
+    "InspectApiMaintenanceBookingItem",
+    "InspectApiMaintenanceCreate",
+    "InspectApiMaintenanceCreateDefinition",
+    "InspectApiMaintenanceDefinition",
+    "InspectApiMaintenanceEdge",
+    "InspectApiMaintenanceGeneric",
+    "InspectApiMaintenanceInheritable",
+    "InspectApiMaintenanceInstance",
+    "InspectApiMaintenanceOnce",
+    "InspectApiMaintenancePattern",
+    "InspectApiMaintenanceRecurring",
+    "InspectApiMaintenanceResource",
+    "InspectApiMaintenanceSchedule",
+    "InspectApiMaintenanceUpdate",
+    "InspectApiMaintenanceUpdater",
+    "InspectApiMaintenanceWindow",
+    "InspectApiUpdateMaintenanceData",
+    "InspectApiUpdateMaintenanceRequest",
+    "InspectApiUpdateMaintenanceResponse",
+    "InspectApiUpdateMaintenanceResult",
+    "InspectApiValidateMaintenanceData",
+    "InspectApiValidateMaintenanceRequest",
+    "MaintenanceAction",
+    "MaintenanceChange",
+    "MaintenanceImpact",
+    "MaintenanceImpactContributor",
+    "MaintenanceImpactEntry",
+    "MaintenanceResult",
+    "MaintenanceState",
+    "MaintenanceTrigger",
+]
diff --git a/src/videoipath_automation_tool/apps/inspect/snapshot.py b/src/videoipath_automation_tool/apps/inspect/snapshot.py
index 7b2493a..e52467a 100644
--- a/src/videoipath_automation_tool/apps/inspect/snapshot.py
+++ b/src/videoipath_automation_tool/apps/inspect/snapshot.py
@@ -42,12 +42,14 @@
     InspectSeverity,
     format_repr,
 )
+from videoipath_automation_tool.apps.inspect.model.maintenance import InspectApiMaintenanceBookingItem
 
 if TYPE_CHECKING:
     from videoipath_automation_tool.apps.inspect.api import InspectAPI
     from videoipath_automation_tool.apps.inspect.domain.alarm import InspectAlarm
     from videoipath_automation_tool.apps.inspect.domain.device import InspectDevice
     from videoipath_automation_tool.apps.inspect.domain.edge import InspectEdge
+    from videoipath_automation_tool.apps.inspect.domain.maintenance import InspectMaintenanceBooking
     from videoipath_automation_tool.apps.inspect.domain.module import InspectModule
     from videoipath_automation_tool.apps.inspect.domain.port import InspectPort
     from videoipath_automation_tool.apps.inspect.domain.service import InspectService
@@ -73,6 +75,7 @@ def __init__(
         device_level: HydrationLevel = HydrationLevel.SKELETON,
         path_items: list[InspectApiPathItem] | None = None,
         alarm_items: list[InspectApiAlarmItem] | None = None,
+        maintenance_items: list[InspectApiMaintenanceBookingItem] | None = None,
     ) -> None:
         self._fetcher = fetcher
         self._lock = threading.RLock()
@@ -111,6 +114,9 @@ def __init__(
         self._section_loaded: dict[str, bool] = {"paths": False, "alarms": False}
         self._section_fetched_at: dict[str, datetime] = {}
 
+        self._maintenance_by_id: dict[str, InspectApiMaintenanceBookingItem] = {}
+        self._maintenance_cache: dict[str, InspectMaintenanceBooking] = {}
+
         # Section: current alarms (status/alarms/current), indexed by resource key
         self._alarms: list[InspectApiAlarmItem] = []
         self._alarms_by_device_id: dict[str, list[InspectApiAlarmItem]] = {}
@@ -138,6 +144,10 @@ def __init__(
             self._index_paths(path_items)
             self._section_loaded["paths"] = True
             self._section_fetched_at["paths"] = self._created_at
+        if maintenance_items is not None:
+            self._maintenance_by_id = {item.id: item for item in maintenance_items}
+            self._section_loaded["maintenance"] = True
+            self._section_fetched_at["maintenance"] = self._created_at
         if alarm_items is not None:
             self._index_alarms(alarm_items)
             self._section_loaded["alarms"] = True
@@ -166,6 +176,7 @@ def from_full_response(
             edge_items=collector.external_edges_by_device_key_items,
             device_level=HydrationLevel.FULL,
             path_items=collector.inspect.path_items,
+            maintenance_items=collector.maintenance_booking_items,
         )
 
     # Backwards-compatible alias for the original draft API.
@@ -516,6 +527,65 @@ def clear_staged(
                     continue
                 self._pending_edits.pop(key, None)
 
+    # --- Maintenance ---
+
+    @property
+    def maintenance_bookings(self) -> list[InspectMaintenanceBooking]:
+        self._ensure_section_maintenance()
+        return [self.get_maintenance_booking(booking_id) for booking_id in self._maintenance_by_id]
+
+    def get_maintenance_record(self, booking_id: str) -> InspectApiMaintenanceBookingItem | None:
+        self._ensure_section_maintenance()
+        return self._maintenance_by_id.get(booking_id)
+
+    def get_maintenance_booking(self, booking_id: str) -> InspectMaintenanceBooking | None:
+        from videoipath_automation_tool.apps.inspect.domain.maintenance import InspectMaintenanceBooking
+
+        if self.get_maintenance_record(booking_id) is None:
+            return None
+        if booking_id not in self._maintenance_cache:
+            self._maintenance_cache[booking_id] = InspectMaintenanceBooking(snapshot=self, id=booking_id)
+        return self._maintenance_cache[booking_id]
+
+    def maintenance_for_resource(self, kind: str, resource_id: str) -> list[InspectMaintenanceBooking]:
+        """Bookings explicitly selecting a resource or a descendant (device/module).
+
+        Matches use collector contexts, not guessed ID prefixes. This performs no
+        topology hydration. Ancestor-wide bookings remain available on the ancestor.
+        """
+        result = []
+        for booking in self.maintenance_bookings:
+            item = booking.raw
+            if kind == "edge":
+                matched = resource_id in booking.edge_ids
+            else:
+                resources = [*item.devices, *item.modules, *item.ports]
+                resources += [r for edge in item.edges for r in (edge.from_, edge.to)]
+                field = {"device": "devicePid", "module": "modulePid", "port": "portPid"}[kind]
+                values = [getattr(r.context, field) for r in resources]
+                if kind == "device":
+                    values = [self._resolve_device_id(v) for v in values]
+                matched = resource_id in values
+            if matched:
+                result.append(booking)
+        return result
+
+    def invalidate_maintenance(self) -> None:
+        """Invalidate after a successful write without I/O or clearing staged edits.
+
+        Reroutes may change status beyond the selected targets. Re-fetch statuses
+        lazily; a refresh failure cannot turn an accepted write into a failure.
+        """
+        with self._lock:
+            self._section_loaded["maintenance"] = False
+            self._section_fetched_at.pop("maintenance", None)
+            self._maintenance_by_id.clear()
+            self._maintenance_cache.clear()
+            self._stale_devices.update(self._devices_by_id)
+            self._stale_pairs.update(self._edge_pairs)
+            self._mark_paths_stale()
+            self._mark_alarms_stale()
+
     # --- Refresh ---
 
     def refresh(self) -> InspectSnapshot:
@@ -825,6 +895,15 @@ def _ensure_section_alarms(self) -> None:
             self._section_loaded["alarms"] = True
             self._section_fetched_at["alarms"] = _now()
 
+    def _ensure_section_maintenance(self) -> None:
+        with self._lock:
+            if self._section_loaded.get("maintenance") or self._fetcher is None:
+                return
+            items = self._fetcher.get_maintenance_section()
+            self._maintenance_by_id = {item.id: item for item in items}
+            self._section_loaded["maintenance"] = True
+            self._section_fetched_at["maintenance"] = _now()
+
     # --- Internal: indexing ---
 
     def _index_device(self, node: InspectApiNodeStatusItem, level: HydrationLevel) -> None:
diff --git a/src/videoipath_automation_tool/connector/vip_rest_connector.py b/src/videoipath_automation_tool/connector/vip_rest_connector.py
index 52a1905..8f03bcc 100644
--- a/src/videoipath_automation_tool/connector/vip_rest_connector.py
+++ b/src/videoipath_automation_tool/connector/vip_rest_connector.py
@@ -18,7 +18,12 @@ class VideoIPathRestConnector(VideoIPathBaseConnector):
                 "/rest/v2/actions/status/network/",
                 "/rest/v2/actions/status/tags/",
             },
-            "EXACT_MATCHES": {"/rest/v2/actions/status/pathman/validateTopologyUpdate"},
+            "EXACT_MATCHES": {
+                "/rest/v2/actions/status/pathman/validateTopologyUpdate",
+                "/rest/v2/actions/status/pathman/updateMaintenance",
+                "/rest/v2/actions/status/pathman/validateMaintenanceImpactDetailed",
+                "/rest/v2/actions/status/pathman/fetchMaintenanceImpact",
+            },
         },
     }
 
diff --git a/tests/e2e/apps/test_inspect_maintenance.py b/tests/e2e/apps/test_inspect_maintenance.py
new file mode 100644
index 0000000..d3534ee
--- /dev/null
+++ b/tests/e2e/apps/test_inspect_maintenance.py
@@ -0,0 +1,304 @@
+"""Maintenance contracts on test-owned mocks, with booking-before-resource cleanup.
+
+Run: poetry run test-e2e tests/e2e/apps/test_inspect_maintenance.py
+Requires the normal .env configuration and VideoIPath 2026.2.x. No global sweep.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterator
+from datetime import UTC, datetime, time, timedelta
+from time import monotonic, sleep
+from typing import Literal
+from zoneinfo import ZoneInfo
+
+import pytest
+
+from videoipath_automation_tool.apps.inspect import (
+    InspectEntityNotFoundError,
+    InspectMaintenanceBooking,
+    InspectMaintenanceConflictError,
+    MaintenanceBookingSpec,
+    MaintenanceOnceSchedule,
+    MaintenanceRecurringSchedule,
+    MaintenanceState,
+    MaintenanceTargets,
+)
+from videoipath_automation_tool.apps.videoipath_app import VideoIPathApp
+
+from ..helpers import E2E_TAG, TopologyBuilder, edges_between, remove_devices, unique_label
+
+pytestmark = pytest.mark.e2e
+
+
+@pytest.fixture(scope="session")
+def e2e_sweep() -> None:
+    """Focused lifecycle tests clean their own resources, without sweeping other runs."""
+
+
+@pytest.fixture
+def maintenance_label(app: VideoIPathApp, topology_builder: TopologyBuilder) -> Iterator[str]:
+    """Own every booking via a unique label/tag; verify cleanup even after test failure."""
+    if not app._videoipath_connector.videoipath_version.startswith("2026.2."):
+        pytest.skip("Maintenance E2E targets the verified VideoIPath 2026.2 API.")
+    label = unique_label("MAINTENANCE")
+    try:
+        yield label
+    finally:
+        # Also finds a successful creation whose response was lost. Label updates
+        # retain the unique tag, so cleanup does not depend on the original title.
+        app.inspect.refresh(load="skeleton")
+        leftovers = _owned_bookings(app, label)
+        for booking in leftovers:
+            if booking.locked:
+                spec = MaintenanceBookingSpec(
+                    label=booking.label,
+                    description=booking.description,
+                    tags=booking.tags,
+                    targets=MaintenanceTargets(
+                        devices=booking.device_ids,
+                        modules=booking.module_ids,
+                        ports=booking.port_ids,
+                        edges=booking.edge_ids,
+                    ),
+                    schedule=MaintenanceOnceSchedule(start=booking.starts_at, end=booking.ends_at),
+                    action=booking.action,
+                    trigger=booking.trigger,
+                    allow_overlap=booking.allow_overlap,
+                    switch_format_state_on_reroute=booking.switch_format_state_on_reroute,
+                )
+                app.inspect.update_maintenance_booking(booking.id, spec, locked=False)
+        if leftovers:
+            app.inspect.delete_maintenance_bookings([b.id for b in leftovers])
+        _wait_bookings(app, label, expected_ids=set())
+        # Do not remove resources if booking cleanup failed above.
+        owned_devices = set(topology_builder.device_ids)
+        remove_devices(app, owned_devices)
+        app.inspect.refresh(load="skeleton")
+        assert owned_devices.isdisjoint(d.id for d in app.inspect.devices), "Test topology cleanup failed"
+        for device_id in owned_devices:
+            with pytest.raises(ValueError, match="No device with id"):
+                app.inventory.get_device(device_id=device_id, config_only=True)
+
+
+@pytest.mark.parametrize("frequency", ["once", "daily", "weekly", "monthly"])
+def test_maintenance_lifecycle(
+    app: VideoIPathApp,
+    topology_builder: TopologyBuilder,
+    maintenance_label: str,
+    frequency: Literal["once", "daily", "weekly", "monthly"],
+) -> None:
+    """Persist dates/metadata, protect revisions, preserve locks, and isolate sibling bookings."""
+    (device_id,) = topology_builder.add_devices([("MAINTENANCE-A", 2)])
+    device = app.inspect.get_device(device_id)
+    assert device is not None
+    now = datetime.now(UTC).replace(microsecond=0)
+    schedule = _schedule(frequency, now)
+    windows = schedule.expand() if isinstance(schedule, MaintenanceRecurringSchedule) else [schedule]
+    spec = MaintenanceBookingSpec(
+        label=maintenance_label,
+        description="E2E original maintenance",
+        tags=[E2E_TAG, maintenance_label],
+        targets=MaintenanceTargets(devices=[device]),
+        schedule=schedule,
+    )
+    # Isolated mocks have no services. Preview must not create any booking.
+    assert app.inspect.validate_maintenance_booking(spec) == {}
+    app.inspect.refresh()
+    assert _owned_bookings(app, maintenance_label) == []
+    created = app.inspect.create_maintenance_booking(spec)
+    assert created.ok and len(created.details) == len(windows)
+    ids = set(created.details)
+    bookings = _wait_bookings(app, maintenance_label, ids, states=dict.fromkeys(ids, MaintenanceState.SCHEDULED))
+    assert sorted((b.starts_at, b.ends_at) for b in bookings) == sorted((w.start, w.end) for w in windows)
+    assert all(b.rev == created.details[b.id].rev and not created.details[b.id].isCancel for b in bookings)
+    for booking in bookings:
+        assert booking.label == maintenance_label and booking.description == spec.description
+        assert set(booking.tags) == set(spec.tags)
+        assert booking.device_ids == spec.targets.devices
+        assert [d.id for d in booking.devices] == [device_id]
+        assert not booking.locked and not booking.allow_overlap and not booking.switch_format_state_on_reroute
+        assert booking.action == "nothing" and booking.trigger == "create"
+    assert ids <= {b.id for b in app.inspect.get_device(device_id).maintenance_bookings}
+    assert {b.id for b in app.inspect.find_maintenance_bookings("all", maintenance_label.lower())} == ids
+    assert {b.id for b in app.inspect.find_maintenance_bookings("scheduled", maintenance_label)} == ids
+    assert app.inspect.find_maintenance_bookings("active", maintenance_label) == []
+    impact = app.inspect.get_maintenance_impact(sorted(ids))
+    assert set(impact) <= ids and all(not services for services in impact.values())
+
+    booking = min(bookings, key=lambda b: b.starts_at)
+    original_rev = booking.rev
+    siblings = {b.id: b.raw.model_dump() for b in bookings if b.id != booking.id}
+    window_spec = spec.model_copy(
+        update={"schedule": MaintenanceOnceSchedule(start=booking.starts_at, end=booking.ends_at)}
+    )
+    assert app.inspect.validate_maintenance_booking(window_spec, booking_id=booking.id) == {}
+    assert app.inspect.update_maintenance_booking(booking.id, window_spec, locked=True, expected_rev=original_rev).ok
+    locked = app.inspect.get_maintenance_booking(booking.id)
+    assert locked is not None and locked.locked and locked.rev != original_rev
+    locked_rev = locked.rev
+    revised = window_spec.model_copy(
+        update={"label": unique_label("MAINTENANCE-RENAMED"), "description": "E2E revised maintenance"}
+    )
+    with pytest.raises(InspectMaintenanceConflictError) as conflict:
+        app.inspect.update_maintenance_booking(booking.id, revised, expected_rev=original_rev)
+    assert conflict.value.booking_ids == [booking.id]
+    assert conflict.value.expected_rev == original_rev and conflict.value.actual_rev == locked_rev
+    app.inspect.refresh()
+    unchanged = app.inspect.get_maintenance_booking(booking.id)
+    assert unchanged.rev == locked_rev and unchanged.label == maintenance_label
+
+    # A real metadata edit with locked omitted must preserve the lock.
+    assert app.inspect.update_maintenance_booking(booking.id, revised).ok
+    updated = app.inspect.get_maintenance_booking(booking.id)
+    assert updated.locked and updated.label == revised.label and updated.description == revised.description
+    assert updated.starts_at == window_spec.schedule.start and updated.ends_at == window_spec.schedule.end
+    assert set(updated.tags) == set(spec.tags)
+    assert maintenance_label not in updated.label  # Subsequent searches must find this ID through its unique tag.
+    assert app.inspect.update_maintenance_booking(booking.id, revised, locked=False).ok
+    assert not app.inspect.get_maintenance_booking(booking.id).locked
+
+    # Start one explicit ID, including an occurrence created by recurrence.
+    active_end = datetime.now(UTC).replace(microsecond=0) + timedelta(hours=1)
+    active_spec = revised.model_copy(update={"schedule": MaintenanceOnceSchedule(end=active_end)})
+    assert app.inspect.update_maintenance_booking(booking.id, active_spec).ok
+    states = dict.fromkeys(ids, MaintenanceState.SCHEDULED) | {booking.id: MaintenanceState.ACTIVE}
+    current = _wait_bookings(app, maintenance_label, ids, states=states)
+    active = next(b for b in current if b.id == booking.id)
+    # Start-now uses the server clock, which need not match the test runner's clock.
+    assert active.starts_at < window_spec.schedule.start
+    assert active.ends_at == active_end
+    assert {b.id for b in app.inspect.find_maintenance_bookings("active", maintenance_label)} == {booking.id}
+    assert {b.id for b in app.inspect.find_maintenance_bookings("scheduled", maintenance_label)} == ids - {booking.id}
+    assert {b.id: b.raw.model_dump() for b in current if b.id != booking.id} == siblings
+
+    deleted = app.inspect.delete_maintenance_bookings(sorted(ids))
+    assert deleted.ok and set(deleted.details) == ids
+    assert all(change.isCancel for change in deleted.details.values())
+    assert _owned_bookings(app, maintenance_label) == []  # Successful writes invalidate cached reads.
+    _wait_bookings(app, maintenance_label, expected_ids=set())
+    assert app.inspect.get_maintenance_booking(booking.id) is None
+    with pytest.raises(InspectEntityNotFoundError):
+        app.inspect.update_maintenance_booking(booking.id, window_spec)
+
+
+@pytest.mark.parametrize("kind", ["module", "port", "edge"])
+def test_maintenance_resource_targets(
+    app: VideoIPathApp,
+    topology_builder: TopologyBuilder,
+    maintenance_label: str,
+    kind: Literal["module", "port", "edge"],
+) -> None:
+    """Domain targets round-trip as selected resources with inverse booking relationships."""
+    id_a, id_b = topology_builder.add_devices([("MAINTENANCE-A", 2), ("MAINTENANCE-B", 2)])
+    topology_builder.link(id_a, id_b)
+    device = app.inspect.get_device(id_a)
+    assert device is not None
+    port = next(p for p in device.ports if p.vertex_out is not None)
+    resource = {"module": port.module, "port": port, "edge": edges_between(app, id_a, id_b)[0]}[kind]
+    targets = MaintenanceTargets(**{f"{kind}s": [resource]})
+    start = datetime.now(UTC).replace(microsecond=0) + timedelta(days=1)
+    spec = MaintenanceBookingSpec(
+        label=maintenance_label,
+        tags=[E2E_TAG, maintenance_label],
+        targets=targets,
+        schedule=MaintenanceOnceSchedule(start=start, end=start + timedelta(hours=1)),
+    )
+    assert app.inspect.validate_maintenance_booking(spec) == {}
+    result = app.inspect.create_maintenance_booking(spec)
+    assert result.ok and len(result.details) == 1
+    (booking,) = _wait_bookings(app, maintenance_label, set(result.details))
+    assert getattr(booking, f"{kind}_ids") == getattr(targets, f"{kind}s")
+    assert [r.id for r in getattr(booking, f"{kind}s")] == [resource.id]
+    assert booking.id in {b.id for b in resource.maintenance_bookings}
+    assert booking.id in {b.id for b in app.inspect.get_device(id_a).maintenance_bookings}
+    app.inspect.refresh(load="full")
+    loaded = app.inspect.get_maintenance_booking(booking.id)
+    assert loaded is not None and loaded.raw == booking.raw
+    # Fixture also verifies cleanup for bookings left on the server by the test.
+
+
+def test_immediate_open_ended_and_staged_topology_edits(
+    app: VideoIPathApp, topology_builder: TopologyBuilder, maintenance_label: str
+) -> None:
+    """Immediate/open-ended timestamps round-trip; maintenance cannot flush or discard topology edits."""
+    (device_id,) = topology_builder.add_devices([("MAINTENANCE-OPEN", 2)])
+    device = app.inspect.get_device(device_id)
+    assert device is not None
+    pending_label = f"{maintenance_label}-PENDING-DEVICE"
+    device.label = pending_label
+    spec = MaintenanceBookingSpec(
+        label=maintenance_label,
+        tags=[E2E_TAG, maintenance_label],
+        targets=MaintenanceTargets(devices=[device]),
+        schedule=MaintenanceOnceSchedule(),
+    )
+    assert app.inspect.validate_maintenance_booking(spec) == {}
+    created = app.inspect.create_maintenance_booking(spec)
+    assert created.ok and len(created.details) == 1
+    assert device.label == pending_label
+    assert app.inspect.get_device(device_id).label == pending_label
+    # A fresh view ignores uncommitted edits, proving the maintenance action did not commit them.
+    (booking,) = _wait_bookings(
+        app, maintenance_label, set(created.details), states=dict.fromkeys(created.details, MaintenanceState.ACTIVE)
+    )
+    assert app.inspect.get_device(device_id).label == topology_builder.labels[device_id]
+    assert booking.ends_at is None and booking.schedule.infinite
+    assert booking.starts_at.tzinfo is not None
+    end = datetime.now(UTC).replace(microsecond=0) + timedelta(hours=1)
+    finite = spec.model_copy(update={"schedule": MaintenanceOnceSchedule(start=booking.starts_at, end=end)})
+    assert app.inspect.update_maintenance_booking(booking.id, finite, locked=True).ok
+    current = app.inspect.get_maintenance_booking(booking.id)
+    assert current.ends_at == end and not current.schedule.infinite and current.locked
+    # Teardown must unlock this remaining booking, delete it, then remove its mock device.
+
+
+def _schedule(
+    frequency: Literal["once", "daily", "weekly", "monthly"], now: datetime
+) -> MaintenanceOnceSchedule | MaintenanceRecurringSchedule:
+    if frequency == "once":
+        return MaintenanceOnceSchedule(start=now + timedelta(days=1), end=now + timedelta(days=2))
+    zone = ZoneInfo("Europe/Berlin")
+    start = (now.astimezone(zone) + timedelta(days=1)).replace(hour=0, minute=0, second=0, microsecond=0)
+    if frequency == "monthly":
+        # Anchor on day 1, ensuring at least two independent occurrences in every month.
+        start = (start + timedelta(days=32)).replace(day=1)
+    days = {"daily": 3, "weekly": 14, "monthly": 65}[frequency]
+    return MaintenanceRecurringSchedule(
+        frequency=frequency,
+        start=start,
+        end=start + timedelta(days=days),
+        timezone=zone.key,
+        local_start=time(4),
+        local_end=time(5),
+        weekdays=[1, 3] if frequency == "weekly" else [],
+    )
+
+
+def _owned_bookings(app: VideoIPathApp, label: str) -> list[InspectMaintenanceBooking]:
+    return [
+        b
+        for b in app.inspect.find_maintenance_bookings(search=label)
+        if b.label == label or b.label.startswith(f"{label}-") or label in b.tags
+    ]
+
+
+def _wait_bookings(
+    app: VideoIPathApp,
+    label: str,
+    expected_ids: set[str],
+    *,
+    states: dict[str, MaintenanceState] | None = None,
+) -> list[InspectMaintenanceBooking]:
+    deadline = monotonic() + 10
+    while True:
+        app.inspect.refresh(load="skeleton")
+        bookings = _owned_bookings(app, label)
+        observed = {b.id: b.state for b in bookings}
+        if set(observed) == expected_ids and (states is None or observed == states):
+            return bookings
+        if monotonic() >= deadline:
+            pytest.fail(
+                f"Maintenance collector did not converge: expected IDs {expected_ids}, states {states}; got {observed}"
+            )
+        sleep(0.2)
diff --git a/tests/inspect/test_exports.py b/tests/inspect/test_exports.py
index e348479..0aae0fa 100644
--- a/tests/inspect/test_exports.py
+++ b/tests/inspect/test_exports.py
@@ -11,6 +11,7 @@
     alarms,
     collector,
     common,
+    maintenance,
     ngraph,
     tags,
     update_topology,
@@ -24,6 +25,7 @@ def test_model_package_all_aggregates_submodules() -> None:
         *alarms.__all__,
         *collector.__all__,
         *common.__all__,
+        *maintenance.__all__,
         *ngraph.__all__,
         *tags.__all__,
         *update_topology.__all__,
diff --git a/tests/inspect/test_maintenance.py b/tests/inspect/test_maintenance.py
new file mode 100644
index 0000000..a1b13e7
--- /dev/null
+++ b/tests/inspect/test_maintenance.py
@@ -0,0 +1,578 @@
+"""Maintenance contracts: real wire shapes, fake transport, no live data."""
+
+from __future__ import annotations
+
+from copy import deepcopy
+from datetime import UTC, datetime, time, timedelta
+from types import SimpleNamespace
+from typing import Any
+from zoneinfo import ZoneInfo
+
+import pytest
+from pydantic import ValidationError
+
+from videoipath_automation_tool.apps.inspect import (
+    InspectApp,
+    InspectEntityNotFoundError,
+    InspectMaintenanceConflictError,
+    InspectMaintenanceError,
+    MaintenanceBookingSpec,
+    MaintenanceOnceSchedule,
+    MaintenanceRecurringSchedule,
+    MaintenanceTargets,
+)
+from videoipath_automation_tool.apps.inspect.api import InspectAPI, queries
+from videoipath_automation_tool.apps.inspect.model.collector import (
+    InspectApiCollectorResponse,
+    InspectApiExternalEdgesByDeviceKeyItem,
+    InspectApiNodeStatusItem,
+)
+from videoipath_automation_tool.apps.inspect.model.common import InspectApiRestV2Header
+from videoipath_automation_tool.apps.inspect.snapshot import HydrationLevel, InspectSnapshot
+from videoipath_automation_tool.connector.vip_rest_connector import VideoIPathRestConnector
+
+START = datetime(2030, 1, 1, tzinfo=UTC)
+
+
+@pytest.fixture
+def booking() -> dict[str, Any]:
+    return {
+        "_id": "booking-a",
+        "_vid": "view-a",
+        "rev": "revision-a",
+        "generic": {
+            "descriptor": {"label": "maintenance-a", "desc": "Example maintenance"},
+            "state": 0,
+            "locked": False,
+        },
+        "scheduleInfo": {"startTimestamp": 1893456000000, "endTimestamp": 1893542400000, "infinite": False},
+        "action": "nothing",
+        "trigger": "create",
+        "allowOverlap": False,
+        "switchFormatStateOnReroute": False,
+        "tags": ["tag-a"],
+        "devices": [{"label": "device-a", "context": {"devicePid": "device-a"}, "pid": ["device-a"]}],
+        "modules": [
+            {
+                "label": "module-1",
+                "context": {"devicePid": "device-a", "modulePid": "device-a.dev.0"},
+                "pid": ["device-a", "dev", "0"],
+            }
+        ],
+        "ports": [
+            {
+                "label": "port-out-1",
+                "context": {
+                    "devicePid": "device-a",
+                    "modulePid": "device-a.dev.0",
+                    "portPid": "device-a.dev.0.port-out-1",
+                },
+                "pid": ["device-a", "dev", "0", "port-out-1"],
+            }
+        ],
+        "edges": [],
+    }
+
+
+def test_once_schedule_timezone_nulls_and_validation() -> None:
+    start = datetime(2030, 1, 1, 1, tzinfo=ZoneInfo("Europe/Berlin"))
+    assert MaintenanceOnceSchedule(start=start).to_wire().startTimestamp == 1893456000000
+    assert MaintenanceOnceSchedule().to_wire().model_dump() == {
+        "type": "once",
+        "startTimestamp": None,
+        "endTimestamp": None,
+    }
+    with pytest.raises(ValidationError):
+        MaintenanceOnceSchedule(start=datetime(2030, 1, 1))  # noqa: DTZ001 — invalid input under test
+    with pytest.raises(ValidationError):
+        MaintenanceOnceSchedule(start=START, end=START)
+
+
+def test_once_schedule_orders_instants_across_dst_fold() -> None:
+    zone = ZoneInfo("Europe/Berlin")
+    first = datetime(2030, 10, 27, 2, 45, tzinfo=zone, fold=0)
+    second = datetime(2030, 10, 27, 2, 15, tzinfo=zone, fold=1)
+    wire = MaintenanceOnceSchedule(start=first, end=second).to_wire()
+    assert wire.endTimestamp - wire.startTimestamp == 30 * 60 * 1000
+    with pytest.raises(ValidationError):
+        MaintenanceOnceSchedule(start=second, end=first)
+
+
+@pytest.mark.parametrize("frequency,code,weekdays", [("daily", 0, []), ("weekly", 1, [1, 5]), ("monthly", 2, [])])
+def test_recurring_encoding(frequency: str, code: int, weekdays: list[int]) -> None:
+    schedule = _recurring(
+        frequency=frequency, weekdays=weekdays, pattern_iteration_filter=[2], instance_iteration_filter=[3]
+    )
+    wire = schedule.to_wire().model_dump()
+    assert wire == {
+        "type": "recurring",
+        "pattern": {
+            "pattern": code,
+            "startTime": 1893456000000,
+            "endTime": 1896134400000,
+            "timeZoneId": "Europe/Berlin",
+            "weekDays": weekdays,
+            "iterationFilter": [2],
+        },
+        "instance": {"localStartTime": 3600, "localEndTime": 7200, "iterationFilter": [3]},
+    }
+
+
+@pytest.mark.parametrize(
+    "changes",
+    [
+        {"timezone": "not-a-zone"},
+        {"end": START},
+        {"start": datetime(2030, 1, 1)},  # noqa: DTZ001 — invalid input under test
+        {"frequency": "weekly"},
+        {"frequency": "weekly", "weekdays": [0]},
+        {"frequency": "weekly", "weekdays": [1, 1]},
+        {"weekdays": [2]},
+        {"local_end": time(1)},
+        {"local_start": time(1, tzinfo=UTC)},
+        {"local_start": time(1, microsecond=1)},
+    ],
+)
+def test_invalid_recurring_rules(changes: dict[str, Any]) -> None:
+    with pytest.raises(ValidationError):
+        _recurring(**changes)
+
+
+def test_recurrence_is_batched_as_dated_windows() -> None:
+    app, rest = _app([])
+    app.create_maintenance_booking(_spec(schedule=_recurring()))
+    assert len(rest.posts) == 1
+    creates = rest.posts[0][1]["data"]["create"]
+    assert len(creates) == 31
+    assert all(c["scheduleInfo"]["type"] == "once" for c in creates)
+    assert creates[0]["scheduleInfo"] == {
+        "type": "once",
+        "startTimestamp": 1893456000000,
+        "endTimestamp": 1893459600000,
+    }
+    assert creates[1]["scheduleInfo"]["startTimestamp"] - creates[0]["scheduleInfo"]["startTimestamp"] == 86400000
+
+
+def test_weekly_monthly_overnight_and_dst_expansion() -> None:
+    weekly = _recurring(frequency="weekly", weekdays=[1, 5]).expand()
+    assert all(w.start.astimezone(ZoneInfo("Europe/Berlin")).isoweekday() in (1, 5) for w in weekly)
+    monthly = _recurring(
+        frequency="monthly",
+        start=datetime(2030, 1, 31, tzinfo=UTC),
+        end=datetime(2030, 5, 1, tzinfo=UTC),
+        local_end=time(4),
+    ).expand()
+    assert [w.start.month for w in monthly] == [1, 3]  # No February 31 or April 31.
+    overnight = _recurring(local_start=time(23), local_end=time(1)).expand()[0]
+    assert overnight.end - overnight.start == timedelta(hours=2)
+    dst = _recurring(
+        start=datetime(2030, 3, 30, tzinfo=UTC),
+        end=datetime(2030, 4, 2, tzinfo=UTC),
+        local_start=time(1),
+        local_end=time(4),
+    ).expand()
+    assert [w.end - w.start for w in dst] == [timedelta(hours=3), timedelta(hours=2), timedelta(hours=3)]
+    with pytest.raises(ValueError, match="Nonexistent"):
+        _recurring(
+            start=datetime(2030, 3, 30, tzinfo=UTC),
+            end=datetime(2030, 4, 2, tzinfo=UTC),
+            local_start=time(2, 30),
+            local_end=time(4),
+        ).expand()
+    ambiguous = _recurring(
+        start=datetime(2030, 10, 27, tzinfo=UTC),
+        end=datetime(2030, 10, 28, tzinfo=UTC),
+        local_start=time(2, 30),
+        local_end=time(3),
+    ).expand()
+    assert ambiguous[0].start.hour == 0  # First 02:30 (summer offset).
+
+
+def test_expansion_validation_precedes_writes() -> None:
+    app, rest = _app([])
+    for schedule in [
+        _recurring(pattern_iteration_filter=[1]),
+        _recurring(instance_iteration_filter=[1]),
+        _recurring(end=START + timedelta(minutes=1)),
+        _recurring(end=START + timedelta(days=1002)),
+    ]:
+        with pytest.raises(ValueError):
+            app.create_maintenance_booking(_spec(schedule=schedule))
+    assert not rest.posts
+
+
+def test_targets_normalize_objects_and_keep_canonical_identifiers() -> None:
+    snapshot = _snapshot()
+    device = snapshot.get_device("device-a")
+    module = device.modules[0]
+    port = module.ports[0]
+    targets = MaintenanceTargets(devices=[device, "device-a"], modules=[module], ports=[port])
+    assert targets.pids == ["device-a", "device-a.dev.0", "device-a.dev.0.port-out-1"]
+    assert targets.devices == ["device-a"]
+    with pytest.raises(ValidationError):
+        MaintenanceTargets(devices=[module])
+    with pytest.raises(ValidationError):
+        MaintenanceTargets(devices=[""])
+    with pytest.raises(ValidationError):
+        MaintenanceTargets(devices="device-a")
+    with pytest.raises(ValidationError):
+        MaintenanceBookingSpec(label="a", targets=MaintenanceTargets(), schedule=MaintenanceOnceSchedule())
+
+
+def test_virtual_device_pid_differs_from_id() -> None:
+    node = InspectApiNodeStatusItem(_id="virtual-1", _vid="1", deviceId="virtual.1", pid="virtual-1")
+    snapshot = InspectSnapshot(device_items=[node])
+    assert MaintenanceTargets(devices=[snapshot.get_device("virtual.1")]).pids == ["virtual-1"]
+
+
+def test_lazy_section_filters_and_resource_relationships(booking: dict[str, Any]) -> None:
+    active = deepcopy(booking)
+    active.update(_id="booking-b", tags=["SPECIAL"])
+    active["generic"]["state"] = 1
+    future = deepcopy(booking)
+    future.update(_id="booking-c")
+    future["generic"]["state"] = 9
+    app, rest = _app([booking, active, future])
+    assert rest.get_calls == []
+    assert len(app.maintenance_bookings) == 3
+    assert app.get_maintenance_booking("booking-a") is app.maintenance_bookings[0]
+    assert app.get_maintenance_booking("missing") is None
+    assert len(app.find_maintenance_bookings("scheduled", "MAINTENANCE")) == 1
+    assert len(app.find_maintenance_bookings("active", "special")) == 1
+    assert app.get_maintenance_booking("booking-c").state == 9
+    with pytest.raises(ValueError):
+        app.find_maintenance_bookings("invalid")
+    item = app.get_maintenance_booking("booking-a")
+    assert item.starts_at == START
+    assert item.ends_at == START + timedelta(days=1)
+    assert item.description == "Example maintenance"
+    assert item.action == "nothing" and item.trigger == "create"
+    assert not item.allow_overlap and not item.switch_format_state_on_reroute
+    assert item.devices[0].id == "device-a"
+    assert item.modules[0].id == "device-a.dev.0"
+    assert item.ports[0].id == "device-a.dev.0.port-out-1"
+    assert item.id in [b.id for b in item.devices[0].maintenance_bookings]
+    assert item.id in [b.id for b in item.modules[0].maintenance_bookings]
+    assert item.id in [b.id for b in item.ports[0].maintenance_bookings]
+    assert rest.get_calls == [queries.maintenance_section()]
+
+
+def test_full_snapshot_reuses_bookings_and_preserves_unresolved_ids(booking: dict[str, Any]) -> None:
+    booking["scheduleInfo"].update(infinite=True, endTimestamp=9223372036854775807)
+    response = InspectApiCollectorResponse.model_validate(
+        {
+            "header": _header().model_dump(),
+            "data": {"status": {"collector": {"maintenanceBookings": {"_items": [booking]}}}},
+        }
+    )
+    snapshot = InspectSnapshot.from_full_response(response)
+    item = snapshot.maintenance_bookings[0]
+    assert item.ends_at is None
+    assert item.devices == [] and item.modules == [] and item.ports == []
+    assert item.raw.ports[0].pid == ["device-a", "dev", "0", "port-out-1"]
+    assert snapshot.section_fetched_at("maintenance") is not None
+
+
+def test_empty_reads_and_full_section_do_not_hydrate_devices() -> None:
+    app, rest = _app([])
+    node = InspectApiNodeStatusItem(_id="device-a", _vid="device-a", deviceId="device-a")
+    app._snapshot = InspectSnapshot(fetcher=app._inspect_api, device_items=[node])
+    assert app.maintenance_bookings == []
+    assert app.find_maintenance_bookings("active") == []
+    assert app.get_maintenance_booking("missing") is None
+    assert not app._snapshot.is_device_hydrated("device-a")
+    assert rest.get_calls == [queries.maintenance_section()]
+    response = InspectApiCollectorResponse.model_validate(
+        {"header": _header().model_dump(), "data": {"status": {"collector": {"maintenanceBookings": {"_items": []}}}}}
+    )
+    app._snapshot = InspectSnapshot.from_full_response(response, fetcher=app._inspect_api)
+    rest.fail_get = True
+    assert app.maintenance_bookings == []
+    assert app.validate_maintenance_booking(_spec()) == {}
+
+
+def test_edge_targets_and_local_module_keys_resolve(booking: dict[str, Any]) -> None:
+    app, rest = _app([booking])
+    record = app._snapshot.get_device_record("device-a")
+    module = record.node.modules.pop("device-a.dev.0")
+    record.node.modules["module-local-key"] = module
+    endpoint = deepcopy(booking["ports"][0])
+    rest.bookings[0].update(
+        devices=[], modules=[], ports=[], edges=[{"id": "edge-a", "from": endpoint, "to": endpoint}]
+    )
+    status = endpoint | {"pid": endpoint["context"]["portPid"]}
+    pair = InspectApiExternalEdgesByDeviceKeyItem.model_validate(
+        {
+            "_id": "device-a::device-b",
+            "_vid": "pair-a",
+            "primary": {
+                "devicePid": "device-a",
+                "data": {"edge-a": {"id": "edge-a", "fromStatus": status, "toStatus": status}},
+            },
+            "secondary": {"devicePid": "device-b", "data": {}},
+        }
+    )
+    app._snapshot = InspectSnapshot(
+        fetcher=app._inspect_api, device_items=[record.node], edge_items=[pair], device_level=HydrationLevel.FULL
+    )
+    item = app.maintenance_bookings[0]
+    edge = item.edges[0]
+    assert MaintenanceTargets(edges=[edge, "edge-a"]).edges == ["edge-a"]
+    assert [b.id for b in edge.maintenance_bookings] == [item.id]
+    device = app._snapshot.get_device("device-a")
+    assert [b.id for b in device.maintenance_bookings] == [item.id]
+    assert [b.id for b in device.modules[0].maintenance_bookings] == [item.id]
+    assert [b.id for b in device.ports[0].maintenance_bookings] == [item.id]
+    assert rest.get_calls == [queries.maintenance_section()]
+
+
+def test_result_uses_server_detail_ids_and_never_guesses() -> None:
+    app, rest = _app([])
+    assert app.create_maintenance_booking(_spec()).details == {}
+    rest.details = {"server-booking-a": {"rev": "revision-a", "isCancel": False, "status": 0, "type": "generic"}}
+    result = app.create_maintenance_booking(_spec())
+    assert list(result.details) == ["server-booking-a"]
+    assert result.details["server-booking-a"].rev == "revision-a"
+
+
+def test_create_update_delete_payloads_and_revision(booking: dict[str, Any]) -> None:
+    booking["generic"]["locked"] = True
+    app, rest = _app([booking])
+    spec = _spec()
+    assert app.create_maintenance_booking(spec).ok
+    create = rest.posts[-1][1]["data"]
+    assert create["update"] == [] and create["delete"] == []
+    definition = create["create"][0]["serviceDefinition"]
+    assert definition["devicePids"] == ["device-a"] and "pids" not in definition
+    assert app.update_maintenance_booking("booking-a", spec, expected_rev="revision-a").ok
+    update = rest.posts[-1][1]["data"]["update"][0]
+    assert update["id"] == "booking-a" and update["rev"] == "revision-a"
+    assert update["inheritable"]["locked"] is True
+    assert update["updater"]["pids"] == ["device-a"] and "devicePids" not in update["updater"]
+    assert app.update_maintenance_booking("booking-a", spec, locked=False).ok
+    assert not rest.posts[-1][1]["data"]["update"][0]["inheritable"]["locked"]
+    assert app.delete_maintenance_bookings(["booking-a", "booking-a"]).ok
+    assert rest.posts[-1][1]["data"] == {"create": [], "update": [], "delete": ["booking-a"]}
+    assert all(p.endswith("/updateMaintenance") for p, _ in rest.posts)
+
+
+def test_update_uses_fresh_revision_over_cached_booking(booking: dict[str, Any]) -> None:
+    app, rest = _app([booking])
+    assert app.maintenance_bookings[0].rev == "revision-a"
+    rest.bookings[0]["rev"] = "revision-b"
+    rest.bookings[0]["generic"]["locked"] = True
+    app.update_maintenance_booking("booking-a", _spec())
+    update = rest.posts[-1][1]["data"]["update"][0]
+    assert update["rev"] == "revision-b" and update["inheritable"]["locked"]
+
+
+@pytest.mark.parametrize("action", ["nothing", "invalidate", "reroute", "rerouteSA"])
+def test_action_settings_are_serialized(action: str) -> None:
+    app, rest = _app([])
+    app.create_maintenance_booking(
+        _spec(action=action, trigger="active", allow_overlap=True, switch_format_state_on_reroute=True)
+    )
+    definition = rest.posts[-1][1]["data"]["create"][0]["serviceDefinition"]
+    assert definition["action"] == action and definition["trigger"] == "active"
+    assert definition["allowOverlap"] and definition["switchFormatStateOnReroute"]
+
+
+def test_missing_or_conflicting_update_does_not_write(booking: dict[str, Any]) -> None:
+    app, rest = _app([booking])
+    with pytest.raises(InspectEntityNotFoundError):
+        app.update_maintenance_booking("absent", _spec())
+    with pytest.raises(InspectMaintenanceConflictError) as caught:
+        app.update_maintenance_booking("booking-a", _spec(), expected_rev="old")
+    assert caught.value.actual_rev == "revision-a"
+    assert rest.posts == []
+
+
+def test_write_revalidates_nested_lists_without_io() -> None:
+    app, rest = _app([])
+    spec = _spec()
+    spec.targets.devices.clear()
+    with pytest.raises(ValidationError):
+        app.create_maintenance_booking(spec)
+    assert not rest.posts and not rest.get_calls
+
+
+def test_recurring_update_requires_explicit_one_time_window(booking: dict[str, Any]) -> None:
+    app, rest = _app([booking])
+    with pytest.raises(ValueError, match="one-time schedule"):
+        app.update_maintenance_booking("booking-a", _spec(schedule=_recurring()))
+    assert not rest.posts and not rest.get_calls
+
+
+def test_success_invalidates_without_losing_staged_edits_or_result(booking: dict[str, Any]) -> None:
+    app, rest = _app([booking])
+    held = app.maintenance_bookings[0]
+    snapshot = app._snapshot
+    snapshot.stage_edit("device", "device-a", "descriptor.label", "pending")
+    rest.fail_get = True
+    assert app.create_maintenance_booking(_spec()).ok  # Invalidation performs no I/O.
+    assert snapshot.get_staged_edits("device", "device-a") == {"descriptor.label": "pending"}
+    assert "device-a" in snapshot._stale_devices
+    assert not snapshot._section_loaded["maintenance"]
+    with pytest.raises(InspectMaintenanceError):
+        _ = held.label
+    rest.fail_get = False
+    rest.bookings[0]["generic"]["descriptor"]["label"] = "updated"
+    assert held.label == "updated"
+
+
+@pytest.mark.parametrize("failure", ["result", "header", "malformed", "timeout"])
+def test_failed_writes_are_typed_and_never_retried(booking: dict[str, Any], failure: str) -> None:
+    app, rest = _app([booking])
+    _ = app.maintenance_bookings
+    rest.failure = failure
+    with pytest.raises(InspectMaintenanceError) as caught:
+        app.create_maintenance_booking(_spec())
+    assert caught.value.operation == "create"
+    assert len(rest.posts) == 1
+    assert app._snapshot._section_loaded["maintenance"]
+
+
+def test_preview_and_current_impact_are_read_only(booking: dict[str, Any]) -> None:
+    app, rest = _app([booking])
+    rest.impact = {
+        "service-a": {
+            "label": "service-a",
+            "start": 0,
+            "end": 1000,
+            "protection": "none",
+            "maxSaBefore": 0,
+            "maxSaAfter": 2,
+            "entries": [
+                {
+                    "start": 0,
+                    "end": 1000,
+                    "saBefore": 0,
+                    "saAfter": 2,
+                    "contributors": {"booking-a": {"label": "maintenance-a"}},
+                }
+            ],
+        }
+    }
+    impact = app.validate_maintenance_booking(_spec(), booking_id="booking-a")
+    assert impact["service-a"][0].entries[0].contributors["booking-a"].label == "maintenance-a"
+    assert rest.posts[-1][1]["data"]["id"] == "booking-a"
+    assert app.get_maintenance_impact(["booking-a"])["booking-a"]["service-a"].maxSaAfter == 2
+    assert all("updateMaintenance" not in path for path, _ in rest.posts)
+    assert rest.get_calls == []
+
+
+@pytest.mark.parametrize("ids", [[], [""], [" a "], "booking-a"])
+def test_invalid_booking_ids(ids: Any) -> None:
+    app, rest = _app([])
+    with pytest.raises(ValueError):
+        app.delete_maintenance_bookings(ids)
+    with pytest.raises(ValueError):
+        app.get_maintenance_impact(ids)
+    assert rest.posts == []
+
+
+def test_exact_connector_allowlist() -> None:
+    connector = object.__new__(VideoIPathRestConnector)
+    for action in ("updateMaintenance", "fetchMaintenanceImpact", "validateMaintenanceImpactDetailed"):
+        connector._validate_url(f"/rest/v2/actions/status/pathman/{action}", "POST")
+    with pytest.raises(ValueError):
+        connector._validate_url("/rest/v2/actions/status/pathman/unrelatedAction", "POST")
+
+
+def _recurring(**changes: Any) -> MaintenanceRecurringSchedule:
+    return MaintenanceRecurringSchedule(
+        **(
+            {
+                "frequency": "daily",
+                "start": START,
+                "end": START + timedelta(days=31),
+                "timezone": "Europe/Berlin",
+                "local_start": time(1),
+                "local_end": time(2),
+            }
+            | changes
+        )
+    )
+
+
+def _spec(**changes: Any) -> MaintenanceBookingSpec:
+    return MaintenanceBookingSpec(
+        **(
+            {
+                "label": "maintenance-a",
+                "targets": MaintenanceTargets(devices=["device-a"]),
+                "schedule": MaintenanceOnceSchedule(start=START, end=START + timedelta(days=1)),
+            }
+            | changes
+        )
+    )
+
+
+def _header(**changes: Any) -> InspectApiRestV2Header:
+    return InspectApiRestV2Header(
+        **(
+            {"auth": True, "caption": "OK", "code": "OK", "id": "0", "msg": [], "ok": True, "user": "test-user"}
+            | changes
+        )
+    )
+
+
+def _snapshot(fetcher: InspectAPI | None = None) -> InspectSnapshot:
+    node = InspectApiNodeStatusItem.model_validate(
+        {
+            "_id": "device-a",
+            "_vid": "1",
+            "deviceId": "device-a",
+            "pid": "device-a",
+            "context": {"devicePid": "device-a"},
+            "modules": {
+                "device-a.dev.0": {
+                    "pid": "device-a.dev.0",
+                    "context": {"devicePid": "device-a", "modulePid": "device-a.dev.0"},
+                    "ports": {"device-a.dev.0.port-out-1": {"pid": "device-a.dev.0.port-out-1"}},
+                }
+            },
+        }
+    )
+    return InspectSnapshot(fetcher=fetcher, device_items=[node], device_level=HydrationLevel.FULL)
+
+
+def _app(bookings: list[dict[str, Any]]) -> tuple[InspectApp, FakeRest]:
+    rest = FakeRest(bookings)
+    with pytest.warns(UserWarning, match="beta"):
+        app = InspectApp(SimpleNamespace(rest=rest, videoipath_version="2026.2.0"))
+    app._snapshot = _snapshot(app._inspect_api)
+    return app, rest
+
+
+class FakeRest:
+    def __init__(self, bookings: list[dict[str, Any]]) -> None:
+        self.bookings = deepcopy(bookings)
+        self.get_calls: list[str] = []
+        self.posts: list[tuple[str, dict[str, Any]]] = []
+        self.fail_get = False
+        self.failure: str | None = None
+        self.impact: dict[str, Any] = {}
+        self.details: dict[str, Any] = {}
+
+    def get(self, path: str, **kwargs: Any) -> SimpleNamespace:
+        self.get_calls.append(path)
+        if self.fail_get:
+            raise ConnectionError("test read failed")
+        assert path == queries.maintenance_section()
+        return SimpleNamespace(
+            header=_header(), data={"status": {"collector": {"maintenanceBookings": {"_items": self.bookings}}}}
+        )
+
+    def post(self, path: str, body: Any) -> SimpleNamespace:
+        self.posts.append((path, body.model_dump(mode="json", by_alias=True)))
+        if self.failure == "timeout":
+            raise TimeoutError("test timeout")
+        data: Any = {"result": {"ok": self.failure != "result", "msg": ["test response"]}, "details": self.details}
+        if self.failure == "malformed":
+            data = {}
+        elif path.endswith("validateMaintenanceImpactDetailed"):
+            data = self.impact
+        elif path.endswith("fetchMaintenanceImpact"):
+            data = {"booking-a": self.impact}
+        return SimpleNamespace(header=_header(ok=self.failure != "header"), data=data)