Provider contract

Contract providers implement so the orchestrator can snapshot, plan, and apply changes safely.

Providers are the integrations CrossWatch talks to.

Examples: Plex, Jellyfin, Emby, Trakt, SIMKL, MDBList, AniList.

The “provider contract” is why CrossWatch can treat them consistently.

Why you should care

Provider health decides whether CrossWatch writes or skips.
Provider snapshots decide what planner thinks “changed”.
Provider write responses decide whether applier marks items as confirmed or unresolved.

What you will notice when a provider breaks

auth_failed health → the pair is skipped.
down health → writes are skipped.
empty or stale snapshots → guardrails may block removals.
weak write confirmations → lots of “skipped” or “unresolved”.

How provider inventories become snapshots: Snapshots
How diffs are produced: Planner
How writes are executed: Applier
Why CrossWatch may refuse deletes: Guardrails

This doc explains the provider interface the orchestrator expects.

If providers implement it correctly, runs stay predictable, safe, and debuggable.

Code:

cw_platform/orchestrator/_types.py (InventoryOps)
cw_platform/orchestrator/_providers.py (loader)

The big picture

The orchestrator does not “know” Plex, SIMKL, Trakt, etc.

It knows InventoryOps objects.

Each provider module exports an ops instance (usually OPS or ADAPTER) that implements a contract:

identify itself (name, label)
declare supported features (features)
declare capability flags (capabilities)
build indices (build_index)
apply writes (add, remove)
optionally report health (health)
optionally provide activity checkpoints (activities)

That is it.

Required methods

`name() -> str`

Canonical provider id used in config and state:

"PLEX", "SIMKL", "TRAKT", "JELLYFIN", etc.

Must be stable and unique.

`label() -> str`

Human-readable name used in UI logs:

"Plex"
"SIMKL"
etc.

`features() -> dict[str, bool]`

Return which features are supported:

Example:

{
  "watchlist": true,
  "ratings": true,
  "history": true,
  "playlists": false
}

The orchestrator uses this to skip snapshot building and feature runs early.

`capabilities() -> dict[str, Any]`

Extra flags used for behavior switches.

Common keys the orchestrator looks at:

features: optional per-feature override (same idea as features())
index_semantics: "present" (default) or "delta"
verify_after_write: bool
observed_deletes: bool
checkpoint: bool or string (provider-defined)
feature-specific checkpoint hints (optional)

If you don’t provide capabilities, the orchestrator assumes safest defaults:

present snapshots
no verify-after-write support
observed deletes allowed only in two-way logic after guards

`build_index(cfg: dict, feature: str) -> dict | list`

Must return the provider’s snapshot for that feature.

Allowed return shapes:

dict mapping provider keys → item dicts
list of item dicts

Orchestrator will normalize into:

canonical keys using cw_platform.id_map.canonical_key(item)

Minimum recommended item fields:

type: movie|show|season|episode
title (optional but helpful)
year (optional but helpful)
ids: dict with at least one stable ID (imdb, tmdb, tvdb, simkl, trakt, etc.)

Provider-specific write payload can live under:

item["plex"], item["simkl"], item["trakt"], item["jellyfin"], etc.

`add(cfg: dict, items: list[dict], feature: str, dry_run: bool) -> dict`

Applies adds/upserts on the provider.

For ratings: add = upsert rating
For history: add = mark watched / add play record (provider-dependent)
For playlists: add = add item to playlist

Return a result dict compatible with Applier.

`remove(cfg: dict, items: list[dict], feature: str, dry_run: bool) -> dict`

Applies removals/unrates on the provider.

Return a result dict compatible with Applier.

Optional but strongly recommended

`health(cfg: dict, emit: callable | None = None) -> dict`

Health reports drive run gating and UI visibility.

Recommended response:

{
  "ok": true,
  "status": "ok",          // ok | down | auth_failed | degraded
  "features": { "watchlist": true, "ratings": true, ... },
  "latency_ms": 120,
  "api": { "calls": 8, "errors": 0, "rate_limited": 0 }
}

