diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 1518462..79894b0 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -46,22 +46,35 @@ jobs:
       - run: pnpm run build
 
   conformance:
-    name: Conformance (ts)
+    name: Conformance (${{ matrix.router }})
     runs-on: ubuntu-latest
+    strategy:
+      fail-fast: false
+      matrix:
+        router: [ts, go]
     steps:
       - uses: actions/checkout@v4
 
       - uses: pnpm/action-setup@v4
+        if: matrix.router == 'ts'
 
       - uses: actions/setup-node@v4
+        if: matrix.router == 'ts'
         with:
           node-version-file: '.tool-versions'
           cache: pnpm
 
-      - run: pnpm install --frozen-lockfile
-
       - name: Build the TS router
-        run: pnpm -C routers/typescript build
+        if: matrix.router == 'ts'
+        run: |
+          pnpm install --frozen-lockfile
+          pnpm -C routers/typescript build
+
+      - name: Set up Go
+        if: matrix.router == 'go'
+        uses: actions/setup-go@v5
+        with:
+          go-version: 'stable'
 
       - uses: astral-sh/setup-uv@v5
 
@@ -71,8 +84,8 @@ jobs:
           uv venv --python 3.10 .venv
           uv pip install --python .venv/bin/python schemathesis pytest httpx jsonschema
 
-      - name: Run conformance against the TS router
+      - name: Run conformance (${{ matrix.router }})
         working-directory: conformance
         env:
-          ROUTER: ts
+          ROUTER: ${{ matrix.router }}
         run: .venv/bin/python -m pytest -q
