Tolerate objects that do not fit the model in list() and watch()

CHANGELOG.md

@@ -21,6 +21,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `kubex_core.models.types`: `Time`, `MicroTime`, `Quantity`, `NullAsEmptyList`,
   `NullAsEmptyDict`.
 - `ObjectMetadata.managed_fields`/`self_link` and `ListMetadata.shard_info`.
+- `invalid=InvalidObjects.INCLUDE` for `Api.list()` and `Api.watch()`: objects that do not fit
+  the model come back as `InvalidObject`s (in `ListWithInvalid.invalid`, or as
+  `InvalidWatchEvent`s) instead of raising.
+- `WatchEvent.is_initial_events_end`.
 
 ### Fixed
 

CLAUDE.md

@@ -99,13 +99,16 @@ kubex/                          # Main package — PEP 420 namespace package (no
     ├── response.py             # Response dataclass + HeadersWrapper
     ├── params.py               # API option classes (ListOptions, GetOptions, DeleteOptions, ExecOptions, AttachOptions, PortForwardOptions, etc.)
                                 #   + Timeout, TimeoutTypes — HTTP timeout configuration
+                                #   + InvalidObjects enum (RAISE/INCLUDE) for list()/watch(invalid=)
+    ├── invalid_object.py       # InvalidObject (raw + ValidationError + parsed ObjectMetadata) and ListWithInvalid(ListEntity)
+                                #   — what list()/watch(invalid=InvalidObjects.INCLUDE) return for objects that do not fit the model
     ├── json_patch.py           # JSON Patch RFC 6902 operation models (JsonPatchAdd, etc.) + JsonPatch RootModel
     ├── json_pointer.py         # JSON Pointer RFC 6901 implementation (JsonPointer custom str type)
     ├── patch.py                # Patch protocol + ApplyPatch, MergePatch, StrategicMergePatch
                                 #   Re-exports JsonPatch models and JsonPointer from json_patch/json_pointer
     ├── subresource.py          # Subresource definitions
     ├── exec_channels.py        # WebSocket channel protocol (V5ChannelProtocol, channel ID constants, select_protocol()) + portforward helpers (data_channel_for_port_index, error_channel_for_port_index, port_prefix_encode/decode)
-    ├── watch_event.py          # WatchEvent[ResourceType] and EventType enum
+    ├── watch_event.py          # WatchEvent[ResourceType] (+ is_initial_events_end), InvalidWatchEvent and EventType enum
     └── request_builder/        # Constructs HTTP requests from API calls
         ├── _paths.py           # path_segment()/object_url()/collection_url() — the only way builders build URLs:
                                 #   validates (no empty/None name — that would hit the collection URL) and percent-encodes names/namespaces
@@ -172,6 +175,7 @@ test/                           # Test suite
 │   ├── test_portforward.py     # E2E tests for Pod portforward subresource (forward() + listen() against K3S)
 │   ├── test_bearer_token_auth.py # E2E tests for bearer-token auth (no proxy): valid SA token succeeds, invalid token returns 401
 │   ├── test_api_safety.py      # E2E: empty-name rejection, dry-run delete, escaped names, server-side apply
+│   ├── test_invalid_objects.py # E2E: list()/watch(invalid=INCLUDE) with a CRD object the model rejects; initial-events-end bookmark
 │   └── proxy/                  # E2E tests for trust_env proxy support (Squid on shared Docker network)
 │       ├── _constants.py       # SQUID_PROXY_USER / SQUID_PROXY_PASSWORD shared by conftest and tests
 │       ├── conftest.py         # Fixtures: Docker network, K3S-in-network, Squid, proxy_url, proxy config,
@@ -190,6 +194,7 @@ test/                           # Test suite
 │       └── test_refreshable_token.py # Token refresh and async lock tests
 ├── test_error_handling.py      # handle_request_error() and exception hierarchy tests
 ├── test_metadata_accessor.py   # MetadataAccessor (get, list, patch) tests
+├── test_invalid_objects.py     # list()/watch(invalid=) parsing rules, is_initial_events_end, overload types (assert_type)
 ├── test_api_request_validation.py # Api-level name/namespace validation (nothing reaches the client)
 ├── test_models/                # Unit tests for kubex-core models, field types and generated-model behaviour
 │   └── test_eviction.py        # Eviction model tests
