CLAUDE.md

@@ -100,10 +100,6 @@ Passwords in connections are stored as `$VAR_NAME` references. At run time they
 
 Each driver in `pipekit/drivers/` inherits from `base.py::Driver` and implements: `browse_fields`, `list_tables`, `list_schemas`, `get_columns`, `map_type`, `default_expression`, `quote_identifier`. The wizard UI calls `/api/introspect/*` which dispatches to the appropriate driver.
 
-## Wizard Entry Modes
-
-Step 1 offers two paths. **Browse a table** (`/wizard/tables` → `/wizard/columns`) is the original flow: pick one source table, introspect its columns. **Write SQL** (`/wizard/sql` → `POST /wizard/sql/columns`) starts from an arbitrary query — the source query is used verbatim and its result columns are discovered by `Driver.introspect_query_columns`, which runs the query wrapped to fetch ~no rows (`_zero_row_wrap`: derived-table + `WHERE 1=0` by default; DB2 appends `FETCH FIRST 1 ROW ONLY` since DB2 for i forbids `WITH` inside a nested table expression). Both paths land on `wizard_step3.html` (the SQL path passes `sql_mode=True`) and POST to `/wizard/create`, which branches on `entry_mode`. SQL-mode dest types default to `text` — there's no result-set type metadata over jrunner's CSV output, so the user adjusts types by hand. `columns_json` is never read at run time (the engine does `CREATE staging LIKE dest` + `SELECT *`); it only drives the dest `CREATE TABLE`.
-
 ## Module Columns
 
 Modules store their column mapping as `columns_json` — a JSON list of dicts with keys `source_name`, `source_type`, `dest_name`, `dest_type`. The engine uses this to build the staging CREATE TABLE and the merge INSERT column lists.

pipekit/api/app.py

@@ -7,100 +7,14 @@ content-negotiation complexity and keeps the API curl-testable.
 
 from __future__ import annotations
 
-import logging
-import sys
 from contextlib import asynccontextmanager
 
-from fastapi import FastAPI, Request
+from fastapi import FastAPI
-from fastapi.exception_handlers import http_exception_handler
-from starlette.exceptions import HTTPException as StarletteHTTPException
 
 from .. import __version__, db, jrunner
 from ..web import mount_web
 from .routes import connections, introspect, modules, runs, system
 
-_log = logging.getLogger("pipekit.web")
-
-
-def _configure_logging() -> None:
-    """Route pipekit's loggers to stderr (→ journal) with a consistent format.
-
-    Uvicorn configures only its own loggers, not the root, so without this our
-    records would fall through to the last-resort handler (WARNING+ only) and
-    INFO breadcrumbs would vanish. Idempotent — safe to call per create_app."""
-    base = logging.getLogger("pipekit")
-    if not base.handlers:
-        handler = logging.StreamHandler(sys.stderr)
-        handler.setFormatter(logging.Formatter(
-            "%(asctime)s %(levelname)s [%(name)s] %(message)s"))
-        base.addHandler(handler)
-        base.setLevel(logging.INFO)
-        base.propagate = False
-
-
-_SECRET_KEYS = ("password", "passwd", "pwd", "secret", "token")
-
-
-def _redact_body(raw: bytes, content_type: str, limit: int = 2000) -> str:
-    """Decode a captured request body for logging, masking secret-looking
-    fields (passwords etc.) so they never reach the journal."""
-    if not raw:
-        return ""
-    text = raw.decode("utf-8", "replace")
-    ct = (content_type or "").lower()
-    try:
-        if "application/json" in ct:
-            import json as _json
-            obj = _json.loads(text)
-            if isinstance(obj, dict):
-                obj = {k: ("***" if any(s in k.lower() for s in _SECRET_KEYS) else v)
-                       for k, v in obj.items()}
-                text = _json.dumps(obj)
-        else:
-            from urllib.parse import parse_qsl, urlencode
-            pairs = parse_qsl(text, keep_blank_values=True)
-            if pairs:
-                text = urlencode([(k, "***" if any(s in k.lower() for s in _SECRET_KEYS) else v)
-                                  for k, v in pairs])
-    except Exception:                                                # noqa: BLE001
-        pass
-    return text[:limit] + ("…[truncated]" if len(text) > limit else "")
-
-
-class _RequestBodyCapture:
-    """Buffer each HTTP request body onto the scope (``pk_raw_body``) so the
-    exception handler can log the submitted payload on failure, then replay it
-    downstream so route handlers still read the body normally. Bodies here are
-    small form/JSON posts — no large uploads go through HTTP."""
-
-    def __init__(self, app):
-        self.app = app
-
-    async def __call__(self, scope, receive, send):
-        if scope["type"] != "http":
-            await self.app(scope, receive, send)
-            return
-        body = bytearray()
-        while True:
-            message = await receive()
-            if message["type"] == "http.request":
-                body += message.get("body", b"")
-                if not message.get("more_body", False):
-                    break
-            else:  # http.disconnect
-                break
-        scope["pk_raw_body"] = bytes(body)
-        replayed = False
-
-        async def replay():
-            nonlocal replayed
-            if not replayed:
-                replayed = True
-                return {"type": "http.request", "body": bytes(body), "more_body": False}
-            return {"type": "http.disconnect"}
-
-        await self.app(scope, replay, send)
-
 
 @asynccontextmanager
 async def _lifespan(app: FastAPI):
