fix(embed): conformance mock + TypeScript router vs the real Monad API [PRO-455]

conformance/mock_monad.py

@@ -11,26 +11,92 @@
 - ``GET /v1/{kind}s`` (catalog)      → a bare array of connector types.
 - ``GET /v1/{org}/{kind}s``          → ``{ "<kind>s": [...] | null, "pagination": {...} }``
                                         (the list is **null**, not ``[]``, when empty).
+- ``GET /v1/{org}/{kind}s/{id}``     → the connector record + ``component_of``,
+                                        the pipelines it is a node of.
 - ``POST /v2/{org}/pipelines``       → **201** with the full pipeline object.
 - ``GET  /v2/{org}/pipelines``       → ``{ "pipelines": [...], "pagination": {...} }``.
 - ``GET  /v2/{org}/pipelines/{id}``  → the full pipeline object at top level
                                         (``nodes``/``edges``/``enabled`` — no ``config`` wrapper).
 - ``POST /v2/{org}/outputs``         → the created output object.
 - ``POST /v3/sessions``              → ``{ "session_token", "expires_at" }``.
+
+**Pagination is enforced, and the default limit is 10** — matching every list
+handler in Monad (``api/pkg/routes/v2/routes/pipelines.go:727`` and friends).
+A router that omits ``limit`` gets 10 rows here exactly as it would in
+production, which is what turns the "only ever sees 10" class of bug into a
+test failure instead of a live incident.
 """
 
 from __future__ import annotations
 
 import json
 import re
 import threading
+import urllib.parse
 from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 
+# Monad's list handlers all default to 10 and enforce no maximum.
+DEFAULT_LIMIT = 10
+
+# Enough configured inputs that a router which does not page sees a truncated
+# list. `test_connectors_pagination_is_exhaustive` pins the full count.
+_SEEDED_INPUTS = 12
+
+# Inputs the bulk-pipeline scenario wires, one pipeline each, to push the
+# pipeline list past its first page.
+_BULK_INPUTS = 12
+
+
+def _seed_connectors() -> dict[str, dict[str, dict]]:
+    """The tenant's configured connectors, keyed by id.
+
+    `outputs` is deliberately empty so `GET /v1/{org}/outputs` returns **null**
+    — the real-Monad behavior its own spec gets wrong, and the reason live mode
+    exists. `out_store`/`out_conf` are the host's pre-provisioned components:
+    resolvable by id, but not part of the tenant's own list.
+    """
+    inputs = {
+        f"cfg_{i}": {
+            "id": f"cfg_{i}",
+            "type": "aws-cloudtrail" if i % 2 else "okta-systemlog",
+            "name": f"Configured {i}",
+            "organization_id": "org_conf",
+            "config": {"settings": {}, "secrets": {}},
+        }
+        for i in range(1, _SEEDED_INPUTS + 1)
+    }
+    # Ids the lifecycle scenarios wire, resolvable but outside the listed page
+    # so they cannot perturb the pagination counts. `bulk_*` exists to push the
+    # pipeline list past its first page.
+    extras = ["in_conf", "in_dup_conflict"] + [f"bulk_{i}" for i in range(1, _BULK_INPUTS + 1)]
+    for extra in extras:
+        inputs[extra] = {
+            "id": extra,
+            "type": "aws-cloudtrail",
+            "name": extra,
+            "organization_id": "org_conf",
+            "config": {"settings": {}, "secrets": {}},
+            "listed": False,
+        }
+    outputs = {
+        oid: {
+            "id": oid,
+            "type": "dev-null",
+            "name": oid,
+            "organization_id": "org_conf",
+            "config": {"settings": {}, "secrets": {}},
+            "listed": False,
+        }
+        for oid in ("out_store", "out_conf")
+    }
+    return {"input": inputs, "output": outputs}
+
 
 class _State:
     def __init__(self) -> None:
         self.lock = threading.Lock()
         self.pipelines: dict[str, dict] = {}
+        self.connectors = _seed_connectors()
         self._counter = 0
 
     def new_id(self, prefix: str) -> str:
@@ -39,6 +105,16 @@ def new_id(self, prefix: str) -> str:
             return f"{prefix}_{self._counter}"
 
 
+def _paginate(rows: list, limit: int, offset: int) -> tuple[list | None, dict]:
+    """Slice `rows` the way Monad does, and build the `pagination` sibling.
+
+    An empty page comes back as ``None``, not ``[]`` — see the module note.
+    `total` is the full count, not the page length.
+    """
+    page = rows[offset : offset + limit]
+    return (page or None), {"limit": limit, "offset": offset, "total": len(rows)}
+
+
 def _component_ids(pipeline: dict) -> frozenset:
     """The set of component ids a pipeline wires — used to detect a duplicate
     connection (Monad rejects connecting the same components twice → 409)."""
@@ -84,11 +160,14 @@ def _pipeline_summary(p: dict) -> dict:
 
 
 def _output_view(oid: str, body: dict) -> dict:
+    # `type` is the canonical field; `output_type` is the deprecated alias the
+    # API still accepts, with `type` winning when both are sent
+    # (api/pkg/routes/v2/routes/organization_outputs.go:188).
     return {
         "id": oid,
         "name": body.get("name", ""),
         "description": body.get("description", ""),
-        "type": body.get("output_type", "dev-null"),
+        "type": body.get("type") or body.get("output_type") or "dev-null",
         "organization_id": "org_conf",
         "managed_by": "",
         "config": {"settings": {}, "secrets": {}},
@@ -97,6 +176,36 @@ def _output_view(oid: str, body: dict) -> dict:
     }
 
 
+def _connector_view(record: dict) -> dict:
+    """A configured connector as the list/detail endpoints return it."""
+    return {k: v for k, v in record.items() if k != "listed"}
+
+
+def _component_of(pipelines: dict[str, dict], component_id: str) -> list[dict]:
+    """The pipelines a component is a node of — Monad's `component_of`.
+
+    Deliberately the same narrow projection the datastore emits
+    (core/pkg/datastore/postgres/pipelines.go:292): **no `status`, no `nodes`**.
+    A router that needs the peer component has to fetch the pipeline detail.
+    """
+    return [
+        {
+            "id": p.get("id"),
+            "organization_id": "org_conf",
+            "name": p.get("name", ""),
+            "description": p.get("description", ""),
+            "enabled": bool(p.get("enabled")),
+            "component_tier": 0,
+            "input_id": "",
+            "managed_by": "",
+            "created_at": "2026-01-01T00:00:00Z",
+            "updated_at": "2026-01-01T00:00:00Z",
+        }
+        for p in pipelines.values()
+        if component_id in _component_ids(p)
+    ]
+
+
 def _make_handler(state: _State):
     class Handler(BaseHTTPRequestHandler):
         def log_message(self, *_args) -> None:  # silence per-request logging
@@ -133,6 +242,20 @@ def do_PATCH(self):  # noqa: N802
         def do_DELETE(self):  # noqa: N802
             self._route("DELETE")
 
+        def _limit_offset(self) -> tuple[int, int]:
+            """Parse `limit`/`offset` exactly as Monad's handlers do: anything
+            unparseable or out of range falls back to the default."""
+            query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
+
+            def _int(name: str, default: int, floor: int) -> int:
+                try:
+                    value = int(query.get(name, [""])[0])
+                except (TypeError, ValueError):
+                    return default
+                return value if value >= floor else default
+
+            return _int("limit", DEFAULT_LIMIT, 1), _int("offset", 0, 0)
+
         def _route(self, method: str) -> None:
             path = self.path.split("?", 1)[0]
             body = self._read() if method in ("POST", "PATCH") else {}
@@ -166,35 +289,49 @@ def _route(self, method: str) -> None:
                 )
 
             # A tenant's configured connectors — wrapped alongside `pagination`,
-            # and returned as null (not []) when the tenant has none. Exercise
-            # both shapes: a populated `inputs`, an empty (null) `outputs`.
+            # paged with Monad's default limit of 10, and returned as null (not
+            # []) when the page is empty. Exercise both shapes: a populated
+            # `inputs` that spans two pages, an empty (null) `outputs`.
             m = re.fullmatch(r"/v1/([^/]+)/(inputs|outputs)", path)
             if method == "GET" and m:
                 kind = m.group(2)
-                rows = (
-                    None
-                    if kind == "outputs"
-                    else [
-                        {
-                            "id": "cfg_1",
-                            "type": "aws-cloudtrail",
-                            "name": "Configured",
-                            "organization_id": "org_conf",
-                            "config": {"settings": {}, "secrets": {}},
-                        }
+                limit, offset = self._limit_offset()
+                with state.lock:
+                    all_rows = [
+                        _connector_view(c)
+                        for c in state.connectors[kind[:-1]].values()
+                        if c.get("listed", True)
                     ]
-                )
-                return self._send(
-                    200,
-                    {
-                        kind: rows,
-                        "pagination": {"limit": 1000, "offset": 0, "total": 0 if rows is None else 1},
-                    },
-                )
+                rows, pagination = _paginate(all_rows, limit, offset)
+                return self._send(200, {kind: rows, "pagination": pagination})
+
+            # A single configured connector + `component_of`: the pipelines it is
+            # a node of. This is what lets a router resolve a connector's
+            # pipeline in one call instead of walking every pipeline in the org.
+            m = re.fullmatch(r"/v1/([^/]+)/(inputs|outputs)/([^/]+)", path)
+            if m and method in ("GET", "DELETE"):
+                kind, cid = m.group(2)[:-1], m.group(3)
+                with state.lock:
+                    record = state.connectors[kind].get(cid)
+                    if record is None:
+                        return self._send(404, {"error": f"{kind} not found"})
+                    if method == "DELETE":
+                        del state.connectors[kind][cid]
+                        return self._send(204)
+                    view = {
+                        **_connector_view(record),
+                        "component_of": _component_of(state.pipelines, cid),
+                    }
+                return self._send(200, view)
 
             # Create an output (e.g. the dev/null sink) — the created record.
+            # It joins the tenant's connectors so it is resolvable by id and
+            # deletable, like any other output.
             if method == "POST" and re.fullmatch(r"/v2/([^/]+)/outputs", path):
-                return self._send(200, _output_view(state.new_id("out"), body))
+                view = _output_view(state.new_id("out"), body)
+                with state.lock:
+                    state.connectors["output"][view["id"]] = {**view, "listed": False}
+                return self._send(200, view)
 
             # pipelines collection (create / list) — check before status/detail
             if re.fullmatch(r"/v2/([^/]+)/pipelines/?", path):
@@ -214,15 +351,11 @@ def _route(self, method: str) -> None:
                     # Real Monad returns 201 with the full pipeline record.
                     return self._send(201, view)
                 if method == "GET":
+                    limit, offset = self._limit_offset()
                     with state.lock:
                         items = [_pipeline_summary(p) for p in state.pipelines.values()]
-                    return self._send(
-                        200,
-                        {
-                            "pipelines": items,
-                            "pagination": {"limit": 1000, "offset": 0, "total": len(items)},
-                        },
-                    )
+                    page, pagination = _paginate(items, limit, offset)
+                    return self._send(200, {"pipelines": page or [], "pagination": pagination})
 
             if method == "GET" and re.fullmatch(r"/v2/([^/]+)/pipelines/([^/]+)/status", path):
                 pid = path.rsplit("/", 2)[-2]
@@ -259,9 +392,6 @@ def _route(self, method: str) -> None:
                         state.pipelines.pop(pid, None)
                     return self._send(204)
 
-            if method == "DELETE" and re.fullmatch(r"/v1/([^/]+)/(inputs|outputs)/([^/]+)", path):
-                return self._send(204)
-
             return self._send(404, {"error": f"mock unhandled {method} {path}"})
 
     return Handler

conformance/monad_schemas.py

@@ -18,6 +18,11 @@
   ``null`` (not ``[]``) for a tenant with none, though its spec types it
   ``array``. Live conformance is what proves this, but pinning it here stops the
   mock from regressing to a non-null-emitting shape.
+- ``component_of`` entries are pinned *without* ``status`` or ``nodes``: the
+  datastore projection behind them fills only a subset of ``models.Pipeline``
+  (core/pkg/datastore/postgres/pipelines.go:292), so a router that tried to read
+  a pipeline's wiring straight off ``component_of`` would work against a
+  too-generous mock and fail live.
 """
 
 # POST /v3/sessions — the embed session mint (swagger leaves the body untyped;
