feat: per-connection URL params and per-pipeline env vars

- connections: params (JSON) merged into connection URL query string, applied
  to both sling env.yaml and atlas URLs; user's foreign_key_checks overrides
  the default =0
- pipelines: env map written to replication yaml top-level env:
- UI: key=value textarea in connection dialog and pipeline form (new/edit)
- docs: TiDB Error 8048 (SERIALIZABLE) troubleshooting — set
  tidb_skip_isolation_level_check=1 as a connection param

README.md

@@ -38,6 +38,11 @@ pnpm dev        # 或 pnpm build && pnpm start
 - `schema_sync` 开启时，Atlas 负责目标端表结构；此时即使选择 `full-refresh`，实际执行会翻译为 `truncate`（保留 Atlas 建的表结构重灌全量数据，避免 sling drop+重建丢掉索引/约束）。
 - `incremental` 需要为每张表配置 `primary_key` 和 `update_key`。
 - sling 连接 URL 自动带 `foreign_key_checks=0`（会话级），避免外键顺序问题；约束定义本身保留。
+- 连接支持自定义「连接参数」（URL query string，sling/Atlas 均生效），流水线支持自定义「环境变量」（写入 replication 的 `env:`）。
+
+## 已知问题
+
+- TiDB 目标库报 `Error 8048: isolation level 'SERIALIZABLE' is not supported`：见 [docs/tidb-serializable-isolation.md](docs/tidb-serializable-isolation.md)，给连接加参数 `tidb_skip_isolation_level_check=1` 即可。
 
 ## 数据与安全说明
 

app/api/connections/[id]/route.ts

@@ -1,4 +1,4 @@
-import { deleteConnection, updateConnection } from "@/lib/db";
+import { deleteConnection, updateConnection, normalizeJsonColumn } from "@/lib/db";
 
 export const runtime = "nodejs";
 export const dynamic = "force-dynamic";