@@ -110,28 +24,7 @@ async def _lifespan(app: FastAPI):
 
 
 def create_app() -> FastAPI:
-    _configure_logging()
     app = FastAPI(title="Pipekit", version=__version__, lifespan=_lifespan)
-    app.add_middleware(_RequestBodyCapture)
-
-    @app.exception_handler(StarletteHTTPException)
-    async def _logged_http_exception_handler(request: Request,
-                                             exc: StarletteHTTPException):
-        # FastAPI turns HTTPException into a normal response and never logs it,
-        # so failures surfaced this way (e.g. wizard dest-provisioning errors)
-        # were invisible in the journal. Log them — with the submitted payload,
-        # secrets masked — then defer to the default handler for the response.
-        body = _redact_body(request.scope.get("pk_raw_body", b""),
-                            request.headers.get("content-type", ""))
-        if exc.status_code >= 500:
-            _log.error("%s %s -> %d: %s | body: %s", request.method,
-                       request.url.path, exc.status_code, exc.detail, body,
-                       exc_info=exc)
-        elif exc.status_code >= 400:
-            _log.warning("%s %s -> %d: %s | body: %s", request.method,
-                         request.url.path, exc.status_code, exc.detail, body)
-        return await http_exception_handler(request, exc)
-
     app.include_router(system.router)
     app.include_router(connections.router, prefix="/api")
     app.include_router(introspect.router, prefix="/api")

pipekit/drivers/base.py

@@ -167,30 +167,6 @@ class Driver(abc.ABC):
         """DDL to create new_table with the same columns as source_table."""
         return f"CREATE TABLE {new_table} (LIKE {source_table} INCLUDING ALL);"
 
-    def build_add_column_sql(self, qualified_table: str, column: dict) -> str:
-        """DDL to append one column to an existing table. ALTER can only add
-        at the end — which keeps the positional load aligned as long as the
-        column is also last in the source projection."""
-        name = column["dest_name"]
-        dtype = (column.get("dest_type") or "text").strip()
-        if not dtype:
-            raise ValueError(f"column {name!r} has no dest_type")
-        return (f"ALTER TABLE {qualified_table} "
-                f"ADD COLUMN {self.quote_identifier(name)} {dtype};")
-
-    def column_inventory(self, conn: dict, schema: str,
-                          table: str) -> "list[dict]":
-        """Ordered [{name, type}] of an existing dest table, or [] if absent.
-        Default uses INFORMATION_SCHEMA (PG/MSSQL); override for other catalogs."""
-        sql = (
-            "SELECT column_name, data_type FROM information_schema.columns "
-            f"WHERE table_schema='{schema}' AND table_name='{table}' "
-            "ORDER BY ordinal_position"
-        )
-        r = self.query(conn, sql)
-        return [{"name": row[0].strip(), "type": row[1].strip()}
-                for row in r.rows if row and row[0]]
-
     def check_dest_table(self, conn: dict, schema: str,
                           table: str) -> "set[str] | None":
         """Return lowercase column names of an existing dest table, or None
@@ -205,24 +181,6 @@ class Driver(abc.ABC):
             return None
         return {row[0].strip().lower() for row in r.rows if row and row[0]}
 
-    # ---- Arbitrary-query introspection (SQL-entry wizard path) ----
-    def introspect_query_columns(self, conn: dict, sql: str) -> list[str]:
-        """Return the result-set column names of an arbitrary user query,
-        fetching (near) zero rows. Used by the wizard's SQL-entry path to
-        seed the column-mapping grid when the source isn't a single table."""
-        wrapped = self._zero_row_wrap(sql)
-        r = self.query(conn, wrapped)
-        return [(c or "").strip() for c in r.columns]
-
-    def _zero_row_wrap(self, sql: str) -> str:
-        """Wrap an arbitrary query so it yields its columns but ~no rows.
-        Generic default wraps it as a derived table with a false predicate;
-        this works on dialects where a sub-SELECT may carry its own WITH
-        (e.g. PostgreSQL). Dialects that forbid WITH inside a nested table
-        expression (DB2 for i, MSSQL) override this."""
-        body = sql.strip().rstrip(";").rstrip()
-        return f"SELECT * FROM (\n{body}\n) AS _pk_introspect WHERE 1=0"
-
     # ---- Shared helper ----
     def query(self, conn: dict, sql: str) -> jrunner.QueryResult:
         """Run `sql` in jrunner query mode against `conn`."""

pipekit/drivers/db2.py

@@ -115,13 +115,6 @@ class DB2Driver(Driver):
         v = result.rows[0][0].strip()
         return v or None
 
-    def _zero_row_wrap(self, sql: str) -> str:
-        # DB2 for i forbids a WITH clause inside a nested table expression,
-        # so we can't wrap the query as a derived table. Append a fetch-limit
-        # to the outer statement instead. (0 is rejected — must be positive.)
-        body = sql.strip().rstrip(";").rstrip()
-        return f"{body}\nFETCH FIRST 1 ROW ONLY"
-
     def qualified_table_name(self, table: str, *, schema: str) -> str:
         return f"{self.quote_identifier(schema)}.{self.quote_identifier(table)}"
 