@@ -300,6 +305,9 @@ All models inherit from `BaseK8sModel` which uses `alias_generator=to_camel`, `p
 ### Configuration auto-loading
 `create_client()` → tries kubeconfig file first → falls back to in-cluster pod environment.
 
+### Objects that do not fit the model
+`Api.list()` and `Api.watch()` take `invalid: InvalidObjectsTypes` (`InvalidObjects` enum or `"raise"`/`"include"`); overloads pick the return type. `RAISE` (default) raises pydantic's `ValidationError`. `INCLUDE` returns `ListWithInvalid` (`items` + `invalid: list[InvalidObject]`) or yields `InvalidWatchEvent` and keeps streaming. Only ADDED/MODIFIED/DELETED objects are tolerated, and only when their metadata parses with a name; ERROR events, unparseable bookmarks and non-JSON lines still raise. `MetadataAccessor` has no such option (`PartialObjectMetadata` has only optional fields). The watch loop parses before `yield`, so the yield is never inside an `except`.
+
 ### Ellipsis sentinel for optional overrides
 `Ellipsis` (`...`) is used as a sentinel to distinguish "not provided" (use the default) from `None` (explicitly disabled). This pattern is used in two places:
 - **Namespace**: `...` = use the `Api` instance default; `None` = explicitly no namespace.

docs/operations/crud.md

@@ -60,6 +60,22 @@ Pass `namespace=None` to list across all namespaces even when the `Api` instance
 all_pods = await api.list(namespace=None)
 ```
 