diff --git a/routers/go/README.md b/routers/go/README.md
new file mode 100644
index 0000000..ba70e77
--- /dev/null
+++ b/routers/go/README.md
@@ -0,0 +1,114 @@
+# Monad Embed — Go router
+
+A mountable `/embed` backend router for a Monad embed integration, on the
+standard `net/http`. Implements the [`/embed` route contract](../../packages/embed/openapi/embed.openapi.yaml).
+
+```go
+import embed "github.com/monad-inc/embed/routers/go"
+
+h := embed.Router(embed.Config{
+    APIKey: os.Getenv("MONAD_API_KEY"),
+    // your auth → the tenant's Monad team id
+    GetCustomerOrgID: func(r *http.Request) (string, error) { return sessionOrg(r), nil },
+    // server-side lookup of a tenant's pre-provisioned components (which
+    // destination an input feeds, etc.)
+    GetProvisionedComponents: func(org string) embed.Provision {
+        return embed.Provision{DestinationOutputID: stores[org]}
+    },
+})
+
+mux.Handle("/embed/", http.StripPrefix("/embed", h))
+```
+
+Mount under `/embed` (via `http.StripPrefix`) so the router sees `/session`,
+`/pipelines/ingress`, etc. The browser holds only a session token; this router
+holds the API key and is the seam to the Monad API.
+
+## Mounting on common frameworks
+
+`embed.Router(cfg)` is a standard `http.Handler` that expects root-relative
+paths, so mount it under `/embed` and strip the prefix. Assume `h := embed.Router(cfg)`.
+
+**net/http** — standard library (Go 1.22+)
+
+```go
+mux := http.NewServeMux()
+mux.Handle("/embed/", http.StripPrefix("/embed", h))
+```
+
+**chi** — [go-chi/chi](https://github.com/go-chi/chi)
+
+```go
+r := chi.NewRouter()
+r.Mount("/embed", http.StripPrefix("/embed", h))
+```
+
+**gorilla/mux** — [gorilla/mux](https://github.com/gorilla/mux)
+
+```go
+r := mux.NewRouter()
+r.PathPrefix("/embed/").Handler(http.StripPrefix("/embed", h))
+```
+
+**Echo** — [labstack/echo](https://github.com/labstack/echo)
+
+```go
+e.Any("/embed/*", echo.WrapHandler(http.StripPrefix("/embed", h)))
+```
+
+**Gin** — [gin-gonic/gin](https://github.com/gin-gonic/gin)
+
+```go
+r.Any("/embed/*any", gin.WrapH(http.StripPrefix("/embed", h)))
+```
+
+## Config
+
+| Field                      | Purpose                                                                                                        |
+| -------------------------- | -------------------------------------------------------------------------------------------------------------- |
+| `APIKey`                   | Long-lived Monad API key (server-side only).                                                                   |
+| `APIBase`                  | Monad API base. Optional — defaults to `https://app.monad.com/api` (production).                               |
+| `FrameOrigin`              | Iframe origin returned by `GET /embed/config`. Optional — defaults to production.                              |
+| `GetCustomerOrgID`         | **The tenant-isolation boundary** — see below. Return `("", nil)` or an error to reject (→ 401).               |
+| `GetProvisionedComponents` | Per-tenant `Provision{ DestinationOutputID, SourceInputID }`. Nil → ingress uses dev/null, egress unavailable. |
+| `CatalogAllow`             | Restrict the catalog to these connector type ids. Nil → all.                                                   |
+
+### `GetCustomerOrgID` is a security boundary
+
+This callback is the only thing standing between one of your tenants and
+another's data. The router holds a Monad API key with access to every
+organization your key can reach; whatever org id this returns is the org the
+request then reads, writes and deletes in. Get it wrong and you serve tenant A
+the pipelines of tenant B.
+
+So derive the org **from your own authenticated session** — the thing your auth
+middleware already verified:
+
+```go
+GetCustomerOrgID: func(r *http.Request) (string, error) {
+    user, ok := auth.FromContext(r.Context()) // set by your auth middleware
+    if !ok {
+        return "", errors.New("not signed in")
+    }
+    return user.MonadOrgID, nil // your tenant → Monad org mapping
+}
+```
+
+Never take it from anything the caller controls — a header, a query param, a
+request body field, or a client-supplied JWT claim you have not verified. Those
+are all attacker-chosen values, and the router will use them verbatim.
+
+Mount the router behind your auth middleware; it performs no authentication of
+its own beyond calling this hook, and treats `("", nil)` or a non-nil error as
+`401`.
+
+For the full trust model — why each embed tenant needs its **own dedicated
+Monad org**, and the recommendation to monitor for orphaned resources — see
+[`docs/embed-security-and-operations.md`](../../docs/embed-security-and-operations.md).
+
+You own this mapping because only you can know it: Monad has no endpoint that
+turns your product's bearer token into an org id. (`GET /v1/organizations` maps
+a _Monad API key_ to the orgs it can reach — not your user to their tenant.)
+Store the tenant → org id association when you provision the tenant.
+
+Zero external dependencies. `go test ./...`, `go vet ./...`, `gofmt` clean.
diff --git a/routers/go/client.go b/routers/go/client.go
new file mode 100644
index 0000000..750cd8a
--- /dev/null
+++ b/routers/go/client.go
@@ -0,0 +1,397 @@
+package embed
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"slices"
+	"time"
+)
+
+// client is the hand-written Monad API client — the Go equivalent of the TS
+// kit. It sequences Monad's /v1 + /v2 + /v3 calls; the router layer (embed.go)
+// turns its results into the /embed contract's responses.
+type client struct {
+	apiKey       string
+	apiBase      string
+	hc           *http.Client
+	pollAttempts int
+	pollInterval time.Duration
+}
+
+func newClient(cfg Config) *client {
+	return &client{
+		apiKey:       cfg.APIKey,
+		apiBase:      cfg.APIBase,
+		hc:           &http.Client{Timeout: 30 * time.Second},
+		pollAttempts: 15,
+		pollInterval: 2 * time.Second,
+	}
+}
+
+// componentKind is the contract's `kind` — the only two values any /embed route
+// accepts. Mirrors the ComponentKind enum in embed.openapi.yaml.
+type componentKind string
+
+const (
+	kindInput  componentKind = "input"
+	kindOutput componentKind = "output"
+)
+
+// collection is the Monad path segment for a kind. Spelled out rather than
+// appending "s", so the mapping is a fact rather than a coincidence of English.
+func (k componentKind) collection() string {
+	if k == kindInput {
+		return "inputs"
+	}
+	return "outputs"
+}
+
+// pageSize is how many rows to request when draining a Monad list. Monad
+// defaults `limit` to 10 and enforces no maximum, so this is a round-trip /
+// response-size trade-off only: correctness comes from draining every page.
+const pageSize = 200
+
+// maxPages bounds draining a single list, so an upstream that never returns a
+// short page (e.g. offset-clamping) can't spin forever / exhaust memory.
+// maxPages * pageSize = 200k rows — far beyond any real tenant.
+const maxPages = 1000
+
+// upstreamError carries a failed Monad call's status + detail. The router maps
+// the status onto the contract's error model (404→not_found, 409→conflict,
+// otherwise 502 upstream_error).
+type upstreamError struct {
+	status int
+	detail string
+}
+
+func (e *upstreamError) Error() string { return e.detail }
+
+// do performs an authenticated request and returns the raw JSON body. A >=400
+// response becomes an error, which the router maps to 502 upstream_error.
+func (c *client) do(ctx context.Context, method, path string, body any) ([]byte, error) {
+	var rdr io.Reader
+	if body != nil {
+		b, err := json.Marshal(body)
+		if err != nil {
+			return nil, err
+		}
+		rdr = bytes.NewReader(b)
+	}
+	req, err := http.NewRequestWithContext(ctx, method, c.apiBase+path, rdr)
+	if err != nil {
+		return nil, err
+	}
+	req.Header.Set("Authorization", "ApiKey "+c.apiKey)
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.hc.Do(req)
+	if err != nil {
+		return nil, err
+	}
+	defer resp.Body.Close()
+
+	data, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, err
+	}
+	if resp.StatusCode >= 400 {
+		// Keep the FULL upstream body in the detail: it is logged server-side
+		// only (never returned to the browser), and truncating it can drop the
+		// validation details that make a failure diagnosable. The status is a
+		// struct field, so it isn't repeated in the string.
+		return nil, &upstreamError{status: resp.StatusCode, detail: fmt.Sprintf("%s: %s", path, string(data))}
+	}
+	return data, nil
+}
+
+func (c *client) mintSession(ctx context.Context, org string) (Session, error) {
+	data, err := c.do(ctx, "POST", "/v3/sessions", map[string]any{
+		"ttl_seconds": 1800, "organization_id": org,
+	})
+	if err != nil {
+		return Session{}, err
+	}
+	var out struct {
+		SessionToken string `json:"session_token"`
+		ExpiresAt    string `json:"expires_at"`
+	}
+	if err := json.Unmarshal(data, &out); err != nil {
+		return Session{}, err
+	}
+	return Session{SessionToken: out.SessionToken, OrganizationID: org, ExpiresAt: out.ExpiresAt}, nil
+}
+
+func (c *client) listCatalog(ctx context.Context, kind componentKind, allow []string) ([]CatalogType, error) {
+	data, err := c.do(ctx, "GET", "/v1/"+kind.collection(), nil)
+	if err != nil {
+		return nil, err
+	}
+	var raw []struct {
+		TypeID string `json:"type_id"`
+		Name   string `json:"name"`
+	}
+	if err := json.Unmarshal(data, &raw); err != nil {
+		return nil, err
+	}
+	out := []CatalogType{}
+	for _, t := range raw {
+		// allow is a short operator-configured set of connector types, so a
+		// linear check is simpler than building a set and the list is tiny.
+		if len(allow) > 0 && !slices.Contains(allow, t.TypeID) {
+			continue
+		}
+		out = append(out, CatalogType{TypeID: t.TypeID, Name: t.Name})
+	}
+	return out, nil
+}
+
+// listConnectors returns every connector of a kind the tenant has configured.
+//
+// Monad pages this at limit=10 by default with no maximum, and the /embed
+// contract returns a bare array with no pagination — so draining the pages is
+// the router's job. A single large limit instead would silently truncate
+// whichever tenant outgrows it.
+func (c *client) listConnectors(ctx context.Context, org string, kind componentKind) ([]ConfiguredConnector, error) {
+	key := kind.collection()
+	out := []ConfiguredConnector{}
+	// A short page is the ONLY safe stop: trusting pagination.total to break
+	// early truncates the list when Monad under-reports it. maxPages bounds a
+	// pathological upstream that never returns a short page.
+	for pageIndex := 0; ; pageIndex++ {
+		if pageIndex >= maxPages {
+			return nil, fmt.Errorf("pagination exceeded %d pages draining %s", maxPages, key)
+		}
+		offset := pageIndex * pageSize
+		path := fmt.Sprintf("/v1/%s/%s?limit=%d&offset=%d", url.PathEscape(org), key, pageSize, offset)
+		data, err := c.do(ctx, "GET", path, nil)
+		if err != nil {
+			return nil, err
+		}
+		// Rows come wrapped as { inputs: [...] } / { outputs: [...] } alongside a
+		// `pagination` object. Decode only the connector array for this kind so
+		// the sibling can't force a type mismatch; the array is `null`, not `[]`,
+		// when the page is empty.
+		var page map[string]json.RawMessage
+		if err := json.Unmarshal(data, &page); err != nil {
+			return nil, err
+		}
+		var rows []struct {
+			ID   string `json:"id"`
+			Type string `json:"type"`
+			Name string `json:"name"`
+		}
+		if raw, ok := page[key]; ok {
+			if err := json.Unmarshal(raw, &rows); err != nil {
+				return nil, err
+			}
+		}
+		for _, r := range rows {
+			// Monad names the type slug `type` here; the contract calls it typeId.
+			out = append(out, ConfiguredConnector{ID: r.ID, TypeID: r.Type, Name: r.Name})
+		}
+		// A short page is the last page.
+		if len(rows) < pageSize {
+			return out, nil
+		}
+	}
+}
+
+type wireNode struct {
+	Slug          string `json:"slug"`
+	ComponentID   string `json:"component_id"`
+	ComponentType string `json:"component_type"`
+	Enabled       bool   `json:"enabled"`
+}
+
+// wirePipeline creates an input→output pipeline and polls until it reports
+// Running (or Erroring). The shared primitive behind ingress and egress.
+func (c *client) wirePipeline(ctx context.Context, org, inputID, outputID, name string) (BuiltPipeline, error) {
+	body := map[string]any{
+		"name":        name,
+		"description": "Created when the connector was configured via embed",
+		"enabled":     true,
+		"nodes": []wireNode{
+			{Slug: "in", ComponentID: inputID, ComponentType: "input", Enabled: true},
+			{Slug: "out", ComponentID: outputID, ComponentType: "output", Enabled: true},
+		},
+		"edges": []map[string]any{{
+			"from_node_instance_id": "in",
+			"to_node_instance_id":   "out",
+			"description":           "all records",
+			"conditions":            map[string]any{"operator": "always"},
+		}},
+	}
+	data, err := c.do(ctx, "POST", "/v2/"+url.PathEscape(org)+"/pipelines/", body)
+	if err != nil {
+		return BuiltPipeline{}, err
+	}
+	var created struct {
+		ID string `json:"id"`
+	}
+	if err := json.Unmarshal(data, &created); err != nil {
+		return BuiltPipeline{}, err
+	}
+
+	// The pipeline now EXISTS. Polling for Running is a courtesy — a transient
+	// status-poll failure (or a cancelled request) must NOT fail the request, or
+	// the caller treats a live pipeline as un-built and retries into a duplicate.
+	// On a poll error or cancellation we return the pipeline with its last-known
+	// status instead.
+	status := "Pending"
+	for i := 0; i < c.pollAttempts; i++ {
+		sd, err := c.do(ctx, "GET", "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(created.ID)+"/status", nil)
+		if err != nil {
+			break
+		}
+		var s struct {
+			Status string `json:"status"`
+		}
+		if err := json.Unmarshal(sd, &s); err == nil && s.Status != "" {
+			status = s.Status
+		}
+		if status == "Running" || status == "Erroring" {
+			break
+		}
+		select {
+		case <-ctx.Done():
+			return BuiltPipeline{PipelineID: created.ID, OutputID: outputID, Status: status, Active: false}, nil
+		case <-time.After(c.pollInterval):
+		}
+	}
+	return BuiltPipeline{PipelineID: created.ID, OutputID: outputID, Status: status, Active: status == "Running"}, nil
+}
+
+// buildDevNull creates a throwaway dev/null output, then wires the input to it.
+func (c *client) buildDevNull(ctx context.Context, org, inputID, name string) (BuiltPipeline, error) {
+	data, err := c.do(ctx, "POST", "/v2/"+url.PathEscape(org)+"/outputs", map[string]any{
+		// `type` is the canonical field; `output_type` is a deprecated alias.
+		"type":        "dev-null",
+		"name":        name + " → /dev/null",
+		"description": "Auto-created sink for embed pipeline",
+		"promise_id":  "",
+		"config":      map[string]any{"settings": map[string]any{}, "secrets": map[string]any{}},
+	})
+	if err != nil {
+		return BuiltPipeline{}, err
+	}
+	var out struct {
+		ID string `json:"id"`
+	}
+	if err := json.Unmarshal(data, &out); err != nil {
+		return BuiltPipeline{}, err
+	}
+	built, err := c.wirePipeline(ctx, org, inputID, out.ID, name)
+	if err != nil {
+		// The sink was created but wiring failed. Best-effort delete it so a
+		// retry doesn't accumulate dev/null outputs; if the cleanup itself fails,
+		// leave it for orphan monitoring (docs/embed-security-and-operations.md)
+		// and surface the original error. Create/delete are non-atomic — Monad
+		// offers no transaction.
+		_, _ = c.do(ctx, "DELETE", "/v1/"+url.PathEscape(org)+"/outputs/"+url.PathEscape(out.ID), nil)
+		return BuiltPipeline{}, err
+	}
+	return built, nil
+}
+
+type pipeNode struct {
+	ComponentID   string `json:"component_id"`
+	ComponentType string `json:"component_type"`
+}
+
+// pipelineNodes reads a pipeline's wiring. The detail response is flat — the
+// nodes sit at the top level, not under a `config` envelope.
+func (c *client) pipelineNodes(ctx context.Context, org, pipelineID string) ([]pipeNode, error) {
+	data, err := c.do(ctx, "GET", "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(pipelineID), nil)
+	if err != nil {
+		return nil, err
+	}
+	var detail struct {
+		Nodes []pipeNode `json:"nodes"`
+	}
+	if err := json.Unmarshal(data, &detail); err != nil {
+		return nil, err
+	}
+	return detail.Nodes, nil
+}
+
+// pipelineFor resolves the pipeline a configured connector is wired into.
+//
+// Monad answers this directly: GET /v1/{org}/{kind}s/{id} returns
+// `component_of`, the pipelines the component is a node of. Walking the org's
+// pipeline list instead would be both O(n) and wrong — that list pages at 10,
+// so any pipeline past the first page would read as "not connected".
+//
+// `component_of` carries no wiring (the datastore fills only a summary
+// projection), so resolving the peer connector costs one further fetch.
+// Returns an upstreamError with status 404 when the connector does not exist.
+func (c *client) pipelineFor(ctx context.Context, org string, kind componentKind, connectorID string) (PipelineStatus, error) {
+	data, err := c.do(ctx, "GET",
+		"/v1/"+url.PathEscape(org)+"/"+kind.collection()+"/"+url.PathEscape(connectorID), nil)
+	if err != nil {
+		return PipelineStatus{}, err
+	}
+	var connector struct {
+		ComponentOf []struct {
+			ID      string `json:"id"`
+			Enabled bool   `json:"enabled"`
+		} `json:"component_of"`
+	}
+	if err := json.Unmarshal(data, &connector); err != nil {
+		return PipelineStatus{}, err
+	}
+	// The embed flow wires each connector into exactly one pipeline (Monad
+	// guarantees a pipeline has a single input, and embed never adds a connector
+	// to a second pipeline), so component_of has at most one entry and taking the
+	// first is unambiguous.
+	if len(connector.ComponentOf) == 0 || connector.ComponentOf[0].ID == "" {
+		return PipelineStatus{HasPipeline: false}, nil
+	}
+	p := connector.ComponentOf[0]
+	ps := PipelineStatus{HasPipeline: true, Enabled: p.Enabled, PipelineID: p.ID}
+
+	peer := kindOutput
+	if kind == kindOutput {
+		peer = kindInput
+	}
+	nodes, err := c.pipelineNodes(ctx, org, p.ID)
+	if err != nil {
+		return PipelineStatus{}, err
+	}
+	for _, n := range nodes {
+		if n.ComponentType != string(peer) {
+			continue
+		}
+		if kind == kindInput {
+			ps.OutputID = n.ComponentID
+		} else {
+			ps.InputID = n.ComponentID
+		}
+		break
+	}
+	return ps, nil
+}
+
+// setEnabled flips the pipeline's enabled flag.
+//
+// PATCH is a true partial update: omitted fields keep their stored value and
+// the node/edge graph is preserved untouched. Reading the pipeline and sending
+// it back would replace the graph with whatever subset of fields the round trip
+// happened to reproduce — silently dropping node `config_overrides` and edge
+// `schema_detection_spec`.
+func (c *client) setEnabled(ctx context.Context, org, pipelineID string, enabled bool) error {
+	_, err := c.do(ctx, "PATCH", "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(pipelineID),
+		map[string]any{"enabled": enabled})
+	return err
+}
+
+// del removes a resource. Order matters at the call site: the pipeline
+// references its input/output, so it must be deleted first.
+func (c *client) del(ctx context.Context, path string) error {
+	_, err := c.do(ctx, "DELETE", path, nil)
+	return err
+}
diff --git a/routers/go/cmd/conformance/main.go b/routers/go/cmd/conformance/main.go
new file mode 100644
index 0000000..309c9c1
--- /dev/null
+++ b/routers/go/cmd/conformance/main.go
@@ -0,0 +1,64 @@
+// Command conformance boots the Go /embed router for the conformance harness.
+//
+// In the default (mock) mode the harness points it at the in-memory mock via
+// MONAD_API_BASE; in live mode the same knobs are set to real Monad staging
+// values. Every value falls back to the mock fixture default, so the existing
+// hermetic run is unchanged.
+package main
+
+import (
+	"net/http"
+	"os"
+	"strings"
+
+	embed "github.com/monad-inc/embed/routers/go"
+)
+
+// envOr returns the env var value, or fallback when unset. An explicitly-empty
+// value ("") is returned as-is — a live tenant may have no provisioned store.
+func envOr(name, fallback string) string {
+	if v, ok := os.LookupEnv(name); ok {
+		return v
+	}
+	return fallback
+}
+
+// splitAllow parses a comma-separated allow-list; empty → nil (expose all).
+func splitAllow(s string) []string {
+	var out []string
+	for _, part := range strings.Split(s, ",") {
+		if p := strings.TrimSpace(part); p != "" {
+			out = append(out, p)
+		}
+	}
+	return out
+}
+
+func main() {
+	store := envOr("MONAD_STORE_ID", "out_store")
+	source := envOr("MONAD_SOURCE_ID", "in_source")
+
+	cfg := embed.Config{
+		APIKey:      envOr("MONAD_API_KEY", "conf-key"),
+		APIBase:     os.Getenv("MONAD_API_BASE"),
+		FrameOrigin: envOr("MONAD_FRAME_ORIGIN", "https://app.monad.com/embed"),
+		GetCustomerOrgID: func(r *http.Request) (string, error) {
+			return envOr("MONAD_ORG_ID", "org_conf"), nil
+		},
+		GetProvisionedComponents: func(org string) embed.Provision {
+			return embed.Provision{DestinationOutputID: store, SourceInputID: source}
+		},
+		CatalogAllow: splitAllow(envOr("MONAD_CATALOG_ALLOW", "aws-cloudtrail,okta-systemlog")),
+	}
+
+	mux := http.NewServeMux()
+	mux.Handle("/embed/", http.StripPrefix("/embed", embed.Router(cfg)))
+
+	port := os.Getenv("PORT")
+	if port == "" {
+		port = "8791"
+	}
+	if err := http.ListenAndServe("127.0.0.1:"+port, mux); err != nil {
+		panic(err)
+	}
+}
diff --git a/routers/go/embed.go b/routers/go/embed.go
new file mode 100644
index 0000000..44ef073
--- /dev/null
+++ b/routers/go/embed.go
@@ -0,0 +1,446 @@
+// Package embed is a mountable /embed backend router for a Monad embed
+// integration. It implements the /embed route contract
+// (packages/embed/openapi/embed.openapi.yaml) on the standard net/http, so it
+// mounts under any router:
+//
+//	h := embed.Router(embed.Config{
+//		APIKey:      os.Getenv("MONAD_API_KEY"),
+//		APIBase:     "https://app.monad.com/api",
+//		FrameOrigin: "https://app.monad.com/embed",
+//		GetCustomerOrgID: func(r *http.Request) (string, error) { return sessionOrg(r), nil },
+//	})
+//	mux.Handle("/embed/", http.StripPrefix("/embed", h))
+//
+// The browser holds only a session token; this router (holding the API key) is
+// the seam to the Monad API.
+package embed
+
+import (
+	"encoding/json"
+	"errors"
+	"log"
+	"net/http"
+	"net/url"
+)
+
+// Provision is what the host pre-provisions per tenant, resolved server-side —
+// never sent by the browser.
+type Provision struct {
+	// DestinationOutputID is the ingress target. Empty → ingress goes to a dev/null sink.
+	DestinationOutputID string
+	// SourceInputID is the egress source. Empty → egress is unavailable.
+	SourceInputID string
+}
+
+// Config is the uniform router configuration shared across every language.
+type Config struct {
+	// APIKey is the long-lived Monad API key. Server-side only.
+	APIKey string
+	// APIBase is the Monad API base. Defaults to "https://app.monad.com/api"
+	// (production); set only for non-prod.
+	APIBase string
+	// FrameOrigin is returned by GET /embed/config. Defaults to
+	// "https://app.monad.com/embed" (production); set only for non-prod.
+	FrameOrigin string
+	// GetCustomerOrgID maps the authenticated request to the caller's Monad team id.
+	// The one seam only the host can fill. Return a non-nil error (or "") to
+	// reject an unauthenticated caller — the router responds 401.
+	GetCustomerOrgID func(r *http.Request) (string, error)
+	// GetProvisionedComponents returns a tenant's pre-provisioned resources. Nil →
+	// ingress uses dev/null and egress is unavailable.
+	GetProvisionedComponents func(org string) Provision
+	// CatalogAllow restricts the catalog to these connector type ids. Nil → all.
+	CatalogAllow []string
+}
+
+// ---- response shapes (camelCase per the /embed contract) ----
+
+type configResponse struct {
+	FrameOrigin string `json:"frameOrigin"`
+	APIBase     string `json:"apiBase"`
+}
+
+// Session is a minted embed session, shaped to drop straight into the iframe mount.
+type Session struct {
+	SessionToken   string `json:"sessionToken"`
+	OrganizationID string `json:"organizationId"`
+	ExpiresAt      string `json:"expiresAt"`
+}
+
+// CatalogType is a connector type available to embed.
+type CatalogType struct {
+	TypeID string `json:"typeId"`
+	Name   string `json:"name"`
+}
+
+// ConfiguredConnector is a connector a tenant has already configured.
+type ConfiguredConnector struct {
+	ID     string `json:"id"`
+	TypeID string `json:"typeId"`
+	Name   string `json:"name"`
+}
+
+// BuiltPipeline is the result of standing up a pipeline.
+type BuiltPipeline struct {
+	PipelineID string `json:"pipelineId"`
+	OutputID   string `json:"outputId"`
+	Status     string `json:"status"`
+	Active     bool   `json:"active"`
+}
+
+// PipelineStatus is a configured connector's resolved pipeline state.
+type PipelineStatus struct {
+	HasPipeline bool   `json:"hasPipeline"`
+	Enabled     bool   `json:"enabled"`
+	PipelineID  string `json:"pipelineId,omitempty"`
+	InputID     string `json:"inputId,omitempty"`
+	OutputID    string `json:"outputId,omitempty"`
+}
+
+type errorBody struct {
+	Code    string `json:"code"`
+	Message string `json:"message"`
+}
+
+// ---- router ----
+
+type handler struct {
+	cfg Config
+	c   *client
+}
+
+// Router builds the /embed router. Mount it under "/embed" (e.g. via
+// http.StripPrefix) so it sees paths like "/session" and "/pipelines/ingress".
+func Router(cfg Config) http.Handler {
+	if cfg.GetCustomerOrgID == nil {
+		panic("embed: Config.GetCustomerOrgID is required")
+	}
+	// Production defaults — set APIBase/FrameOrigin only for non-prod.
+	if cfg.APIBase == "" {
+		cfg.APIBase = "https://app.monad.com/api"
+	}
+	if cfg.FrameOrigin == "" {
+		cfg.FrameOrigin = "https://app.monad.com/embed"
+	}
+	h := &handler{cfg: cfg, c: newClient(cfg)}
+	mux := http.NewServeMux()
+	mux.HandleFunc("GET /config", h.getConfig)
+	mux.HandleFunc("POST /session", h.session)
+	mux.HandleFunc("GET /catalog", h.catalog)
+	mux.HandleFunc("GET /connectors", h.connectors)
+	mux.HandleFunc("POST /pipelines/ingress", h.ingress)
+	mux.HandleFunc("POST /pipelines/egress", h.egress)
+	mux.HandleFunc("GET /pipelines", h.status)
+	mux.HandleFunc("POST /pipelines/state", h.state)
+	mux.HandleFunc("POST /pipelines/remove", h.remove)
+	mux.HandleFunc("/", h.notFound)
+	return mux
+}
+
+// ---- handlers ----
+
+func (h *handler) getConfig(w http.ResponseWriter, _ *http.Request) {
+	writeJSON(w, http.StatusOK, configResponse{FrameOrigin: h.cfg.FrameOrigin, APIBase: h.cfg.APIBase})
+}
+
+func (h *handler) session(w http.ResponseWriter, r *http.Request) {
+	org, ok := h.tenant(w, r)
+	if !ok {
+		return
+	}
+	s, err := h.c.mintSession(r.Context(), org)
+	if err != nil {
+		upstream(w, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, s)
+}
+
+func (h *handler) catalog(w http.ResponseWriter, r *http.Request) {
+	if _, ok := h.tenant(w, r); !ok {
+		return
+	}
+	kind, ok := kindQuery(w, r.URL.Query().Get("kind"))
+	if !ok {
+		return
+	}
+	types, err := h.c.listCatalog(r.Context(), kind, h.cfg.CatalogAllow)
+	if err != nil {
+		upstream(w, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, types)
+}
+
+func (h *handler) connectors(w http.ResponseWriter, r *http.Request) {
+	org, ok := h.tenant(w, r)
+	if !ok {
+		return
+	}
+	kind, ok := kindQuery(w, r.URL.Query().Get("kind"))
+	if !ok {
+		return
+	}
+	rows, err := h.c.listConnectors(r.Context(), org, kind)
+	if err != nil {
+		upstream(w, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, rows)
+}
+
+func (h *handler) ingress(w http.ResponseWriter, r *http.Request) {
+	org, ok := h.tenant(w, r)
+	if !ok {
+		return
+	}
+	var body struct {
+		InputID string `json:"inputId"`
+		Name    string `json:"name"`
+	}
+	if !decode(w, r, &body) {
+		return
+	}
+	if !requireStr(w, body.InputID, "inputId") || !requireStr(w, body.Name, "name") {
+		return
+	}
+	prov := h.prov(org)
+	var built BuiltPipeline
+	var err error
+	if prov.DestinationOutputID != "" {
+		built, err = h.c.wirePipeline(r.Context(), org, body.InputID, prov.DestinationOutputID, body.Name)
+	} else {
+		built, err = h.c.buildDevNull(r.Context(), org, body.InputID, body.Name)
+	}
+	if err != nil {
+		upstream(w, err)
+		return
+	}
+	writeJSON(w, http.StatusCreated, built)
+}
+
+func (h *handler) egress(w http.ResponseWriter, r *http.Request) {
+	org, ok := h.tenant(w, r)
+	if !ok {
+		return
+	}
+	var body struct {
+		OutputID string `json:"outputId"`
+		Name     string `json:"name"`
+	}
+	if !decode(w, r, &body) {
+		return
+	}
+	if !requireStr(w, body.OutputID, "outputId") || !requireStr(w, body.Name, "name") {
+		return
+	}
+	prov := h.prov(org)
+	if prov.SourceInputID == "" {
+		writeErr(w, http.StatusInternalServerError, "internal_error",
+			"No source input is provisioned for this tenant; egress cannot be built.")
+		return
+	}
+	built, err := h.c.wirePipeline(r.Context(), org, prov.SourceInputID, body.OutputID, body.Name)
+	if err != nil {
+		upstream(w, err)
+		return
+	}
+	writeJSON(w, http.StatusCreated, built)
+}
+
+func (h *handler) status(w http.ResponseWriter, r *http.Request) {
+	org, ok := h.tenant(w, r)
+	if !ok {
+		return
+	}
+	connectorID := r.URL.Query().Get("connectorId")
+	if !requireStr(w, connectorID, "connectorId") {
+		return
+	}
+	kind, ok := kindQuery(w, r.URL.Query().Get("kind"))
+	if !ok {
+		return
+	}
+	ps, err := h.c.pipelineFor(r.Context(), org, kind, connectorID)
+	if err != nil {
+		upstream(w, err)
+		return
+	}
+	writeJSON(w, http.StatusOK, ps)
+}
+
+func (h *handler) state(w http.ResponseWriter, r *http.Request) {
+	org, ok := h.tenant(w, r)
+	if !ok {
+		return
+	}
+	var body struct {
+		PipelineID string `json:"pipelineId"`
+		Enabled    *bool  `json:"enabled"`
+	}
+	if !decode(w, r, &body) {
+		return
+	}
+	if !requireStr(w, body.PipelineID, "pipelineId") {
+		return
+	}
+	if body.Enabled == nil {
+		writeErr(w, http.StatusBadRequest, "invalid_request", "Field 'enabled' must be a boolean.")
+		return
+	}
+	if err := h.c.setEnabled(r.Context(), org, body.PipelineID, *body.Enabled); err != nil {
+		upstream(w, err)
+		return
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func (h *handler) remove(w http.ResponseWriter, r *http.Request) {
+	org, ok := h.tenant(w, r)
+	if !ok {
+		return
+	}
+	var body struct {
+		ConnectorID string `json:"connectorId"`
+		Kind        string `json:"kind"`
+	}
+	if !decode(w, r, &body) {
+		return
+	}
+	if !requireStr(w, body.ConnectorID, "connectorId") {
+		return
+	}
+	kind, ok := kindQuery(w, body.Kind)
+	if !ok {
+		return
+	}
+	ctx := r.Context()
+
+	ps, err := h.c.pipelineFor(ctx, org, kind, body.ConnectorID)
+	if err != nil {
+		upstream(w, err)
+		return
+	}
+	if ps.PipelineID != "" {
+		if err := h.c.del(ctx, "/v2/"+url.PathEscape(org)+"/pipelines/"+url.PathEscape(ps.PipelineID)); err != nil {
+			upstream(w, err)
+			return
+		}
+	}
+
+	if kind == kindInput {
+		if err := h.c.del(ctx, "/v1/"+url.PathEscape(org)+"/inputs/"+url.PathEscape(body.ConnectorID)); err != nil {
+			upstream(w, err)
+			return
+		}
+		// The tenant's provisioned store is shared; only a sink this pipeline
+		// created on the fly gets torn down with it.
+		prov := h.prov(org)
+		keepStore := prov.DestinationOutputID != "" && prov.DestinationOutputID == ps.OutputID
+		if ps.OutputID != "" && !keepStore {
+			if err := h.c.del(ctx, "/v1/"+url.PathEscape(org)+"/outputs/"+url.PathEscape(ps.OutputID)); err != nil {
+				upstream(w, err)
+				return
+			}
+		}
+	} else {
+		// Keep the shared source input; remove only the user's output.
+		if err := h.c.del(ctx, "/v1/"+url.PathEscape(org)+"/outputs/"+url.PathEscape(body.ConnectorID)); err != nil {
+			upstream(w, err)
+			return
+		}
+	}
+	w.WriteHeader(http.StatusNoContent)
+}
+
+func (h *handler) notFound(w http.ResponseWriter, r *http.Request) {
+	writeErr(w, http.StatusNotFound, "not_found", "No route for "+r.Method+" "+r.URL.Path+".")
+}
+
+// ---- helpers ----
+
+func (h *handler) tenant(w http.ResponseWriter, r *http.Request) (string, bool) {
+	org, err := h.cfg.GetCustomerOrgID(r)
+	if err != nil || org == "" {
+		// Redact the host callback's error detail — it can carry internal
+		// information (DB errors, internal URLs). Log it server-side; return a
+		// generic message, like the upstream-Monad path does.
+		if err != nil {
+			log.Printf("[embed] GetCustomerOrgID failed: %v", err)
+		}
+		writeErr(w, http.StatusUnauthorized, "unauthenticated", "Could not resolve a tenant for the request.")
+		return "", false
+	}
+	return org, true
+}
+
+func (h *handler) prov(org string) Provision {
+	if h.cfg.GetProvisionedComponents != nil {
+		return h.cfg.GetProvisionedComponents(org)
+	}
+	return Provision{}
+}
+
+// kindQuery is the single place a request's `kind` becomes a componentKind —
+// anything else is rejected before it can reach a Monad path.
+func kindQuery(w http.ResponseWriter, value string) (componentKind, bool) {
+	switch componentKind(value) {
+	case kindInput, kindOutput:
+		return componentKind(value), true
+	}
+	writeErr(w, http.StatusBadRequest, "invalid_request", "'kind' must be 'input' or 'output'.")
+	return "", false
+}
+
+func requireStr(w http.ResponseWriter, value, field string) bool {
+	if value == "" {
+		writeErr(w, http.StatusBadRequest, "invalid_request", "Field '"+field+"' is required.")
+		return false
+	}
+	return true
+}
+
+// maxBodyBytes caps a request body. Embed requests carry only ids + names, so
+// 1 MiB is generous; the cap stops a hostile/buggy client (behind host auth)
+// from driving the shared process to OOM with an oversized body.
+const maxBodyBytes = 1 << 20
+
+func decode(w http.ResponseWriter, r *http.Request, v any) bool {
+	r.Body = http.MaxBytesReader(w, r.Body, maxBodyBytes)
+	if err := json.NewDecoder(r.Body).Decode(v); err != nil {
+		writeErr(w, http.StatusBadRequest, "invalid_request", "Request body must be valid JSON or is too large.")
+		return false
+	}
+	return true
+}
+
+func writeJSON(w http.ResponseWriter, status int, v any) {
+	w.Header().Set("Content-Type", "application/json; charset=utf-8")
+	w.WriteHeader(status)
+	_ = json.NewEncoder(w).Encode(v)
+}
+
+func writeErr(w http.ResponseWriter, status int, code, message string) {
+	writeJSON(w, status, errorBody{Code: code, Message: message})
+}
+
+// upstream maps a failed Monad call to the contract's 502 upstream_error.
+func upstream(w http.ResponseWriter, err error) {
+	// Translate Monad's own status codes into the contract's error model.
+	var ue *upstreamError
+	if errors.As(err, &ue) {
+		switch ue.status {
+		case http.StatusNotFound:
+			writeErr(w, http.StatusNotFound, "not_found",
+				"The referenced connector or pipeline does not exist.")
+			return
+		case http.StatusConflict:
+			writeErr(w, http.StatusConflict, "conflict",
+				"The request conflicts with existing state.")
+			return
+		}
+	}
+	// Anything else: log the detail server-side, return a generic 502.
+	log.Printf("[embed] upstream Monad API call failed: %v", err)
+	writeErr(w, http.StatusBadGateway, "upstream_error", "The upstream Monad API request failed.")
+}
diff --git a/routers/go/embed_test.go b/routers/go/embed_test.go
new file mode 100644
index 0000000..63c8d47
--- /dev/null
+++ b/routers/go/embed_test.go
@@ -0,0 +1,432 @@
+package embed_test
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strconv"
+	"strings"
+	"sync"
+	"testing"
+
+	embed "github.com/monad-inc/embed/routers/go"
+)
+
+// recorder captures what the router actually sent upstream, so a test can
+// assert on the request rather than only on the response.
+type recorder struct {
+	mu    sync.Mutex
+	calls []string // "METHOD /path?query"
+	body  map[string]string
+}
+
+func (rec *recorder) record(r *http.Request) {
+	rec.mu.Lock()
+	defer rec.mu.Unlock()
+	call := r.Method + " " + r.URL.Path
+	rec.calls = append(rec.calls, call+"?"+r.URL.RawQuery)
+	if r.Body != nil {
+		b, _ := io.ReadAll(r.Body)
+		rec.body[call] = string(b)
+	}
+}
+
+func (rec *recorder) count(prefix string) int {
+	rec.mu.Lock()
+	defer rec.mu.Unlock()
+	n := 0
+	for _, c := range rec.calls {
+		if strings.HasPrefix(c, prefix) {
+			n++
+		}
+	}
+	return n
+}
+
+// connectorsPage renders one page of `total` configured inputs the way Monad
+// does: the rows under the `<kind>s` key, a `pagination` sibling, and null
+// rather than [] once the page is empty.
+func connectorsPage(total int, q map[string][]string) string {
+	limit, _ := strconv.Atoi(first(q["limit"], "10"))
+	offset, _ := strconv.Atoi(first(q["offset"], "0"))
+	rows := []string{}
+	for i := offset; i < offset+limit && i < total; i++ {
+		rows = append(rows, fmt.Sprintf(`{"id":"cfg_%d","type":"aws-cloudtrail","name":"C%d"}`, i, i))
+	}
+	list := "null"
+	if len(rows) > 0 {
+		list = "[" + strings.Join(rows, ",") + "]"
+	}
+	return fmt.Sprintf(`{"inputs":%s,"pagination":{"limit":%d,"offset":%d,"total":%d}}`,
+		list, limit, offset, total)
+}
+
+func first(values []string, fallback string) string {
+	if len(values) == 0 {
+		return fallback
+	}
+	return values[0]
+}
+
+// mockMonad stands in for the Monad API with just the endpoints these tests
+// hit. `connectorTotal` sizes the configured-inputs list so paging is testable.
+func mockMonad(t *testing.T, connectorTotal int) (*httptest.Server, *recorder) {
+	t.Helper()
+	rec := &recorder{body: map[string]string{}}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		rec.record(r)
+		w.Header().Set("Content-Type", "application/json")
+		switch {
+		case r.Method == "POST" && r.URL.Path == "/v3/sessions":
+			_, _ = w.Write([]byte(`{"session_token":"tok","expires_at":"2026-01-01T00:00:00Z"}`))
+		case r.Method == "GET" && r.URL.Path == "/v1/inputs":
+			_, _ = w.Write([]byte(`[{"type_id":"aws-cloudtrail","name":"AWS"},{"type_id":"secret","name":"Hidden"}]`))
+		case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs":
+			_, _ = w.Write([]byte(connectorsPage(connectorTotal, r.URL.Query())))
+		case r.Method == "POST" && r.URL.Path == "/v2/org_1/outputs":
+			_, _ = w.Write([]byte(`{"id":"out_devnull"}`))
+		case r.Method == "POST" && r.URL.Path == "/v2/org_1/pipelines/":
+			_, _ = w.Write([]byte(`{"id":"pipe_1"}`))
+		case r.Method == "GET" && r.URL.Path == "/v2/org_1/pipelines/pipe_1/status":
+			_, _ = w.Write([]byte(`{"status":"Running"}`))
+
+		// A connector wired into pipe_1, and one wired into nothing.
+		case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs/in_wired":
+			_, _ = w.Write([]byte(`{"id":"in_wired","type":"aws-cloudtrail","name":"Wired",
+				"component_of":[{"id":"pipe_1","name":"p","enabled":true}]}`))
+		case r.Method == "GET" && r.URL.Path == "/v1/org_1/outputs/out_wired":
+			_, _ = w.Write([]byte(`{"id":"out_wired","type":"s3","name":"Wired",
+				"component_of":[{"id":"pipe_1","name":"p","enabled":false}]}`))
+		case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs/in_bare":
+			_, _ = w.Write([]byte(`{"id":"in_bare","type":"aws-cloudtrail","name":"Bare","component_of":[]}`))
+		case r.Method == "GET" && r.URL.Path == "/v1/org_1/inputs/in_gone":
+			http.Error(w, `{"error":"not found"}`, http.StatusNotFound)
+
+		// The pipeline detail: flat, nodes at the top level.
+		case r.Method == "GET" && r.URL.Path == "/v2/org_1/pipelines/pipe_1":
+			_, _ = w.Write([]byte(`{"id":"pipe_1","name":"p","enabled":true,"nodes":[
+				{"id":"n1","slug":"in","component_id":"in_wired","component_type":"input",
+				 "config_overrides":{"keep":"me"}},
+				{"id":"n2","slug":"out","component_id":"out_wired","component_type":"output"}],
+				"edges":[{"from_node_instance_id":"in","to_node_instance_id":"out",
+				 "schema_detection_spec":{"enabled":true}}]}`))
+		case r.Method == "PATCH" && r.URL.Path == "/v2/org_1/pipelines/pipe_1":
+			_, _ = w.Write([]byte(`{"id":"pipe_1"}`))
+
+		case r.Method == "DELETE" && strings.HasPrefix(r.URL.Path, "/v1/org_1/"),
+			r.Method == "DELETE" && strings.HasPrefix(r.URL.Path, "/v2/org_1/"):
+			w.WriteHeader(http.StatusNoContent)
+		default:
+			t.Errorf("unexpected Monad call: %s %s", r.Method, r.URL.Path)
+			http.Error(w, "unmocked", http.StatusInternalServerError)
+		}
+	}))
+	return srv, rec
+}
+
+func newHandler(t *testing.T, monadURL string, over func(*embed.Config)) http.Handler {
+	cfg := embed.Config{
+		APIKey:           "k",
+		APIBase:          monadURL,
+		FrameOrigin:      "https://app.monad.com/embed",
+		GetCustomerOrgID: func(r *http.Request) (string, error) { return "org_1", nil },
+	}
+	if over != nil {
+		over(&cfg)
+	}
+	return embed.Router(cfg)
+}
+
+func do(h http.Handler, method, path, body string) *httptest.ResponseRecorder {
+	var r *http.Request
+	if body != "" {
+		r = httptest.NewRequest(method, path, strings.NewReader(body))
+	} else {
+		r = httptest.NewRequest(method, path, nil)
+	}
+	rec := httptest.NewRecorder()
+	h.ServeHTTP(rec, r)
+	return rec
+}
+
+func TestGetConfig(t *testing.T) {
+	h := newHandler(t, "http://unused", nil)
+	rec := do(h, "GET", "/config", "")
+	if rec.Code != 200 {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	var got map[string]string
+	_ = json.Unmarshal(rec.Body.Bytes(), &got)
+	if got["frameOrigin"] != "https://app.monad.com/embed" {
+		t.Fatalf("frameOrigin = %q", got["frameOrigin"])
+	}
+}
+
+func TestConfigDefaults(t *testing.T) {
+	// APIBase + FrameOrigin omitted → production defaults.
+	h := embed.Router(embed.Config{
+		APIKey:           "k",
+		GetCustomerOrgID: func(r *http.Request) (string, error) { return "org_1", nil },
+	})
+	rec := do(h, "GET", "/config", "")
+	var got map[string]string
+	_ = json.Unmarshal(rec.Body.Bytes(), &got)
+	if got["apiBase"] != "https://app.monad.com/api" || got["frameOrigin"] != "https://app.monad.com/embed" {
+		t.Fatalf("defaults not applied: %v", got)
+	}
+}
+
+func TestSession(t *testing.T) {
+	monad, _ := mockMonad(t, 0)
+	defer monad.Close()
+	rec := do(newHandler(t, monad.URL, nil), "POST", "/session", "")
+	if rec.Code != 200 {
+		t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
+	}
+	var got map[string]string
+	_ = json.Unmarshal(rec.Body.Bytes(), &got)
+	if got["sessionToken"] != "tok" || got["organizationId"] != "org_1" {
+		t.Fatalf("session = %v", got)
+	}
+}
+
+func TestCatalogAllowList(t *testing.T) {
+	monad, _ := mockMonad(t, 0)
+	defer monad.Close()
+	h := newHandler(t, monad.URL, func(c *embed.Config) { c.CatalogAllow = []string{"aws-cloudtrail"} })
+	rec := do(h, "GET", "/catalog?kind=input", "")
+	if rec.Code != 200 {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	var got []map[string]string
+	_ = json.Unmarshal(rec.Body.Bytes(), &got)
+	if len(got) != 1 || got[0]["typeId"] != "aws-cloudtrail" {
+		t.Fatalf("catalog = %v", got)
+	}
+}
+
+func TestIngressDevNull(t *testing.T) {
+	monad, _ := mockMonad(t, 0)
+	defer monad.Close()
+	rec := do(newHandler(t, monad.URL, nil), "POST", "/pipelines/ingress", `{"inputId":"in_1","name":"CT"}`)
+	if rec.Code != 201 {
+		t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String())
+	}
+	var got map[string]any
+	_ = json.Unmarshal(rec.Body.Bytes(), &got)
+	if got["outputId"] != "out_devnull" || got["active"] != true {
+		t.Fatalf("built = %v", got)
+	}
+}
+
+func TestUnauthenticated(t *testing.T) {
+	h := newHandler(t, "http://unused", func(c *embed.Config) {
+		c.GetCustomerOrgID = func(r *http.Request) (string, error) { return "", nil }
+	})
+	rec := do(h, "POST", "/session", "")
+	if rec.Code != 401 {
+		t.Fatalf("status = %d", rec.Code)
+	}
+	var got map[string]string
+	_ = json.Unmarshal(rec.Body.Bytes(), &got)
+	if got["code"] != "unauthenticated" {
+		t.Fatalf("code = %q", got["code"])
+	}
+}
+
+func TestInvalidKind(t *testing.T) {
+	h := newHandler(t, "http://unused", nil)
+	rec := do(h, "GET", "/catalog?kind=nope", "")
+	if rec.Code != 400 {
+		t.Fatalf("status = %d", rec.Code)
+	}
+}
+
+func TestNotFound(t *testing.T) {
+	rec := do(newHandler(t, "http://unused", nil), "GET", "/nope", "")
+	if rec.Code != 404 {
+		t.Fatalf("status = %d", rec.Code)
+	}
+}
+
+// A connector's pipeline is resolved from `component_of` plus one detail fetch
+// for the peer — never by scanning the org's pipeline list, which pages at 10.
+func TestPipelineStatusUsesComponentOf(t *testing.T) {
+	for _, tc := range []struct {
+		name, query   string
+		wantPeerField string
+		wantPeer      string
+		wantEnabled   bool
+	}{
+		{"input", "connectorId=in_wired&kind=input", "outputId", "out_wired", true},
+		{"output", "connectorId=out_wired&kind=output", "inputId", "in_wired", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			monad, rec := mockMonad(t, 0)
+			defer monad.Close()
+			res := do(newHandler(t, monad.URL, nil), "GET", "/pipelines?"+tc.query, "")
+			if res.Code != 200 {
+				t.Fatalf("status = %d body=%s", res.Code, res.Body.String())
+			}
+			var got map[string]any
+			_ = json.Unmarshal(res.Body.Bytes(), &got)
+			if got["hasPipeline"] != true || got["pipelineId"] != "pipe_1" {
+				t.Fatalf("status = %v", got)
+			}
+			if got["enabled"] != tc.wantEnabled {
+				t.Fatalf("enabled = %v, want %v", got["enabled"], tc.wantEnabled)
+			}
+			if got[tc.wantPeerField] != tc.wantPeer {
+				t.Fatalf("%s = %v, want %q", tc.wantPeerField, got[tc.wantPeerField], tc.wantPeer)
+			}
+			if n := rec.count("GET /v2/org_1/pipelines?"); n != 0 {
+				t.Fatalf("listed pipelines %d times; component_of makes the scan unnecessary", n)
+			}
+		})
+	}
+}
+
+func TestPipelineStatusWithoutPipeline(t *testing.T) {
+	monad, _ := mockMonad(t, 0)
+	defer monad.Close()
+	res := do(newHandler(t, monad.URL, nil), "GET", "/pipelines?connectorId=in_bare&kind=input", "")
+	if res.Code != 200 {
+		t.Fatalf("status = %d", res.Code)
+	}
+	var got map[string]any
+	_ = json.Unmarshal(res.Body.Bytes(), &got)
+	if got["hasPipeline"] != false || got["enabled"] != false {
+		t.Fatalf("status = %v", got)
+	}
+}
+
+// A connector that does not exist is the contract's 404, not a generic 502.
+func TestPipelineStatusUnknownConnector(t *testing.T) {
+	monad, _ := mockMonad(t, 0)
+	defer monad.Close()
+	res := do(newHandler(t, monad.URL, nil), "GET", "/pipelines?connectorId=in_gone&kind=input", "")
+	if res.Code != 404 {
+		t.Fatalf("status = %d body=%s", res.Code, res.Body.String())
+	}
+	var got map[string]string
+	_ = json.Unmarshal(res.Body.Bytes(), &got)
+	if got["code"] != "not_found" {
+		t.Fatalf("code = %q", got["code"])
+	}
+}
+
+// PATCH is a true partial update, so the body carries the flag and nothing
+// else. Sending a rebuilt graph would drop whatever fields the round trip
+// failed to reproduce (config_overrides, schema_detection_spec, …).
+func TestSetEnabledSendsOnlyTheFlag(t *testing.T) {
+	monad, rec := mockMonad(t, 0)
+	defer monad.Close()
+	res := do(newHandler(t, monad.URL, nil), "POST", "/pipelines/state",
+		`{"pipelineId":"pipe_1","enabled":false}`)
+	if res.Code != 204 {
+		t.Fatalf("status = %d body=%s", res.Code, res.Body.String())
+	}
+	if got := rec.body["PATCH /v2/org_1/pipelines/pipe_1"]; got != `{"enabled":false}` {
+		t.Fatalf("PATCH body = %s, want only the enabled flag", got)
+	}
+	if n := rec.count("GET /v2/org_1/pipelines/pipe_1?"); n != 0 {
+		t.Fatalf("read the pipeline %d times; a partial PATCH needs no read-modify-write", n)
+	}
+}
+
+// `type` is the canonical field on output create; `output_type` is deprecated.
+func TestBuildDevNullSendsType(t *testing.T) {
+	monad, rec := mockMonad(t, 0)
+	defer monad.Close()
+	if res := do(newHandler(t, monad.URL, nil), "POST", "/pipelines/ingress",
+		`{"inputId":"in_1","name":"CT"}`); res.Code != 201 {
+		t.Fatalf("status = %d body=%s", res.Code, res.Body.String())
+	}
+	var body map[string]any
+	_ = json.Unmarshal([]byte(rec.body["POST /v2/org_1/outputs"]), &body)
+	if body["type"] != "dev-null" {
+		t.Fatalf("output create sent %v, want type=dev-null", body)
+	}
+	if _, legacy := body["output_type"]; legacy {
+		t.Fatalf("output create still sends the deprecated output_type: %v", body)
+	}
+}
+
+// Monad pages every list; the contract returns a bare array. The router has to
+// drain the pages or it truncates the tenant's inventory.
+func TestListConnectorsDrainsEveryPage(t *testing.T) {
+	const total = 450 // spans three pages at the client's page size
+	monad, rec := mockMonad(t, total)
+	defer monad.Close()
+	res := do(newHandler(t, monad.URL, nil), "GET", "/connectors?kind=input", "")
+	if res.Code != 200 {
+		t.Fatalf("status = %d body=%s", res.Code, res.Body.String())
+	}
+	var got []map[string]string
+	_ = json.Unmarshal(res.Body.Bytes(), &got)
+	if len(got) != total {
+		t.Fatalf("got %d connectors, want %d — pages were not drained", len(got), total)
+	}
+	seen := map[string]bool{}
+	for _, c := range got {
+		if seen[c["id"]] {
+			t.Fatalf("duplicate connector %q across pages", c["id"])
+		}
+		seen[c["id"]] = true
+	}
+	if n := rec.count("GET /v1/org_1/inputs?"); n < 2 {
+		t.Fatalf("made %d list calls for %d rows; expected several pages", n, total)
+	}
+}
+
+// An empty list comes back as null, not [], and must still yield an empty
+// array rather than a decode error or a JSON `null` body.
+func TestListConnectorsHandlesNullPage(t *testing.T) {
+	monad, _ := mockMonad(t, 0)
+	defer monad.Close()
+	res := do(newHandler(t, monad.URL, nil), "GET", "/connectors?kind=input", "")
+	if res.Code != 200 {
+		t.Fatalf("status = %d", res.Code)
+	}
+	if body := strings.TrimSpace(res.Body.String()); body != "[]" {
+		t.Fatalf("body = %s, want []", body)
+	}
+}
+
+// Removing an input tears down the pipeline and the input, and the throwaway
+// sink with them — but never the tenant's provisioned store.
+func TestRemoveKeepsProvisionedStore(t *testing.T) {
+	for _, tc := range []struct {
+		name        string
+		provisioned string
+		wantDeleted bool
+	}{
+		{"throwaway sink is removed", "", true},
+		{"provisioned store is kept", "out_wired", false},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			monad, rec := mockMonad(t, 0)
+			defer monad.Close()
+			h := newHandler(t, monad.URL, func(c *embed.Config) {
+				c.GetProvisionedComponents = func(string) embed.Provision {
+					return embed.Provision{DestinationOutputID: tc.provisioned}
+				}
+			})
+			if res := do(h, "POST", "/pipelines/remove",
+				`{"connectorId":"in_wired","kind":"input"}`); res.Code != 204 {
+				t.Fatalf("status = %d body=%s", res.Code, res.Body.String())
+			}
+			if rec.count("DELETE /v2/org_1/pipelines/pipe_1?") != 1 {
+				t.Fatal("the pipeline must be deleted first — it references the connectors")
+			}
+			if rec.count("DELETE /v1/org_1/inputs/in_wired?") != 1 {
+				t.Fatal("the input was not deleted")
+			}
+			if got := rec.count("DELETE /v1/org_1/outputs/out_wired?") == 1; got != tc.wantDeleted {
+				t.Fatalf("output deleted = %v, want %v", got, tc.wantDeleted)
+			}
+		})
+	}
+}
diff --git a/routers/go/go.mod b/routers/go/go.mod
new file mode 100644
index 0000000..02287dc
--- /dev/null
+++ b/routers/go/go.mod
@@ -0,0 +1,3 @@
+module github.com/monad-inc/embed/routers/go
+
+go 1.25