pipekit/jrunner.py

@@ -254,13 +254,10 @@ _EXCEPTION_HEADER_RE = re.compile(
 
 # jrunner runs query-mode SQL with `executeQuery`, which requires the
 # statement to produce a ResultSet. DDL/DML (CREATE, TRUNCATE, INSERT)
-# still executes, but the driver then complains there's no ResultSet. The
+# still executes, but PG then throws "No results were returned by the
-# statement succeeded — ignore the trace. Each dialect words it differently:
+# query." The statement succeeded — ignore the trace.
-#   PG:        "No results were returned by the query"
-#   SQL Server:"The statement did not return a result set."
 _BENIGN_EXCEPTION_SUBSTRINGS = (
     "No results were returned by the query",
-    "The statement did not return a result set",
 )
 
 

pipekit/repo.py

@@ -232,14 +232,6 @@ def update_module(module_id: int, *, name: str | None = None,
     return get_module(module_id)
 
 
-def update_module_columns(module_id: int, columns: list[dict]) -> dict | None:
-    """Replace a module's stored column listing (columns_json)."""
-    with db.connect() as c:
-        c.execute("UPDATE module SET columns_json=?, updated_at=datetime('now') "
-                  "WHERE id=?", (json.dumps(columns), module_id))
-    return get_module(module_id)
-
-
 class ModuleRunning(RuntimeError):
     """Raised by delete_module when the module is currently running."""
 

pipekit/web/app.py

@@ -541,106 +541,15 @@ def wizard_step3(request: Request,
     )
 
 
-@_router.get("/wizard/sql", response_class=HTMLResponse)
-def wizard_sql_entry(request: Request,
-                     source_connection_id: int = Query(...)):
-    """Alternate step 2 — write an arbitrary source query instead of
-    browsing a single table."""
-    conn = repo.get_connection(source_connection_id)
-    if conn is None:
-        raise HTTPException(404, f"connection id={source_connection_id} not found")
-    drv = _driver_for_conn(conn)
-    if drv is None:
-        raise HTTPException(500, "driver row missing for connection")
-    return _templates.TemplateResponse(
-        request,
-        "wizard_sql.html",
-        _ctx(step=2, mode="sql", connection=conn, driver_kind=drv.kind,
-             source_query=""),
-    )
-
-
-@_router.post("/wizard/sql/columns", response_class=HTMLResponse)
-async def wizard_sql_columns(request: Request):
-    """Introspect an arbitrary query's result columns and render the
-    column-mapping grid (SQL-entry variant of step 3)."""
-    form = await request.form()
-    source_connection_id = int(form["source_connection_id"])
-    source_query = (form.get("source_query") or "").strip()
-
-    conn = repo.get_connection(source_connection_id)
-    if conn is None:
-        raise HTTPException(404, f"connection id={source_connection_id} not found")
-    drv = _driver_for_conn(conn)
-    if drv is None:
-        raise HTTPException(500, "driver row missing for connection")
-
-    if not source_query:
-        return _templates.TemplateResponse(
-            request, "wizard_sql.html",
-            _ctx(step=2, mode="sql", connection=conn, driver_kind=drv.kind,
-                 source_query=source_query,
-                 fetch_error="Enter a source query first."),
-        )
-
-    fetch_error: str | None = None
-    columns: list[dict] = []
-    try:
-        names = drv.introspect_query_columns(conn, source_query)
-        for i, name in enumerate(names):
-            columns.append({
-                "position": i + 1,
-                "name": name,
-                "default_dest_name": _sanitize_identifier(name),
-                "default_dest_type": "text",
-                "default_description": "",
-            })
-    except (jrunner.JrunnerError, ValueError) as e:
-        fetch_error = str(e)
-    except Exception as e:                                           # noqa: BLE001
-        fetch_error = f"{type(e).__name__}: {e}"
-
-    if fetch_error or not columns:
-        return _templates.TemplateResponse(
-            request, "wizard_sql.html",
-            _ctx(step=2, mode="sql", connection=conn, driver_kind=drv.kind,
-                 source_query=source_query,
-                 fetch_error=fetch_error or "Query returned no columns."),
-        )
-
-    drivers_by_id = {d["id"]: d for d in repo.list_drivers()}
-    dest_conns = [
-        c for c in repo.list_connections()
-        if drivers_by_id.get(c["driver_id"]) is not None
-    ]
-    default_dest_conn_id = conn.get("default_dest_connection_id")
-    default_dest_schema = conn.get("default_dest_schema") or ""
-
-    return _templates.TemplateResponse(
-        request,
-        "wizard_step3.html",
-        _ctx(step=3, mode="sql", sql_mode=True, connection=conn,
-             all_connections=dest_conns, driver_kind=drv.kind,
-             columns=columns, source_query=source_query,
-             qualified_table="(custom query)", table="", qvals={},
-             table_description="", fetch_error=None,
-             default_module_name="",
-             default_dest_conn_id=default_dest_conn_id,
-             default_dest_schema=default_dest_schema,
-             dest_warn=None),
-    )
-
-
 @_router.post("/wizard/create")
 async def wizard_create(request: Request):
     """Step 4 — build source_query from picks, create the module,
     and provision the destination schema + table."""
     form = await request.form()
 
-    entry_mode = form.get("entry_mode", "table")
     source_connection_id = int(form["source_connection_id"])
     dest_connection_id = int(form["dest_connection_id"])
-    table = form.get("table", "")
+    table = form["table"]
     module_name = form["module_name"].strip()
     dest_table = form["dest_table"].strip()
     merge_strategy = form.get("merge_strategy", "full")
@@ -667,74 +576,42 @@ async def wizard_create(request: Request):
     if dest_drv is None:
         raise HTTPException(500, "driver row missing for dest connection")
 
-    if entry_mode == "sql":
+    qvals: dict = {}
-        # SQL-entry path: the source query is used verbatim and the column
+    for f in src_drv.browse_fields():
-        # mapping comes from parallel arrays seeded by query introspection.
+        v = form.get(f.name)
-        source_query = (form.get("source_query") or "").strip()
+        if v:
-        if not source_query:
+            qvals[f.name] = v
-            raise HTTPException(400, "source query is required")
-        col_names = form.getlist("sql_col_name")
-        dest_names = form.getlist("sql_dest_name")
-        dest_types = form.getlist("sql_dest_type")
-        dest_descs = form.getlist("sql_dest_desc")
-        chosen = []
-        for i, src_name in enumerate(col_names):
-            dest_name = (dest_names[i] if i < len(dest_names) else "").strip()
-            dest_type = (dest_types[i] if i < len(dest_types) else "").strip()
-            desc = (dest_descs[i] if i < len(dest_descs) else "").strip() or None
-            if not dest_name or not dest_type:
-                raise HTTPException(
-                    400, f"column {src_name!r} missing dest_name or dest_type")
-            chosen.append({
-                "source_name": src_name,
-                "source_type": "",
-                "dest_name": dest_name,
-                "dest_type": dest_type,
-                "description": desc,
-            })
-        if not chosen:
-            raise HTTPException(400, "no columns to create")
-    else:
-        qvals: dict = {}
-        for f in src_drv.browse_fields():
-            v = form.get(f.name)
-            if v:
-                qvals[f.name] = v
 
-        all_cols = src_drv.get_columns(src_conn, table, **qvals)
+    all_cols = src_drv.get_columns(src_conn, table, **qvals)
-        by_name = {c.name: c for c in all_cols}
+    by_name = {c.name: c for c in all_cols}
-        chosen = []
+    chosen = []
-        for name in picked:
+    for name in picked:
-            if name not in by_name:
+        if name not in by_name:
-                continue
+            continue
-            src_col = by_name[name]
+        src_col = by_name[name]
-            dest_name = (form.get(f"dest_name__{name}") or "").strip()
+        dest_name = (form.get(f"dest_name__{name}") or "").strip()
-            dest_type = (form.get(f"dest_type__{name}") or "").strip()
+        dest_type = (form.get(f"dest_type__{name}") or "").strip()
-            desc = (form.get(f"dest_desc__{name}") or "").strip() or None
+        desc = (form.get(f"dest_desc__{name}") or "").strip() or None
-            if not dest_name or not dest_type:
+        if not dest_name or not dest_type:
-                raise HTTPException(400, f"column {name!r} missing dest_name or dest_type")
+            raise HTTPException(400, f"column {name!r} missing dest_name or dest_type")
-            chosen.append({
+        chosen.append({
-                "source_name": src_col.name,
+            "source_name": src_col.name,
-                "source_type": src_col.type_raw,
+            "source_type": src_col.type_raw,
-                "dest_name": dest_name,
+            "dest_name": dest_name,
-                "dest_type": dest_type,
+            "dest_type": dest_type,
-                "description": desc,
+            "description": desc,
-            })
+        })
-        if not chosen:
+    if not chosen:
-            raise HTTPException(400, "no columns selected")
+        raise HTTPException(400, "no columns selected")
 
-        qualified_source = src_drv.qualified_table_name(table, **qvals)
+    qualified_source = src_drv.qualified_table_name(table, **qvals)
-        # Alias quoting must use the SOURCE dialect — this query runs on the
+    select_list = ",\n  ".join(
-        # source. (The load maps columns by position, so the alias is cosmetic;
+        f"{src_drv.default_expression(c['source_type'], c['source_name'])} AS "
-        # using dest quoting here leaked e.g. SQL Server [brackets] into a
+        f"{dest_drv.quote_identifier(c['dest_name'])}"
-        # Postgres source query.)
+        for c in chosen
-        select_list = ",\n  ".join(
+    )
-            f"{src_drv.default_expression(c['source_type'], c['source_name'])} AS "
+    source_query_override = (form.get("source_query") or "").strip()
-            f"{src_drv.quote_identifier(c['dest_name'])}"
+    source_query = source_query_override or f"SELECT\n  {select_list}\nFROM {qualified_source}"
-            for c in chosen
-        )
-        source_query_override = (form.get("source_query") or "").strip()
-        source_query = source_query_override or f"SELECT\n  {select_list}\nFROM {qualified_source}"
 
     dest_schema, _, dest_table_bare = dest_table.partition(".")
     if not dest_table_bare:
@@ -759,15 +636,11 @@ async def wizard_create(request: Request):
         missing = [c["dest_name"] for c in chosen
                    if c["dest_name"].lower() not in existing_cols]
         if missing:
-            if entry_mode == "sql":
+            back_qs = urlencode(
-                back_url = f"/wizard/sql?source_connection_id={source_connection_id}"
+                [("source_connection_id", source_connection_id),
-            else:
+                 ("table", table),
-                back_qs = urlencode(
+                 ("table_schema", qvals.get("schema") or qvals.get("library") or ""),
-                    [("source_connection_id", source_connection_id),
+                 *qvals.items()])
-                     ("table", table),
-                     ("table_schema", qvals.get("schema") or qvals.get("library") or ""),
-                     *qvals.items()])
-                back_url = f"/wizard/columns?{back_qs}"
             return _templates.TemplateResponse(
                 request,
                 "wizard_error.html",
@@ -776,7 +649,7 @@ async def wizard_create(request: Request):
                     qualified_dest=qualified_dest,
                     missing=missing,
                     existing=sorted(existing_cols),
-                    back_url=back_url,
+                    back_qs=back_qs,
                 ),
                 status_code=409,
             )
@@ -904,32 +777,6 @@ def _build_comment_sql(dest_drv, qualified_dest: str,
     return "\n".join(stmts)
 
 
-def _ensure_column_ids(cols: list[dict]) -> list[dict]:
-    """Assign a stable `id` (c1, c2, …) to any column row missing one.
-    Ids are the data-movement identity for future schema reconciliation."""
-    import re as _re
-    n = 0
-    for c in cols:
-        m = _re.fullmatch(r"c(\d+)", str(c.get("id") or ""))
-        if m:
-            n = max(n, int(m.group(1)))
-    for c in cols:
-        if not c.get("id"):
-            n += 1
-            c["id"] = f"c{n}"
-    return cols
-
-
-def _next_column_id(cols: list[dict]) -> str:
-    import re as _re
-    n = 0
-    for c in cols:
-        m = _re.fullmatch(r"c(\d+)", str(c.get("id") or ""))
-        if m:
-            n = max(n, int(m.group(1)))
-    return f"c{n + 1}"
-
-
 # ---------------------------------------------------------------------------
 # Connections
 # ---------------------------------------------------------------------------
@@ -1104,85 +951,6 @@ async def watermark_create(request: Request, module_id: int):
     return RedirectResponse(url=f"/modules/{module_id}", status_code=303)
 
 
-@_router.get("/modules/{module_id}/columns/new", response_class=HTMLResponse)
-def column_new(request: Request, module_id: int):
-    module = repo.get_module(module_id)
-    if module is None:
-        raise HTTPException(404, f"module id={module_id} not found")
-    return _templates.TemplateResponse(
-        request, "column_form.html",
-        _ctx(module=module, form_action=f"/modules/{module_id}/columns",
-             cancel_url=f"/modules/{module_id}", error=None, values={}),
-    )
-
-
-@_router.post("/modules/{module_id}/columns")
-async def column_create(request: Request, module_id: int):
-    import json as _json
-    module = repo.get_module(module_id)
-    if module is None:
-        raise HTTPException(404, f"module id={module_id} not found")
-    form = await request.form()
-    values = {
-        "dest_name": (form.get("dest_name") or "").strip(),
-        "dest_type": (form.get("dest_type") or "").strip() or "text",
-        "dest_description": (form.get("dest_description") or "").strip(),
-        "source_name": (form.get("source_name") or "").strip(),
-        "source_type": (form.get("source_type") or "").strip(),
-    }
-
-    def _err(msg: str):
-        return _templates.TemplateResponse(
-            request, "column_form.html",
-            _ctx(module=module, form_action=f"/modules/{module_id}/columns",
-                 cancel_url=f"/modules/{module_id}", error=msg, values=values),
-            status_code=400,
-        )
-
-    if not values["dest_name"]:
-        return _err("dest name is required")
-
-    cols = _ensure_column_ids(_json.loads(module["columns_json"]) if module["columns_json"] else [])
-    if any(c["dest_name"].lower() == values["dest_name"].lower() for c in cols):
-        return _err(f"column {values['dest_name']!r} is already in the listing")
-
-    dest_conn = repo.get_connection(module["dest_connection_id"])
-    dest_drv = _driver_for_conn(dest_conn) if dest_conn else None
-    if dest_drv is None:
-        return _err("destination connection has no driver")
-
-    dest_schema, _, dest_table_bare = module["dest_table"].partition(".")
-    if not dest_table_bare:
-        dest_schema, dest_table_bare = "public", dest_schema
-    try:
-        inv = dest_drv.column_inventory(dest_conn, dest_schema, dest_table_bare)
-    except jrunner.JrunnerError as e:
-        return _err(f"could not read {module['dest_table']}: {e}")
-    if any(ci["name"].lower() == values["dest_name"].lower() for ci in inv):
-        return _err(f"column {values['dest_name']!r} already exists on {module['dest_table']}")
-
-    new_col = {
-        "id": _next_column_id(cols),
-        "source_name": values["source_name"],
-        "source_type": values["source_type"],
-        "dest_name": values["dest_name"],
-        "dest_type": values["dest_type"],
-        "description": values["dest_description"] or None,
-    }
-    qualified = dest_drv.qualified_table_name(dest_table_bare, schema=dest_schema)
-    try:
-        jrunner.run_dest_sql(dest_conn, dest_drv.build_add_column_sql(qualified, new_col))
-        if new_col["description"] and dest_drv.supports_table_comments:
-            jrunner.run_dest_sql(
-                dest_conn, _build_comment_sql(dest_drv, qualified, None, [new_col]))
-    except jrunner.JrunnerError as e:
-        return _err(f"ALTER TABLE failed: {e}")
-
-    cols.append(new_col)
-    repo.update_module_columns(module_id, cols)
-    return RedirectResponse(url=f"/modules/{module_id}", status_code=303)
-
-
 @_router.get("/watermarks/{watermark_id}/edit", response_class=HTMLResponse)
 def watermark_edit(request: Request, watermark_id: int):
     wm = repo.get_watermark(watermark_id)

pipekit/web/templates/_wizard_steps.html

@@ -4,7 +4,7 @@
         <span class="num">1</span> source connection
     </div>
     <div class="step {% if step == 2 %}active{% elif step > 2 %}done{% endif %}">
-        <span class="num">2</span> {% if mode == "sql" %}write SQL{% else %}browse tables{% endif %}
+        <span class="num">2</span> browse tables
     </div>
     <div class="step {% if step == 3 %}active{% elif step > 3 %}done{% endif %}">
         <span class="num">3</span> columns &amp; config

pipekit/web/templates/column_form.html

@@ -1,68 +0,0 @@
-{% extends "base.html" %}
-{% set section = "modules" %}
-{% block title %}Add column · {{ module.name }} — Pipekit{% endblock %}
-
-{% block content %}
-<div class="panel">
-    <header>
-        Add column &middot; {{ module.name }}
-        <span class="subtitle">appends to {{ module.dest_table }}</span>
-        <span style="margin-left:auto"><a href="{{ cancel_url }}">&larr; back to module</a></span>
-    </header>
-    <div class="body">
-        {% if error %}
-        <div class="flash err" style="margin-bottom:0.8rem">{{ error }}</div>
-        {% endif %}
-
-        <div class="flash warn" style="margin-bottom:0.8rem">
-            This runs <code>ALTER TABLE … ADD COLUMN</code>, which appends to the
-            <strong>end</strong> of the table (existing rows preserved, new column NULL).
-            The load is positional — add the matching expression as the
-            <strong>last</strong> column of the source query, or the next run will misalign.
-        </div>
-
-        <form method="post" action="{{ form_action }}">
-            <div class="two-col" style="gap:1rem">
-                <label class="field">
-                    <span>dest column name</span>
-                    <input type="text" name="dest_name" required class="mono"
-                           value="{{ values.dest_name or '' }}">
-                    <span class="help">name in the destination table</span>
-                </label>
-                <label class="field">
-                    <span>dest type</span>
-                    <input type="text" name="dest_type" class="mono"
-                           value="{{ values.dest_type or 'text' }}">
-                    <span class="help">e.g. <code>bigint</code>, <code>text</code>, <code>numeric(15,2)</code></span>
-                </label>
-            </div>
-
-            <label class="field">
-                <span>description</span>
-                <textarea name="dest_description" rows="2">{{ values.dest_description or '' }}</textarea>
-                <span class="help">applied as <code>COMMENT ON COLUMN</code> (if the dest supports it)</span>
-            </label>
-
-            <div class="two-col" style="gap:1rem">
-                <label class="field">
-                    <span>source name <span style="color:var(--text-muted)">(optional)</span></span>
-                    <input type="text" name="source_name" class="mono"
-                           value="{{ values.source_name or '' }}">
-                    <span class="help">informational — e.g. <code>RRN(T)</code></span>
-                </label>
-                <label class="field">
-                    <span>source type <span style="color:var(--text-muted)">(optional)</span></span>
-                    <input type="text" name="source_type" class="mono"
-                           value="{{ values.source_type or '' }}">
-                    <span class="help">informational only</span>
-                </label>
-            </div>
-
-            <div class="actions" style="justify-content:flex-end;margin-top:0.8rem">
-                <a class="btn ghost" href="{{ cancel_url }}">cancel</a>
-                <button type="submit" class="primary">add column</button>
-            </div>
-        </form>
-    </div>
-</div>
-{% endblock %}

pipekit/web/templates/module_detail.html

@@ -56,12 +56,10 @@
             <div class="body"><pre class="sql">{{ module.source_query }}</pre></div>
         </div>
 
+        {% if schema_cols or module.dest_description %}
         <div class="panel">
             <header>Schema
                 <span class="subtitle">{{ schema_cols|length }} column{{ 's' if schema_cols|length != 1 else '' }}</span>
-                <span style="margin-left:auto">
-                    <a class="btn" href="/modules/{{ module.id }}/columns/new">+ add column</a>
-                </span>
             </header>
             <div class="body tight">
                 {% if module.dest_description %}
@@ -88,11 +86,10 @@
                     {% endfor %}
                     </tbody>
                 </table>
-                {% else %}
-                <div class="empty">No columns recorded — add one to define the destination schema.</div>
                 {% endif %}
             </div>
         </div>
+        {% endif %}
 
         {% if preview %}
         <div class="panel">

pipekit/web/templates/wizard_error.html

@@ -8,7 +8,7 @@
         Wizard error
         <span class="subtitle">{{ title }}</span>
         <span style="margin-left:auto">
-            <a href="{{ back_url }}">&larr; back to step 3</a>
+            <a href="/wizard/columns?{{ back_qs }}">&larr; back to step 3</a>
         </span>
     </header>
     <div class="body">

pipekit/web/templates/wizard_sql.html

@@ -1,43 +0,0 @@
-{% extends "base.html" %}
-{% set section = "modules" %}
-{% block title %}New module — write SQL{% endblock %}
-
-{% block content %}
-{% include "_wizard_steps.html" %}
-
-<div class="panel">
-    <header>
-        Step 2 — write the source query
-        <span class="subtitle">{{ connection.name }} ({{ driver_kind }})</span>
-        <span style="margin-left:auto"><a href="/wizard">&larr; different connection</a></span>
-    </header>
-    <div class="body">
-        {% if fetch_error %}
-        <div class="flash err" style="margin-bottom:0.8rem">
-            Could not introspect the query's columns:
-            <pre class="sql" style="margin-top:0.4rem">{{ fetch_error }}</pre>
-        </div>
-        {% endif %}
-
-        <form method="post" action="/wizard/sql/columns">
-            <input type="hidden" name="source_connection_id" value="{{ connection.id }}">
-            <label class="field">
-                <span>source query</span>
-                <textarea name="source_query" id="source-query" rows="18" required
-                          class="mono" style="width:100%"
-                          placeholder="SELECT ... FROM ...">{{ source_query }}</textarea>
-                <span class="help">
-                    Runs verbatim against the source at run time. Use
-                    <code>{name}</code> placeholders for watermarks. On
-                    <em>introspect</em>, pipekit runs the query fetching ~no rows to
-                    discover its result columns.
-                </span>
-            </label>
-            <div style="display:flex;justify-content:flex-end;gap:0.5rem;margin-top:0.8rem">
-                <a class="btn ghost" href="/wizard">cancel</a>
-                <button type="submit" class="primary">introspect columns &rarr;</button>
-            </div>
-        </form>
-    </div>
-</div>
-{% endblock %}

pipekit/web/templates/wizard_step1.html

@@ -36,15 +36,10 @@
             {% endfor %}
             </tbody>
         </table>
-        <div class="body" style="display:flex;justify-content:flex-end;gap:0.5rem;align-items:center">
+        <div class="body" style="display:flex;justify-content:flex-end;gap:0.5rem">
             <a class="btn ghost" href="/">cancel</a>
-            <button type="submit" class="btn" formaction="/wizard/sql">write SQL &rarr;</button>
+            <button type="submit" class="primary">next &rarr;</button>
-            <button type="submit" class="primary" formaction="/wizard/tables">browse a table &rarr;</button>
         </div>
-        <p class="help" style="text-align:right;margin:0 0 0.6rem">
-            <strong>browse a table</strong> picks one table and its columns ·
-            <strong>write SQL</strong> starts from an arbitrary query
-        </p>
         </form>
         {% else %}
         <div class="empty">

pipekit/web/templates/wizard_step3.html

@@ -7,15 +7,9 @@
 
 <div class="panel">
     <header>
-        Step 3 — {% if sql_mode %}map columns &amp; configure merge{% else %}choose columns &amp; configure merge{% endif %}
+        Step 3 — choose columns &amp; configure merge
         <span class="subtitle">{{ qualified_table }}</span>
-        <span style="margin-left:auto">
+        <span style="margin-left:auto"><a href="/wizard/tables?source_connection_id={{ connection.id }}{% for k,v in qvals.items() %}&amp;{{ k }}={{ v }}{% endfor %}&amp;browse=1">&larr; different table</a></span>
-            {% if sql_mode %}
-            <a href="/wizard/sql?source_connection_id={{ connection.id }}">&larr; edit SQL</a>
-            {% else %}
-            <a href="/wizard/tables?source_connection_id={{ connection.id }}{% for k,v in qvals.items() %}&amp;{{ k }}={{ v }}{% endfor %}&amp;browse=1">&larr; different table</a>
-            {% endif %}
-        </span>
     </header>
     <div class="body">
         {% if fetch_error %}
@@ -37,75 +31,36 @@
 {% if not fetch_error %}
 <form method="post" action="/wizard/create">
     <input type="hidden" name="source_connection_id" value="{{ connection.id }}">
-    {% if sql_mode %}
-    <input type="hidden" name="entry_mode" value="sql">
-    {% else %}
     <input type="hidden" name="table" value="{{ table }}">
     {% for k, v in qvals.items() %}
         <input type="hidden" name="{{ k }}" value="{{ v }}">
     {% endfor %}
-    {% endif %}
 
     <div class="two-col">
         <div class="panel">
             <header>
                 Columns
-                {% if sql_mode %}
-                <span class="subtitle">{{ columns|length }} from the query — all created; types default to <code>text</code></span>
-                {% else %}
                 <span class="subtitle">{{ columns|length }} total — uncheck to exclude</span>
                 <span style="margin-left:auto">
                     <button type="button" class="ghost" onclick="toggleAll(true)">all</button>
                     <button type="button" class="ghost" onclick="toggleAll(false)">none</button>
                 </span>
-                {% endif %}
             </header>
             <div class="body tight">
                 <table class="grid picker">
                     <thead>
                         <tr>
-                            {% if not sql_mode %}<th class="pick"></th>{% endif %}
+                            <th class="pick"></th>
                             <th style="width:3em">#</th>
                             <th>source name</th>
-                            {% if not sql_mode %}
                             <th>source type</th>
                             <th style="width:3em">null?</th>
-                            {% endif %}
                             <th>dest name</th>
                             <th>dest type</th>
                             <th>description</th>
                         </tr>
                     </thead>
                     <tbody>
-                    {% if sql_mode %}
-                    {% for c in columns %}
-                        <tr>
-                            <td class="mono">{{ c.position }}</td>
-                            <td class="mono">
-                                {{ c.name }}
-                                <input type="hidden" name="sql_col_name" value="{{ c.name }}">
-                            </td>
-                            <td>
-                                <input type="text" class="mono"
-                                       name="sql_dest_name"
-                                       value="{{ c.default_dest_name }}"
-                                       style="width:100%;font-size:12px">
-                            </td>
-                            <td>
-                                <input type="text" class="mono"
-                                       name="sql_dest_type"
-                                       value="{{ c.default_dest_type }}"
-                                       style="width:100%;font-size:12px">
-                            </td>
-                            <td>
-                                <input type="text"
-                                       name="sql_dest_desc"
-                                       value="{{ c.default_description }}"
-                                       style="width:100%;font-size:12px">
-                            </td>
-                        </tr>
-                    {% endfor %}
-                    {% else %}
                     {% for c in columns %}
                         <tr onclick="var cb=document.getElementById('col-{{ loop.index }}'); if(event.target.tagName!=='INPUT') cb.checked=!cb.checked; fillQuery()">
                             <td class="pick">
@@ -137,7 +92,6 @@
                             </td>
                         </tr>
                     {% endfor %}
-                    {% endif %}
                     </tbody>
                 </table>
             </div>
@@ -238,24 +192,16 @@
                 </div>
             </div>
 
-            <div class="panel" id="query-panel" style="{% if not sql_mode %}display:none{% endif %}">
+            <div class="panel" id="query-panel" style="display:none">
                 <header>Source query
-                    {% if sql_mode %}
-                    <span class="subtitle">runs verbatim at run time</span>
-                    {% else %}
                     <span class="subtitle">auto-generated from picks — edit to add WHERE clause</span>
-                    {% endif %}
                 </header>
                 <div class="body">
                     <textarea name="source_query" id="source-query" rows="10" class="mono"
                               style="width:100%" oninput="checkPlaceholders()"
-                              {% if not sql_mode %}placeholder="Leave blank to auto-generate from column picks"{% endif %}>{% if sql_mode %}{{ source_query }}{% endif %}</textarea>
+                              placeholder="Leave blank to auto-generate from column picks"></textarea>
                     <div id="placeholder-warning" class="flash warn" style="display:none;margin-top:0.4rem"></div>
-                    {% if sql_mode %}
-                    <span class="help">Edit if needed. <code>{name}</code> placeholders resolve from watermarks at run time.</span>
-                    {% else %}
                     <span class="help">Leave blank to auto-generate. Add <code>WHERE col &gt; {name}</code> for incremental filtering.</span>
-                    {% endif %}
                 </div>
             </div>
 
@@ -295,7 +241,6 @@
 
 <script>
     const QUALIFIED_TABLE = {{ qualified_table | tojson }};
-    const SQL_MODE = {{ 'true' if sql_mode else 'false' }};
 
     function toggleAll(val) {
         document.querySelectorAll('.col-check').forEach(function (cb) { cb.checked = val; });
@@ -305,15 +250,12 @@
     function onStrategyChange(val) {
         document.getElementById('mkf').style.display = val === 'incremental' ? '' : 'none';
         document.getElementById('wm-panel').style.display = val === 'incremental' ? '' : 'none';
-        // In SQL mode the query panel is always shown (it holds the user's query).
+        document.getElementById('query-panel').style.display = val === 'incremental' ? '' : 'none';
-        document.getElementById('query-panel').style.display =
+        if (val === 'incremental') fillQuery();
-            (SQL_MODE || val === 'incremental') ? '' : 'none';
-        if (!SQL_MODE && val === 'incremental') fillQuery();
         checkPlaceholders();
     }
 
     function fillQuery() {
-        if (SQL_MODE) return;   // never overwrite a hand-written query
         const ta = document.getElementById('source-query');
         if (!ta) return;
         const checked = [...document.querySelectorAll('.col-check:checked')];