If you can, emit intermediate events:

emit("api:hit", provider="SIMKL", op="GET /sync/activities", ms=123, ok=true)

If health is missing, the orchestrator assumes:

auth OK
provider up, which is less safe.

`activities(cfg: dict) -> dict[str, str]`

Used to compute checkpoints (module_checkpoint()).

Return a mapping of feature→timestamp/string:

{
  "watchlist": "2026-01-28T21:00:00Z",
  "ratings": "2026-01-28T20:10:00Z",
  "history": "2026-01-28T18:00:00Z",
  "updated_at": "2026-01-28T21:00:00Z"
}

If you return this, drop guard becomes much better at detecting “bad empty snapshot”.

Index semantics: present vs delta

Present (default)

build_index returns the full current inventory for the feature.

Orchestrator treats it as authoritative and can apply drop guard.

Delta

build_index returns only “changes since last time” (or some partial view).

Set:

capabilities.index_semantics = "delta"

Orchestrator will merge:

effective = baseline ∪ current_delta

Drop guard is not applied to delta snapshots (they’re not expected to resemble full inventory).

Use delta semantics when the provider API:

can’t cheaply return full lists
or is rate-limited and you do incremental updates (Simkl activities + date_from patterns, etc.)

Verify-after-write support

If you can confirm what actually changed after a write, set:

capabilities.verify_after_write = true

Then the orchestrator may:

downgrade “confirmed” items back to unresolved if verification fails (depending on pipeline wiring)

Practically, verify-after-write is useful when the provider:

accepts the call but silently ignores items (common with bad IDs or library filters)

Good provider write responses (so applier + blackbox work)

Best-case response:

{
  "ok": true,
  "confirmed_keys": ["imdb:tt123", "tmdb:456"],
  "unresolved": []
}

Second best:

{ "ok": true, "confirmed": 2, "unresolved": [ { "ids": { "imdb": "tt999" } } ] }

Avoid:

{ "ok": true } with no counts: it looks like a no-op
unresolved items without IDs: orchestrator can’t track them, so they’ll get retried forever

Persistence flags (items you don’t want in baselines)

The orchestrator skips persisting items when:

_cw_transient == true
_cw_skip_persist == true
_cw_persist == false

Use these on provider-produced index items when:

they are ephemeral API constructs
they will explode baseline size with useless data
they aren’t stable across runs

Also, if you set provider subobject:

item["plex"]["ignored"] == true (or equivalent) the baseline persistence step will skip it.

Context injection (ops.ctx)

The orchestrator tries to set ops.ctx = ctx and/or module.ctx = ctx.

If you use ctx, expect:

ctx.emit / ctx.dbg (logging)
ctx.snap_cache / TTLs
ctx.state_store (sometimes)
ctx.feature / ctx.pair env hints

Providers should not require ctx to exist, but can use it for:

better logging
caching within a run
exposing api hit samples

Implementation checklist

If you’re adding a new provider:

Implement name/label/features/capabilities.
Implement build_index for each supported feature.
Implement add/remove for each supported feature.
Add health and activities if possible.
Ensure every item has ids with at least one stable ID.
Return confirmed_keys for writes whenever possible.

Do this and the orchestrator will behave like a machine, not a mystery.

Orchestrator overview: Orchestrator
Snapshot requirements: Snapshots
Write response expectations: Applier
Diff expectations: Planner
Safety mechanisms you need to support: Guardrails

PreviousState.json NextProvider specifics

Last updated 4 months ago

Was this helpful?

Good night

Why you should care

What you will notice when a provider breaks

The big picture

Required methods

name() -> str

label() -> str

features() -> dict[str, bool]

capabilities() -> dict[str, Any]

build_index(cfg: dict, feature: str) -> dict | list

add(cfg: dict, items: list[dict], feature: str, dry_run: bool) -> dict

remove(cfg: dict, items: list[dict], feature: str, dry_run: bool) -> dict