+### Objects that do not fit the model
+
+By default, `list()` raises pydantic's `ValidationError` if any item does not fit the model. Pass `invalid=InvalidObjects.INCLUDE` (or the string `"include"`) to get a `ListWithInvalid` instead: `items` holds the objects that fit, and `invalid` the others, each an `InvalidObject` with `raw`, `error` and `metadata`:
+
+```python
+from kubex.core.params import InvalidObjects
+
+pods = await api.list(invalid=InvalidObjects.INCLUDE)
+for pod in pods.items:
+    print(pod.metadata.name)
+for bad in pods.invalid:
+    print(f"{bad.metadata.name} does not fit the model: {bad.error}")
+```
+
+`ListWithInvalid` is a `ListEntity`, so `metadata.resource_version` and `metadata.continue_` work as usual; `invalid` is not serialized. The list itself is still checked strictly, and an item without a readable `metadata.name` still raises. See [Objects that do not fit the model](watch.md#objects-that-do-not-fit-the-model) for when this happens and what other calls do with such objects.
+
 ## create
 
 Create a new resource. The returned object is the server-assigned representation (including `metadata.name` when `generate_name` was used):

docs/operations/watch.md

@@ -24,6 +24,7 @@ Each yielded value is a `WatchEvent[ResourceType]`:
 |---|---|---|
 | `event.type` | `EventType` | `ADDED`, `MODIFIED`, `DELETED`, or `BOOKMARK` |
 | `event.object` | `ResourceType | Bookmark` | Fully parsed resource (or `Bookmark` for bookmark events) |
+| `event.is_initial_events_end` | `bool` | `True` only for the bookmark that ends the initial events (see [`send_initial_events` pattern](#send_initial_events-pattern)) |
 
 `ERROR` events are never yielded: the generator raises the matching exception instead — most importantly `Gone` (HTTP 410) when the requested `resourceVersion` is too old. See [Resuming after `Gone`](#resuming-after-gone).
 
@@ -75,17 +76,22 @@ async for event in api.watch(namespace=None):
 
 `send_initial_events=True` makes the stream start with synthetic `ADDED` events for every existing resource, then a `BOOKMARK` annotated with `k8s.io/initial-events-end: "true"`, then live updates — current state and future changes in one stream. The server requires bookmarks for this, so `allow_bookmarks` defaults to `True` when `send_initial_events=True` (passing `allow_bookmarks=False` together with it raises `ValueError`). It needs the API server's `WatchList` feature; a server without it rejects the request with `422`.
 
+`event.is_initial_events_end` is `True` only for that bookmark. Periodic checkpoint bookmarks look the same apart from the annotation, so they return `False`:
+
 ```python
 async for event in api.watch(send_initial_events=True):
-    if event.type == EventType.BOOKMARK:
-        annotations = event.object.metadata.annotations or {}
-        if annotations.get("k8s.io/initial-events-end") == "true":
-            print("initial state complete, watching for changes now")
-        continue
-    print(event.type, event.object.metadata.name)
+    if event.is_initial_events_end:
+        print("initial state complete, watching for changes now")
+    elif event.type != EventType.BOOKMARK:
+        print(event.type, event.object.metadata.name)
 ```
 
-Not every bookmark marks the end of the initial events — periodic checkpoint bookmarks look the same apart from the annotation.
+### Where to resume from
+
+The initial `ADDED` events carry each object's own `resourceVersion`, in no particular order, so none of them is a point to resume from. The end bookmark's `resourceVersion` is the first one:
+
+- If the stream ends before the end bookmark, discard what you collected and start again with `send_initial_events=True`.
+- After it, resume with `watch(resource_version=...)` from the last `resourceVersion` you saw (an event's or a bookmark's), as in [Resuming after `Gone`](#resuming-after-gone).
 
 ## Resuming after `Gone`
 
@@ -123,6 +129,36 @@ async def watch_forever(api: Api[Pod]) -> None:
 
 Events between the list and the watch are not lost: watching from the list's `resourceVersion` delivers everything that changed after it. Changes that happened while the watcher was disconnected for longer than the server keeps history are covered by the re-list, so `handle_existing` should reconcile (update or create local state, and drop objects that no longer exist) rather than assume a clean slate.
 
+## Objects that do not fit the model
+
+By default, an object the model cannot parse raises pydantic's `ValidationError`, which ends the stream. This mostly happens with custom resources: the API server does not re-validate objects stored before a CRD schema change, so an old object can break a model written for the current schema. Reconnecting does not help, because the same object comes back.
+
+Pass `invalid=InvalidObjects.INCLUDE` (or the string `"include"`) to keep going. Such an object then arrives as an `InvalidWatchEvent` instead of a `WatchEvent`:
+
+| Attribute | Type | Description |
+|---|---|---|
+| `event.type` | `EventType` | `ADDED`, `MODIFIED` or `DELETED` |
+| `event.object.raw` | `dict[str, Any]` | The object as the server sent it |
+| `event.object.error` | `ValidationError` | Why it does not fit the model |
+| `event.object.metadata` | `ObjectMetadata` | Its metadata; always has a `name` |
+
+```python
+from kubex.core.params import InvalidObjects
+from kubex.core.watch_event import InvalidWatchEvent
+
+async for event in api.watch(invalid=InvalidObjects.INCLUDE):
+    if isinstance(event, InvalidWatchEvent):
+        print(f"skipping {event.object.metadata.name}: {event.object.error}")
+        continue
+    handle(event)
+```
+
+Both kinds of event have `type` and `object.metadata`, so code that only needs an object's name, namespace or `resourceVersion` (keeping a resume point, or dropping a deleted object from a cache) works on both without the `isinstance` check.
+
+Only `ADDED`, `MODIFIED` and `DELETED` objects are tolerated. These still raise, because they mean the stream itself is broken: an `ERROR` event, a bookmark that does not parse, an object without a readable `metadata.name`, and a line that is not JSON.
+
+`list()` takes the same option; see [Objects that do not fit the model](crud.md#objects-that-do-not-fit-the-model) for lists. Other calls still parse the object with the model: `get()` raises `ValidationError` for it, and so do `delete()`, `patch()` and `replace()`, after the server has carried out the change. Use [`api.metadata`](../subresources/metadata.md), which only parses metadata (for example, to remove a finalizer), or a looser model for those.
+
 ## Server-side timeout
 
 Pass `timeout_seconds=` to set a server-side timeout on the watch call (sent as the Kubernetes `timeoutSeconds` query parameter). The server closes the stream cleanly after this many seconds and the `async for` loop simply ends; reconnect as shown above:

docs/reference/core.md

@@ -20,6 +20,10 @@ Auto-generated reference for the `kubex.core` module.
 
 ::: kubex.core.watch_event
 
+## Objects that do not fit the model
+
+::: kubex.core.invalid_object
+
 ## Patch types
 
 ::: kubex.core.patch