@@ -69,6 +74,37 @@ def connectors_list(kind: str) -> dict:
     }
 
 
+def connector_detail(kind: str) -> dict:
+    """GET /v1/{org}/{kind}s/{id} — the connector record plus ``component_of``,
+    the pipelines it is a node of. This is the one call that answers "which
+    pipeline is this connector wired into", so ``component_of`` is required
+    even when empty."""
+    return {
+        "type": "object",
+        "required": ["id", "type", "name", "component_of"],
+        "properties": {
+            "id": {"type": "string"},
+            "type": {"type": "string"},
+            "name": {"type": "string"},
+            "component_of": {
+                "type": "array",
+                "items": {
+                    "type": "object",
+                    "required": ["id", "enabled"],
+                    "properties": {
+                        "id": {"type": "string"},
+                        "name": {"type": "string"},
+                        "enabled": {"type": "boolean"},
+                        # Pinned absent — see the module note.
+                        "status": False,
+                        "nodes": False,
+                    },
+                },
+            },
+        },
+    }
+
+
 # A pipeline node as the routers read it (match on component_type/component_id).
 _NODE = {
     "type": "object",

conformance/test_mock_fidelity.py

@@ -64,10 +64,58 @@ def test_connectors_list_shape(http, kind):
     _valid(S.connectors_list(kind), r.json())
 
 
-def test_output_create_shape(http):
-    r = http.post(f"/v2/{_ORG}/outputs", json={"output_type": "dev-null", "name": "sink"})
+def test_connectors_list_defaults_to_ten(http):
+    # Monad's list handlers default `limit` to 10 with no maximum. A router that
+    # omits it gets a truncated page — the mock must reproduce that, not paper
+    # over it, or the "only ever sees 10" bug stays invisible until production.
+    unpaged = http.get(f"/v1/{_ORG}/inputs").json()
+    assert len(unpaged["inputs"]) == 10
+    assert unpaged["pagination"] == {"limit": 10, "offset": 0, "total": mock_monad._SEEDED_INPUTS}
+
+    rest = http.get(f"/v1/{_ORG}/inputs?limit=10&offset=10").json()
+    assert len(rest["inputs"]) == mock_monad._SEEDED_INPUTS - 10
+    # Past the end the list is null, not [] — the same shape as "tenant has none".
+    assert http.get(f"/v1/{_ORG}/inputs?limit=10&offset=99").json()["inputs"] is None
+
+
+@pytest.mark.parametrize("kind", ["input", "output"])
+def test_connector_detail_shape(http, kind):
+    cid = "in_conf" if kind == "input" else "out_store"
+    r = http.get(f"/v1/{_ORG}/{kind}s/{cid}")
+    assert r.status_code == 200, r.text
+    _valid(S.connector_detail(kind), r.json())
+    # Unconnected component → an empty `component_of`, not a missing key.
+    assert r.json()["component_of"] == []
+    assert http.get(f"/v1/{_ORG}/{kind}s/nope_404").status_code == 404
+
+
+def test_connector_detail_reports_component_of(http):
+    body = {
+        "name": "component-of",
+        "enabled": True,
+        "nodes": [
+            {"slug": "in", "component_id": "cfg_3", "component_type": "input", "enabled": True},
+            {"slug": "out", "component_id": "out_store", "component_type": "output", "enabled": True},
+        ],
+        "edges": [],
+    }
+    pid = http.post(f"/v2/{_ORG}/pipelines/", json=body).json()["id"]
+    for cid in ("cfg_3", "out_store"):
+        kind = "inputs" if cid.startswith("cfg") else "outputs"
+        entry = http.get(f"/v1/{_ORG}/{kind}/{cid}").json()["component_of"]
+        assert [p["id"] for p in entry] == [pid]
+        assert entry[0]["enabled"] is True
+    http.delete(f"/v2/{_ORG}/pipelines/{pid}")
+
+
+@pytest.mark.parametrize("field", ["type", "output_type"])
+def test_output_create_shape(http, field):
+    # `type` is canonical; `output_type` is the deprecated alias the API still
+    # accepts. Both must round-trip to the same created record.
+    r = http.post(f"/v2/{_ORG}/outputs", json={field: "dev-null", "name": "sink"})
     assert r.status_code == 200
     _valid(S.OUTPUT, r.json())
+    assert r.json()["type"] == "dev-null"
 
 
 def test_pipeline_lifecycle_shapes(http):