Generate Python payload types from a schema derived from shared

.github/actions/python-package/action.yml

@@ -28,7 +28,7 @@ runs:
       working-directory: ${{ inputs.working-directory }}
       run: |
         python scripts/gen_contract.py
-        git diff --exit-code src/selenium_devtools/_contract.py
+        git diff --exit-code src/selenium_devtools/_contract.py src/selenium_devtools/_wire_types.py
 
     - name: 🧪 Unit tests
       shell: bash

.github/workflows/python.yml

@@ -2,30 +2,34 @@ name: Python Adapter
 
 # Triggered by packages/shared/src/** as well, because the adapter's wire
 # contract is GENERATED from it: scope names, the runner id, the collector route,
-# the worker query keys and the run-id env var. A rename there leaves the
-# committed `_contract.py` stale, and stale means the adapter keeps sending a name
-# nothing reads — a silent loss, not an error.
+# the worker query keys, the run-id env var, and every payload type, through
+# `wire-schema.json` (which vitest keeps in step with shared's src). A change
+# there leaves `_contract.py` or `_wire_types.py` stale, and stale means the
+# adapter keeps sending a name or a shape nothing reads — a silent loss, not an
+# error.
 #
 # This was previously excluded to keep a Python job off every PR that touches
 # shared. The objection was that a red Python job would block work unrelated to