@@ -9,6 +9,9 @@ export async function PUT(request: Request, { params }: Ctx) {
   try {
     const { id } = await params;
     const body = await request.json();
+    if (body.params !== undefined) {
+      body.params = normalizeJsonColumn(body.params);
+    }
     const conn = updateConnection(Number(id), body);
     if (!conn) {
       return Response.json({ error: "connection not found" }, { status: 404 });

app/api/connections/route.ts

@@ -1,4 +1,4 @@
-import { createConnection, listConnections } from "@/lib/db";
+import { createConnection, listConnections, normalizeJsonColumn } from "@/lib/db";
 
 export const runtime = "nodejs";
 export const dynamic = "force-dynamic";
@@ -28,6 +28,7 @@ export async function POST(request: Request) {
       user: String(body.user ?? ""),
       password: String(body.password ?? ""),
       database: String(body.database ?? ""),
+      params: normalizeJsonColumn(body.params),
     });
     return Response.json(conn, { status: 201 });
   } catch (err) {

app/api/pipelines/[id]/route.ts

@@ -1,4 +1,4 @@
-import { deletePipeline, getPipeline, updatePipeline } from "@/lib/db";
+import { deletePipeline, getPipeline, updatePipeline, normalizeJsonColumn } from "@/lib/db";
 
 export const runtime = "nodejs";
 export const dynamic = "force-dynamic";
@@ -42,6 +42,9 @@ export async function PUT(request: Request, { params }: Ctx) {
     if (body.streams !== undefined) {
       patch.streams = normalizeStreams(body.streams);
     }
+    if (body.env !== undefined) {
+      patch.env = normalizeJsonColumn(body.env);
+    }
     const pipeline = updatePipeline(Number(id), patch);
     if (!pipeline) {
       return Response.json({ error: "pipeline not found" }, { status: 404 });

app/api/pipelines/route.ts

@@ -1,4 +1,4 @@
-import { createPipeline, listPipelines } from "@/lib/db";
+import { createPipeline, listPipelines, normalizeJsonColumn } from "@/lib/db";
 
 export const runtime = "nodejs";
 export const dynamic = "force-dynamic";
@@ -34,6 +34,7 @@ export async function POST(request: Request) {
       include_fk: body.include_fk === undefined ? 0 : Number(body.include_fk),
       mode: body.mode ? String(body.mode) : "full-refresh",
       streams: normalizeStreams(body.streams),
+      env: normalizeJsonColumn(body.env),
     });
     return Response.json(pipeline, { status: 201 });
   } catch (err) {

app/connections/page.tsx

@@ -117,7 +117,18 @@ export default function ConnectionsPage() {
               {list.map((c) => (
                 <Fragment key={c.id}>
                   <TableRow>
-                    <TableCell className="font-medium">{c.name}</TableCell>
+                    <TableCell className="font-medium">
+                      {c.name}
+                      {c.params && c.params !== "{}" ? (
+                        <div className="mt-0.5 font-mono text-xs text-muted-foreground">
+                          {Object.entries(
+                            JSON.parse(c.params) as Record<string, string>
+                          )
+                            .map(([k, v]) => `${k}=${v}`)
+                            .join(" ")}
+                        </div>
+                      ) : null}
+                    </TableCell>
                     <TableCell className="font-mono text-xs">{c.type}</TableCell>
                     <TableCell className="font-mono text-xs">
                       {c.host}:{c.port}

app/pipelines/[id]/page.tsx

@@ -91,6 +91,22 @@ export default function PipelineDetailPage() {
     }
   }, [pipeline]);
 
+  const envEntries = useMemo<[string, string][]>(() => {
+    if (!pipeline) return [];
+    try {
+      const v = JSON.parse(pipeline.env || "{}") as unknown;
+      if (v && typeof v === "object" && !Array.isArray(v)) {
+        return Object.entries(v as Record<string, unknown>).map(([k, val]) => [
+          k,
+          String(val),
+        ]);
+      }
+    } catch {
+      // fall through
+    }
+    return [];
+  }, [pipeline]);
+
   const latestRun = runs[0] ?? null;
 
   async function runNow() {
@@ -203,6 +219,14 @@ export default function PipelineDetailPage() {
               <dt className="text-muted-foreground">包含外键</dt>
               <dd className="font-medium">{pipeline.include_fk ? "是" : "否"}</dd>
             </div>
+            {envEntries.length > 0 ? (
+              <div>
+                <dt className="text-muted-foreground">环境变量</dt>
+                <dd className="font-mono text-xs">
+                  {envEntries.map(([k, v]) => `${k}=${v}`).join(" ")}
+                </dd>
+              </div>
+            ) : null}
             <div>
               <dt className="text-muted-foreground">创建时间</dt>
               <dd>{fmtTime(pipeline.created_at)}</dd>

components/connection-dialog.tsx

@@ -3,6 +3,7 @@
 import { useState } from "react";
 import { toast } from "sonner";
 import { api, type Connection } from "@/components/api";
+import { keyValueLinesFromJson, parseKeyValueLines } from "@/components/kv";
 import { Button } from "@/components/ui/button";
 import {
   Dialog,
@@ -20,6 +21,7 @@ import {
   SelectTrigger,
   SelectValue,
 } from "@/components/ui/select";
+import { Textarea } from "@/components/ui/textarea";
 
 interface ConnectionDialogProps {
   open: boolean;
@@ -59,6 +61,9 @@ export function ConnectionDialog({
       : EMPTY
   );
   const [saving, setSaving] = useState(false);
+  const [paramsText, setParamsText] = useState(() =>
+    initial ? keyValueLinesFromJson(initial.params) : ""
+  );
 
   const set = (k: keyof typeof EMPTY) => (e: React.ChangeEvent<HTMLInputElement>) =>
     setForm((f) => ({ ...f, [k]: e.target.value }));
@@ -79,6 +84,7 @@ export function ConnectionDialog({
         user: form.user.trim(),
         password: form.password,
         database: form.database.trim(),
+        params: parseKeyValueLines(paramsText),
       });
       if (initial) {
         await api(`/api/connections/${initial.id}`, {
@@ -174,6 +180,21 @@ export function ConnectionDialog({
               onChange={set("database")}
             />
           </div>
+          <div className="grid gap-1.5">
+            <Label htmlFor="conn-params">连接参数</Label>
+            <Textarea
+              id="conn-params"
+              className="font-mono text-xs"
+              rows={3}
+              value={paramsText}
+              onChange={(e) => setParamsText(e.target.value)}
+              placeholder={"tidb_skip_isolation_level_check=1"}
+            />
+            <p className="text-xs text-muted-foreground">
+              每行一个 key=value，作为 URL 查询参数传递，sling / atlas 均生效；TiDB
+              目标库建议设置 tidb_skip_isolation_level_check=1。
+            </p>
+          </div>
         </div>
         <DialogFooter>
           <Button variant="outline" onClick={() => onOpenChange(false)} disabled={saving}>

components/kv.ts

@@ -0,0 +1,34 @@
+// Helpers for the "one key=value per line" textareas used by the connection
+// dialog (URL params) and the pipeline form (replication env).
+
+// Parse textarea lines into an object. Blank lines, lines starting with '#'
+// and lines without a key or '=' are ignored. Values may contain '='.
+export function parseKeyValueLines(text: string): Record<string, string> {
+  const out: Record<string, string> = {};
+  for (const raw of text.split("\n")) {
+    const line = raw.trim();
+    if (!line || line.startsWith("#")) continue;
+    const i = line.indexOf("=");
+    if (i <= 0) continue;
+    const key = line.slice(0, i).trim();
+    const value = line.slice(i + 1).trim();
+    if (key) out[key] = value;
+  }
+  return out;
+}
+
+// Serialize a JSON-object column back into key=value lines for editing.
+export function keyValueLinesFromJson(json: string | undefined | null): string {
+  if (!json) return "";
+  try {
+    const v = JSON.parse(json) as unknown;
+    if (v && typeof v === "object" && !Array.isArray(v)) {
+      return Object.entries(v as Record<string, unknown>)
+        .map(([k, val]) => `${k}=${String(val)}`)
+        .join("\n");
+    }
+  } catch {
+    // fall through
+  }
+  return "";
+}

components/pipeline-form.tsx

@@ -10,6 +10,7 @@ import {
   type StreamConfig,
   type StreamInfo,
 } from "@/components/api";
+import { keyValueLinesFromJson, parseKeyValueLines } from "@/components/kv";
 import { Button } from "@/components/ui/button";
 import {
   Card,
@@ -29,6 +30,7 @@ import {
   SelectValue,
 } from "@/components/ui/select";
 import { Switch } from "@/components/ui/switch";
+import { Textarea } from "@/components/ui/textarea";
 
 const MODES = [
   { value: "full-refresh", label: "full-refresh（全量覆盖，默认）" },
@@ -58,6 +60,9 @@ export function PipelineForm({ initial }: { initial?: Pipeline }) {
     initial?.schema_scope ?? "selected"
   );
   const [includeFk, setIncludeFk] = useState(initial ? !!initial.include_fk : false);
+  const [envText, setEnvText] = useState(() =>
+    initial ? keyValueLinesFromJson(initial.env) : ""
+  );
 
   const [streamList, setStreamList] = useState<StreamInfo[] | null>(null);
   const [streamsLoading, setStreamsLoading] = useState(false);
@@ -149,6 +154,7 @@ export function PipelineForm({ initial }: { initial?: Pipeline }) {
       include_fk: includeFk ? 1 : 0,
       mode,
       streams: JSON.stringify(selectedList),
+      env: parseKeyValueLines(envText),
     };
     try {
       const p = isEdit
@@ -301,6 +307,21 @@ export function PipelineForm({ initial }: { initial?: Pipeline }) {
               disabled={!schemaSync}
             />
           </div>
+          <div className="grid gap-1.5">
+            <Label htmlFor="p-env">环境变量</Label>
+            <Textarea
+              id="p-env"
+              className="font-mono text-xs"
+              rows={3}
+              value={envText}
+              onChange={(e) => setEnvText(e.target.value)}
+              placeholder={"SLING_THREADS=4"}
+            />
+            <p className="text-xs text-muted-foreground">
+              每行一个 key=value，写入 sling replication 的顶层 env（sling 会导出为进程环境变量），如
+              SLING_THREADS=4。
+            </p>
+          </div>
         </CardContent>
       </Card>
 

docs/tidb-serializable-isolation.md

@@ -0,0 +1,53 @@
+# TiDB 目标库：Error 8048 isolation level 'SERIALIZABLE' is not supported
+
+## 现象
+
+目标库为 TiDB 时，sling 数据迁移在每个 stream 的写入阶段失败：
+
+```
+writing to target database [mode: truncate]
+created table `test_demo`.`crm_orders_tmp`
+execution failed
+Error 8048 (HY000): The isolation level 'SERIALIZABLE' is not supported.
+Set tidb_skip_isolation_level_check=1 to skip this error
+```
+
+## 原因
+
+TiDB 不支持 `SERIALIZABLE` 隔离级别（TiDB 默认使用乐观锁/快照隔离）。sling
+在目标端开启事务时请求了 `SERIALIZABLE`，TiDB 默认会直接报错拒绝。
+
+TiDB 提供了兼容开关 `tidb_skip_isolation_level_check=1`：设置后跳过隔离级别
+检查，按 TiDB 实际支持的级别执行。对"测试环境 → 本地 dev 库"的全量数据灌入
+场景，这是安全的。
+
+## 解决：连接参数
+
+在 sling-ui 中编辑 TiDB 目标连接，在「连接参数」中加一行：
+
+```
+tidb_skip_isolation_level_check=1
+```
+
+### 工作原理
+
+连接参数会被拼进连接 URL 的 query string，例如：
+
+```
+mysql://user:pass@tidb-host:4000/dbname?tidb_skip_isolation_level_check=1
+```
+
+sling 底层使用 go-sql-driver/mysql，会把未识别的 DSN 参数作为 session 变量
+在连接建立时执行 `SET <key>=<value>`（本项目内置的 `foreign_key_checks=0`
+也是同一机制）。Atlas 的连接 URL 同样会带上这些参数。
+
+已实测验证：对普通 MySQL 设置该参数会报 `Error 1193: Unknown system
+variable`，证明参数确实被 SET 为 session 变量；TiDB 上该变量存在，即达到
+预期效果。
+
+## 备注
+
+- 该参数只对 TiDB 有效，不要给普通 MySQL/MariaDB 连接设置。
+- 若希望对整个 TiDB 实例生效（所有客户端），也可以在 TiDB 上执行
+  `SET GLOBAL tidb_skip_isolation_level_check=1;`，但用连接参数的方式
+  影响面最小，推荐。

lib/atlas.ts

@@ -1,6 +1,6 @@
 import { execFile } from "child_process";
 import { Connection, Pipeline, StreamConfig, getConnection } from "./db";
-import { connectionUrl, discoverStreams, execSql } from "./sling";
+import { appendParams, connectionUrl, discoverStreams, execSql } from "./sling";
 import { resolveBinary } from "./binaries";
 
 // Atlas needs an empty scratch database ("dev database") to plan diffs.
@@ -10,10 +10,19 @@ import { resolveBinary } from "./binaries";
 // scratch schema on the target server instead — no Docker needed.
 const DEV_SCHEMA = "sling_ui_atlas_dev";
 
+// The scratch schema lives on the target server, so it inherits the target
+// connection's URL params (e.g. tidb_skip_isolation_level_check for TiDB).
 function devDatabaseUrl(tgt: Connection): string {
   const user = encodeURIComponent(tgt.user);
   const pass = encodeURIComponent(tgt.password);
-  return `mysql://${user}:${pass}@${tgt.host}:${tgt.port}/${DEV_SCHEMA}`;
+  const base = `mysql://${user}:${pass}@${tgt.host}:${tgt.port}/${DEV_SCHEMA}`;
+  let params: Record<string, string> = {};
+  try {
+    params = JSON.parse(tgt.params || "{}") as Record<string, string>;
+  } catch {
+    // ignore malformed params
+  }
+  return appendParams(base, params);
 }
 
 // Create the scratch schema if missing and return its URL. Atlas keeps it

lib/db.ts

@@ -14,6 +14,7 @@ export interface Connection {
   user: string;
   password: string;
   database: string;
+  params: string; // JSON object: URL query params, e.g. {"tidb_skip_isolation_level_check":"1"}
   created_at: string;
 }
 
@@ -25,6 +26,7 @@ export interface ConnectionInput {
   user: string;
   password: string;
   database: string;
+  params?: string;
 }
 
 export interface StreamConfig {
@@ -45,6 +47,7 @@ export interface Pipeline {
   include_fk: number; // 1 | 0
   mode: string;
   streams: string; // JSON: StreamConfig[]
+  env: string; // JSON object: sling replication env, e.g. {"SLING_THREADS":"4"}
   created_at: string;
   updated_at: string;
 }
@@ -58,10 +61,18 @@ export interface PipelineInput {
   include_fk?: number;
   mode?: string;
   streams?: string;
+  env?: string;
 }
 
 export type RunStatus = "queued" | "running" | "success" | "failed";
 
+// Normalize a JSON-object column value from an API body: accept either a
+// plain object or a pre-serialized JSON string.
+export function normalizeJsonColumn(raw: unknown): string {
+  if (typeof raw === "string") return raw;
+  return JSON.stringify(raw ?? {});
+}
+
 export interface Run {
   id: number;
   pipeline_id: number;
@@ -87,6 +98,7 @@ function createDb(): Database.Database {
       user TEXT NOT NULL,
       password TEXT NOT NULL DEFAULT '',
       database TEXT NOT NULL,
+      params TEXT NOT NULL DEFAULT '{}',
       created_at TEXT DEFAULT (datetime('now'))
     );
     CREATE TABLE IF NOT EXISTS pipelines (
@@ -99,6 +111,7 @@ function createDb(): Database.Database {
       include_fk INTEGER NOT NULL DEFAULT 0,
       mode TEXT DEFAULT 'full-refresh',
       streams TEXT NOT NULL DEFAULT '[]',
+      env TEXT NOT NULL DEFAULT '{}',
       created_at TEXT DEFAULT (datetime('now')),
       updated_at TEXT DEFAULT (datetime('now'))
     );
@@ -116,9 +129,9 @@ function createDb(): Database.Database {
   // Lightweight migrations for databases created before these columns existed.
   // Several Next.js workers may open the DB concurrently during build, so a
   // racing ALTER can fail with "duplicate column name" — safe to ignore.
-  const addColumnIfMissing = (sql: string, column: string) => {
+  const addColumnIfMissing = (table: string, sql: string, column: string) => {
     const cols = new Set(
-      (db.pragma("table_info(pipelines)") as { name: string }[]).map(
+      (db.pragma(`table_info(${table})`) as { name: string }[]).map(
         (c) => c.name
       )
     );
@@ -130,13 +143,25 @@ function createDb(): Database.Database {
     }
   };
   addColumnIfMissing(
+    "pipelines",
     "ALTER TABLE pipelines ADD COLUMN schema_scope TEXT NOT NULL DEFAULT 'selected'",
     "schema_scope"
   );
   addColumnIfMissing(
+    "pipelines",
     "ALTER TABLE pipelines ADD COLUMN include_fk INTEGER NOT NULL DEFAULT 0",
     "include_fk"
   );
+  addColumnIfMissing(
+    "connections",
+    "ALTER TABLE connections ADD COLUMN params TEXT NOT NULL DEFAULT '{}'",
+    "params"
+  );
+  addColumnIfMissing(
+    "pipelines",
+    "ALTER TABLE pipelines ADD COLUMN env TEXT NOT NULL DEFAULT '{}'",
+    "env"
+  );
   return db;
 }
 
@@ -164,8 +189,8 @@ export function getConnection(id: number): Connection | null {
 export function createConnection(input: ConnectionInput): Connection {
   const res = db
     .prepare(
-      `INSERT INTO connections (name, type, host, port, user, password, database)
+      `INSERT INTO connections (name, type, host, port, user, password, database, params)
-       VALUES (@name, @type, @host, @port, @user, @password, @database)`
+       VALUES (@name, @type, @host, @port, @user, @password, @database, @params)`
     )
     .run({
       name: input.name,
@@ -175,6 +200,7 @@ export function createConnection(input: ConnectionInput): Connection {
       user: input.user,
       password: input.password ?? "",
       database: input.database,
+      params: input.params ?? "{}",
     });
   return getConnection(Number(res.lastInsertRowid))!;
 }
@@ -183,7 +209,7 @@ export function updateConnection(
   id: number,
   input: Partial<ConnectionInput>
 ): Connection | null {
-  const fields = ["name", "type", "host", "port", "user", "password", "database"].filter(
+  const fields = ["name", "type", "host", "port", "user", "password", "database", "params"].filter(
     (k) => input[k as keyof ConnectionInput] !== undefined
   );
   if (fields.length === 0) return getConnection(id);
@@ -216,8 +242,8 @@ export function getPipeline(id: number): Pipeline | null {
 export function createPipeline(input: PipelineInput): Pipeline {
   const res = db
     .prepare(
-      `INSERT INTO pipelines (name, source_conn_id, target_conn_id, schema_sync, schema_scope, include_fk, mode, streams)
+      `INSERT INTO pipelines (name, source_conn_id, target_conn_id, schema_sync, schema_scope, include_fk, mode, streams, env)
-       VALUES (@name, @source_conn_id, @target_conn_id, @schema_sync, @schema_scope, @include_fk, @mode, @streams)`
+       VALUES (@name, @source_conn_id, @target_conn_id, @schema_sync, @schema_scope, @include_fk, @mode, @streams, @env)`
     )
     .run({
       name: input.name,
@@ -228,6 +254,7 @@ export function createPipeline(input: PipelineInput): Pipeline {
       include_fk: input.include_fk ?? 0,
       mode: input.mode ?? "full-refresh",
       streams: input.streams ?? "[]",
+      env: input.env ?? "{}",
     });
   return getPipeline(Number(res.lastInsertRowid))!;
 }
@@ -245,6 +272,7 @@ export function updatePipeline(
     "include_fk",
     "mode",
     "streams",
+    "env",
   ].filter((k) => input[k as keyof PipelineInput] !== undefined);
   if (fields.length === 0) return getPipeline(id);
   const sets = fields.map((f) => `${f} = @${f}`).join(", ");

lib/sling.ts

@@ -19,24 +19,62 @@ export interface DiscoveredStream {
   columns?: string[];
 }
 
-// Build a mysql:// URL from a connection record, escaping credentials.
+// Parse a connection/pipeline JSON-object column; tolerate bad data.
+function parseJsonObject(json: string | undefined | null): Record<string, string> {
+  if (!json) return {};
+  try {
+    const v = JSON.parse(json) as unknown;
+    if (v && typeof v === "object" && !Array.isArray(v)) {
+      return Object.fromEntries(
+        Object.entries(v as Record<string, unknown>).map(([k, val]) => [
+          k,
+          String(val),
+        ])
+      );
+    }
+  } catch {
+    // fall through
+  }
+  return {};
+}
+
+// Append query params to a URL, percent-encoding keys and values.
+export function appendParams(url: string, params: Record<string, string>): string {
+  const entries = Object.entries(params);
+  if (entries.length === 0) return url;
+  const qs = entries
+    .map(([k, v]) => `${encodeURIComponent(k)}=${encodeURIComponent(v)}`)
+    .join("&");
+  return `${url}${url.includes("?") ? "&" : "?"}${qs}`;
+}
+
+// Build a mysql:// URL from a connection record, escaping credentials and
+// merging the connection's extra params into the query string. These params
+// reach both sling (env.yaml) and atlas (--url/--to/--dev-url); the
+// go-sql-driver SETs unknown DSN params as session variables, which is how
+// e.g. tidb_skip_isolation_level_check=1 takes effect.
 export function connectionUrl(conn: Connection): string {
   const user = encodeURIComponent(conn.user);
   const pass = encodeURIComponent(conn.password);
-  return `mysql://${user}:${pass}@${conn.host}:${conn.port}/${conn.database}`;
+  const base = `mysql://${user}:${pass}@${conn.host}:${conn.port}/${conn.database}`;
+  return appendParams(base, parseJsonObject(conn.params));
 }
 
 // Regenerate env.yaml from the connections table. Names are uppercased
 // because sling treats connection names case-insensitively. Session-level
 // foreign_key_checks=0 lets sling truncate/drop FK-referenced parent tables
-// in any order (same approach as mysqldump); constraints stay defined.
+// in any order (same approach as mysqldump); constraints stay defined. It is
+// appended only here (atlas doesn't need it) and never overrides a value the
+// user set explicitly in connection params.
 export function syncEnvYaml(): string {
   fs.mkdirSync(SLING_HOME, { recursive: true });
   const connections: Record<string, { url: string }> = {};
   for (const c of listConnections()) {
-    connections[c.name.toUpperCase()] = {
+    let url = connectionUrl(c);
-      url: `${connectionUrl(c)}?foreign_key_checks=0`,
+    if (!parseJsonObject(c.params)["foreign_key_checks"]) {
-    };
+      url = appendParams(url, { foreign_key_checks: "0" });
+    }
+    connections[c.name.toUpperCase()] = { url };
   }
   const file = path.join(SLING_HOME, "env.yaml");
   fs.writeFileSync(file, dump({ connections }));
@@ -236,7 +274,9 @@ export function buildReplicationYaml(
       object: `${tgt.database}.{stream_table}`,
     },
     streams: streamsYaml,
-    env: { SLING_THREADS: 4 },
+    // Pipeline-level env is exported as process env vars by sling. User
+    // values override the built-in defaults (e.g. SLING_THREADS).
+    env: { SLING_THREADS: "4", ...parseJsonObject(pipeline.env) },
   };
 
   const dir = path.join(DATA_DIR, "replications");