web: quote wizard source-query aliases with the source dialect

wizard_create built the generated source query's column aliases with the
DEST driver's quote_identifier, but that query runs on the SOURCE. A pg->SQL
Server module emitted "AS [col]" (SQL Server brackets) into a Postgres query,
which failed with: syntax error at or near "[". The load maps columns by
position, so the alias is cosmetic — quote it with the source dialect.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

CLAUDE.md

@@ -100,6 +100,10 @@ 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,14 +7,100 @@ 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
+from fastapi import FastAPI, Request
+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):
@@ -24,7 +110,28 @@ 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,6 +167,30 @@ 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
@@ -181,6 +205,24 @@ 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,6 +115,13 @@ 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,10 +254,13 @@ _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 PG then throws "No results were returned by the
+# still executes, but the driver then complains there's no ResultSet. The
-# query." The statement succeeded — ignore the trace.
+# statement succeeded — ignore the trace. Each dialect words it differently:
+#   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,6 +232,14 @@ 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,15 +541,106 @@ 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["table"]
+    table = form.get("table", "")
     module_name = form["module_name"].strip()
     dest_table = form["dest_table"].strip()
     merge_strategy = form.get("merge_strategy", "full")
@@ -576,6 +667,34 @@ async def wizard_create(request: Request):
     if dest_drv is None:
         raise HTTPException(500, "driver row missing for dest connection")
 
+    if entry_mode == "sql":
+        # SQL-entry path: the source query is used verbatim and the column
+        # mapping comes from parallel arrays seeded by query introspection.
+        source_query = (form.get("source_query") or "").strip()
+        if not source_query:
+            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)
@@ -605,9 +724,13 @@ async def wizard_create(request: Request):
             raise HTTPException(400, "no columns selected")
 
         qualified_source = src_drv.qualified_table_name(table, **qvals)
+        # Alias quoting must use the SOURCE dialect — this query runs on the
+        # source. (The load maps columns by position, so the alias is cosmetic;
+        # using dest quoting here leaked e.g. SQL Server [brackets] into a
+        # Postgres source query.)
         select_list = ",\n  ".join(
             f"{src_drv.default_expression(c['source_type'], c['source_name'])} AS "
-        f"{dest_drv.quote_identifier(c['dest_name'])}"
+            f"{src_drv.quote_identifier(c['dest_name'])}"
             for c in chosen
         )
         source_query_override = (form.get("source_query") or "").strip()
@@ -636,11 +759,15 @@ 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_url = f"/wizard/sql?source_connection_id={source_connection_id}"
+            else:
                 back_qs = urlencode(
                     [("source_connection_id", source_connection_id),
                      ("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",
@@ -649,7 +776,7 @@ async def wizard_create(request: Request):
                     qualified_dest=qualified_dest,
                     missing=missing,
                     existing=sorted(existing_cols),
-                    back_qs=back_qs,
+                    back_url=back_url,
                 ),
                 status_code=409,
             )
@@ -777,6 +904,32 @@ 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
 # ---------------------------------------------------------------------------
@@ -951,6 +1104,85 @@ 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> browse tables
+        <span class="num">2</span> {% if mode == "sql" %}write SQL{% else %}browse tables{% endif %}
     </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

@@ -0,0 +1,68 @@
+{% 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,10 +56,12 @@
             <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 %}
@@ -86,10 +88,11 @@
                     {% 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="/wizard/columns?{{ back_qs }}">&larr; back to step 3</a>
+            <a href="{{ back_url }}">&larr; back to step 3</a>
         </span>
     </header>
     <div class="body">

pipekit/web/templates/wizard_sql.html

@@ -0,0 +1,43 @@
+{% 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,10 +36,15 @@
             {% endfor %}
             </tbody>
         </table>
-        <div class="body" style="display:flex;justify-content:flex-end;gap:0.5rem">
+        <div class="body" style="display:flex;justify-content:flex-end;gap:0.5rem;align-items:center">
             <a class="btn ghost" href="/">cancel</a>
-            <button type="submit" class="primary">next &rarr;</button>
+            <button type="submit" class="btn" formaction="/wizard/sql">write SQL &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,9 +7,15 @@
 
 <div class="panel">
     <header>
-        Step 3 — choose columns &amp; configure merge
+        Step 3 — {% if sql_mode %}map columns &amp; configure merge{% else %}choose columns &amp; configure merge{% endif %}
         <span class="subtitle">{{ qualified_table }}</span>
-        <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>
+        <span style="margin-left:auto">
+            {% 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 %}
@@ -31,36 +37,75 @@
 {% 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>
-                            <th class="pick"></th>
+                            {% if not sql_mode %}<th class="pick"></th>{% endif %}
                             <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">
@@ -92,6 +137,7 @@
                             </td>
                         </tr>
                     {% endfor %}
+                    {% endif %}
                     </tbody>
                 </table>
             </div>
@@ -192,16 +238,24 @@
                 </div>
             </div>
 
-            <div class="panel" id="query-panel" style="display:none">
+            <div class="panel" id="query-panel" style="{% if not sql_mode %}display:none{% endif %}">
                 <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()"
-                              placeholder="Leave blank to auto-generate from column picks"></textarea>
+                              {% if not sql_mode %}placeholder="Leave blank to auto-generate from column picks"{% endif %}>{% if sql_mode %}{{ source_query }}{% endif %}</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>
 
@@ -241,6 +295,7 @@
 
 <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; });
@@ -250,12 +305,15 @@
     function onStrategyChange(val) {
         document.getElementById('mkf').style.display = val === 'incremental' ? '' : 'none';
         document.getElementById('wm-panel').style.display = val === 'incremental' ? '' : 'none';
-        document.getElementById('query-panel').style.display = val === 'incremental' ? '' : 'none';
+        // In SQL mode the query panel is always shown (it holds the user's query).
-        if (val === 'incremental') fillQuery();
+        document.getElementById('query-panel').style.display =
+            (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')];