-# this adapter; it does not hold, because the only thing a shared change can turn
-# red here is the drift check, and drift IS related — it means that PR broke this
-# contract. The tests do not read shared at all. Scoped to `src/` so a version
-# bump or a README edit under shared does not trigger it.
+# this adapter; it does not hold, because the only things a shared change can
+# turn red here are the drift check and the tests' wire-schema validation, and
+# both mean that PR broke this contract. Scoped to `src/` and the schema so a
+# version bump or a README edit under shared does not trigger it.
 on:
   push:
     branches:
       - main
     paths:
       - packages/selenium-devtools-py/**
       - packages/shared/src/**
+      - packages/shared/wire-schema.json
       - .github/workflows/python.yml
       - .github/workflows/python-release.yml
       - .github/actions/python-package/**
   pull_request:
     paths:
       - packages/selenium-devtools-py/**
       - packages/shared/src/**
+      - packages/shared/wire-schema.json
       - .github/workflows/python.yml
       - .github/workflows/python-release.yml
       - .github/actions/python-package/**

ARCHITECTURE.md

@@ -124,7 +124,7 @@ Imports from: `core`, `shared`, `selenium-webdriver` (peer). Does not import: ot
 
 ### `packages/selenium-devtools-py` — Python Selenium adapter
 
-The same contract from another language. Not a port of the JavaScript adapters: it shares no code with them, only the `{scope, data}` wire format, so nothing here imports `core` or `shared` — the parts of those it needs are **generated** into `src/selenium_devtools/_contract.py` by `scripts/gen_contract.py`, which doubles as a drift guard.
+The same contract from another language. Not a port of the JavaScript adapters: it shares no code with them, only the `{scope, data}` wire format, so nothing here imports `core` or `shared` — the parts of those it needs are **generated** by `scripts/gen_contract.py`, which doubles as a drift guard: constants and scope names into `src/selenium_devtools/_contract.py`, and payload TypedDicts into `_wire_types.py` from `packages/shared/wire-schema.json`. That schema is itself generated from shared's TypeScript types by `packages/shared/scripts/wire-schema.ts`, one JSON Schema per wire scope, so a language that cannot import shared still reads one source.
 
 Contains:
 

packages/app/test-ui/shell/fixtures.ts

@@ -146,7 +146,8 @@ export function traceLog(overrides: Partial<TraceLog> = {}): TraceLog {
     metadata: testrunnerMetadata,
     commands: loginCommands,
     sources: {},
-    suites: loginRun.frame,
+    // A fragment stands in for the full SuiteStats: the app reads only these fields.
+    suites: loginRun.frame as TraceLog['suites'],
     ...overrides
   }
 }

packages/app/tests/data-manager.test.ts

@@ -237,7 +237,8 @@ function traceLog(overrides: Partial<TraceLog> = {}): TraceLog {
     metadata: { sessionId: SESSION, type: TraceType.Testrunner },
     commands: [command()],
     sources: { '/specs/login.e2e.ts': 'await browser.url(url)' },
-    suites: suitesFrame(suite('login-suite')),
+    // A fragment stands in for the full SuiteStats: the app reads only these fields.
+    suites: suitesFrame(suite('login-suite')) as TraceLog['suites'],
     ...overrides
   }
 }

packages/selenium-devtools-py/README.md

@@ -426,8 +426,9 @@ backend this run reports to — and for trace mode, which opens none at all.
 src/selenium_devtools/
   __init__.py         public API — enable() / disable() / get_capturer()
   constants.py        defaults, env-var names, skip sets, pinned backend version
-  types.py            TypedDicts for the wire payloads (mirror packages/shared)
+  types.py            re-exports the wire payload TypedDicts, plus JSONValue/Scope
   _contract.py        GENERATED from packages/shared — scope names + CONTRACT_VERSION
+  _wire_types.py      GENERATED from packages/shared/wire-schema.json — payload TypedDicts
   utils.py            framework-agnostic helpers (now_ms, iso, to_jsonable, call_source)
   frames.py           pure builders for each {scope,data} payload
   transport.py        stdlib WebSocket client (handshake, masked frames, ping/pong, control reader)
@@ -455,8 +456,9 @@ src/selenium_devtools/
   lifecycle.py        dashboard window open/close + shutdown-on-disconnect
   rerun.py            launch/rerun commands the dashboard's run controls spawn
   pytest_plugin.py    CLI/ini config surface + suite/test tree feeder (opt-in)
-scripts/gen_contract.py   regenerate _contract.py from shared (dev-time; also a drift-guard)
-tests/                stdlib-unittest unit tests (no selenium/pytest needed)
+scripts/gen_contract.py   regenerate _contract.py + _wire_types.py from shared (dev-time; also a drift-guard)
+tests/                stdlib-unittest unit tests (no selenium/pytest needed); every
+                      sent frame is validated against shared's wire schema
 e2e_check.py          real-Chrome smoke (plain script)
 e2e/test_smoke.py     real-Chrome smoke (pytest + plugin)
 (example lives at repo root: examples/selenium-py/scripts/web_form.py)
@@ -471,7 +473,7 @@ coupling is handled explicitly rather than via a `workspace:^`-style resolver:
 |---|---|---|
 | **Adapter** (this package) | `pip install -e` | PyPI: `pip install selenium-devtools-py` |
 | **Backend + UI** (Node) | `node packages/backend/dist/server.js` | npm: `npx @wdio/devtools-backend@<pinned>` |
-| **Wire contract** (`shared`) | regenerated into `_contract.py` | the generated `_contract.py` ships in the wheel |
+| **Wire contract** (`shared`) | regenerated into `_contract.py` + `_wire_types.py` | both generated files ship in the wheel |
 
 `enable()` obtains the backend in this order (local vs published falls out of it):
 
@@ -486,11 +488,17 @@ no auto-resolution, so it's bumped deliberately alongside a contract change.
 Regenerate the contract after any change to `packages/shared`:
 
 ```bash
+pnpm --filter @wdio/devtools-shared gen:wire-schema   # TS types → wire-schema.json
 python3 packages/selenium-devtools-py/scripts/gen_contract.py
 ```
 
-It fails loudly if a scope the adapter needs disappeared from `shared` — a
-build-time drift alarm.
+The first step reads shared's payload types with the TypeScript compiler and
+writes one JSON Schema per wire scope; vitest fails when the committed copy is
+stale. The second turns it into TypedDicts and fails loudly if a scope the
+adapter needs disappeared from `shared`. The unit tests then validate every
+frame the adapter sends against that schema, so a field shared does not
+declare, a missing required one, or a null where shared expects absence all
+fail the build.
 
 ## Test
 
@@ -509,7 +517,7 @@ Two workflows, mirroring the JS split (`ci.yml` tests / `release.yml` publish):
 
 - **`python.yml`** — runs on PRs + pushes touching this package or `shared`:
   unit tests on Python 3.10 + 3.13, a contract-drift check (regenerate
-  `_contract.py`, fail on any diff), and a build + `twine check --strict` so a
+  `_contract.py` and `_wire_types.py`, fail on any diff), and a build + `twine check --strict` so a
   packaging mistake surfaces on the PR rather than under the publish button.
   Zero repo config needed.
 - **`python-release.yml`** — **manual** (`workflow_dispatch`, like the JS
@@ -580,6 +588,7 @@ above. What the JavaScript adapters have and this one does not:
 - **Capture never breaks tests.** Commands are recorded around the real call;
   errors are captured *and re-raised* unchanged; a missing dashboard is a no-op.
 - **Contract drift** is the main long-term risk (see the integration artifact).
-  Mitigated two ways: `_contract.py` is generated from `packages/shared` (scope
-  names + `CONTRACT_VERSION`), and the generator fails if a required scope
-  vanishes. Full field-level type generation is a future step.
+  Mitigated three ways: `_contract.py` is generated from `packages/shared` (scope
+  names + `CONTRACT_VERSION`), the generator fails if a required scope
+  vanishes, and the payload types are generated field by field from
+  `wire-schema.json`, which the unit tests validate every sent frame against.

packages/selenium-devtools-py/changes/generated-wire-types.md

@@ -0,0 +1,11 @@
+---
+minor
+---
+
+Frames now leave out an optional field instead of sending it as null:
+`callSource` on commands and tests, `url` on session metadata, `status` and
+`endTime` on a network request still in flight, and `state` on an unfinished
+suite. A native session's viewport now carries `offsetLeft: 0`, `offsetTop: 0`
+and `scale: 1`, as the JavaScript adapters send it. The payload types in
+`selenium_devtools.types` are now generated from the dashboard's own schema;
+`ElementScripts` is renamed to `ElementScriptsResponse`.

packages/selenium-devtools-py/scripts/gen_contract.py

@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import json
+import keyword
 import re
 import sys
 from pathlib import Path
@@ -200,6 +201,98 @@ def _native_platforms(device_ts: str) -> list[str]:
     return re.findall(r"'([^']+)'", m.group(1))
 
 
+class _TypedDictWriter:
+    """Renders ``wire-schema.json``'s ``$defs`` as TypedDicts.
+
+    Required and optional keys are split across a base class and a
+    ``total=False`` subclass because ``typing.Required`` needs Python 3.11 and
+    the package supports 3.10. An inline object becomes a class named after its
+    owner and key, since a TypedDict field cannot hold an anonymous shape.
+    """
+
+    def __init__(self, defs: dict) -> None:
+        self._defs = defs
+        self._classes: dict[str, str] = {}
+
+    def render(self) -> str:
+        for name, schema in self._defs.items():
+            self._class(name, schema)
+        names = sorted(self._classes)
+        lines = [
+            "# GENERATED by scripts/gen_contract.py from packages/shared/wire-schema.json.",
+            "# Do not edit by hand — run the script to regenerate.",
+            "from __future__ import annotations",
+            "",
+            "from typing import Any, Dict, List, Literal, Optional, TypedDict, Union",
+            "",
+            f"__all__ = {names!r}",
+        ]
+        for name in names:
+            lines += ["", "", self._classes[name]]
+        return "\n".join(lines) + "\n"
+
+    def _class(self, name: str, schema: dict) -> None:
+        if name in self._classes:
+            raise SystemExit(f"wire schema: two shapes both want the name {name}")
+        self._classes[name] = ""
+        props = schema.get("properties", {})
+        required = set(schema.get("required", []))
+        fields = {}
+        for key, prop in props.items():
+            if not key.isidentifier() or keyword.iskeyword(key):
+                raise SystemExit(f"wire schema: {name}.{key} is not a Python identifier")
+            fields[key] = self._type(prop, name + key[:1].upper() + key[1:])
+        req = [f"    {k}: {t}" for k, t in fields.items() if k in required]
+        opt = [f"    {k}: {t}" for k, t in fields.items() if k not in required]
+        if req and opt:
+            body = (
+                f"class _{name}Required(TypedDict):\n" + "\n".join(req) + "\n\n\n"
+                f"class {name}(_{name}Required, total=False):\n" + "\n".join(opt)
+            )
+        elif req:
+            body = f"class {name}(TypedDict):\n" + "\n".join(req)
+        else:
+            body = f"class {name}(TypedDict, total=False):\n" + ("\n".join(opt) or "    pass")
+        self._classes[name] = body
+
+    def _type(self, schema: dict, inline_name: str) -> str:
+        if "$ref" in schema:
+            return schema["$ref"].rsplit("/", 1)[-1]
+        if "anyOf" in schema:
+            parts = [self._type(s, inline_name) for s in schema["anyOf"]]
+            rest = [p for p in parts if p != "None"]
+            inner = rest[0] if len(rest) == 1 else f"Union[{', '.join(rest)}]"
+            return f"Optional[{inner}]" if len(rest) < len(parts) else inner
+        if "enum" in schema:
+            return f"Literal[{', '.join(repr(v) for v in schema['enum'])}]"
+        if "const" in schema:
+            return f"Literal[{schema['const']!r}]"
+        kind = schema.get("type")
+        if kind is None:
+            return "Any"
+        if kind == "array":
+            return f"List[{self._type(schema['items'], inline_name)}]"
+        if kind == "object":
+            if "properties" in schema:
+                self._class(inline_name, schema)
+                return inline_name
+            extra = schema.get("additionalProperties")
+            value = self._type(extra, inline_name) if isinstance(extra, dict) else "Any"
+            return f"Dict[str, {value}]"
+        primitives = {"string": "str", "number": "float", "boolean": "bool", "null": "None"}
+        if kind not in primitives:
+            raise SystemExit(f"wire schema: unsupported type {kind!r} for {inline_name}")
+        return primitives[kind]
+
+
+def _wire_types(schema_path: Path) -> str:
+    if not schema_path.exists():
+        raise SystemExit(
+            f"{schema_path} is missing: run `pnpm --filter @wdio/devtools-shared gen:wire-schema`"
+        )
+    return _TypedDictWriter(json.loads(schema_path.read_text())["$defs"]).render()
+
+
 def main() -> int:
     root = _repo_root()
     shared = root / "packages" / "shared"
@@ -320,10 +413,14 @@ def main() -> int:
         f'ENV_LAUNCH_COMMAND = "{reuse_env["LAUNCH_COMMAND"]}"',
         "",
     ]
-    out = shared.parent / "selenium-devtools-py" / "src" / "selenium_devtools" / "_contract.py"
+    pkg = shared.parent / "selenium-devtools-py" / "src" / "selenium_devtools"
+    out = pkg / "_contract.py"
     out.write_text("\n".join(lines))
     print(f"wrote {out.relative_to(root)}  (contract v{version}, "
           f"{len(data_keys)} data scopes, {len(control)} control scopes)")
+    wire = pkg / "_wire_types.py"
+    wire.write_text(_wire_types(shared / "wire-schema.json"))
+    print(f"wrote {wire.relative_to(root)}")
     return 0