diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 49fbb98..c537df9 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -1,5 +1,9 @@
 name: CI
 
+# Least-privilege token for every job; nothing here writes to the repo.
+permissions:
+  contents: read
+
 on:
   push:
     branches: [main]
@@ -118,7 +122,7 @@ jobs:
 
       - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
         with:
-          node-version: "20"
+          node-version: "22"
           cache: npm
           cache-dependency-path: frontend/package-lock.json
 
@@ -144,7 +148,7 @@ jobs:
 
       - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
         with:
-          node-version: "20"
+          node-version: "22"
           cache: npm
           cache-dependency-path: frontend/package-lock.json
 
@@ -167,7 +171,7 @@ jobs:
 
       - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
         with:
-          node-version: "20"
+          node-version: "22"
           cache: npm
           cache-dependency-path: frontend/package-lock.json
 
@@ -197,7 +201,7 @@ jobs:
 
       - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4.4.0
         with:
-          node-version: "20"
+          node-version: "22"
           cache: npm
           cache-dependency-path: frontend/package-lock.json
 
diff --git a/backend/app/api/v1/routes/column_mappings.py b/backend/app/api/v1/routes/column_mappings.py
index 3559ba1..cc3d70b 100644
--- a/backend/app/api/v1/routes/column_mappings.py
+++ b/backend/app/api/v1/routes/column_mappings.py
@@ -16,7 +16,7 @@
 from fastapi import APIRouter, Depends, HTTPException, Query
 from sqlalchemy.orm import Session
 
-from app.core.url_guard import SSRFBlockedError, validate_url
+from app.core.url_guard import SSRFBlockedError, validate_url_async
 from app.db.models.column_mapping import ColumnMapping
 from app.db.models.task import Task
 from app.db.schemas.column_mapping import (
@@ -390,6 +390,12 @@ async def preview_fields(
     except HTTPException:
         # Re-raise deliberate HTTP errors instead of rewriting them to 400.
         raise
+    except SSRFBlockedError as e:
+        # Raised by the fetch-time re-validation inside fetch_json. It
+        # subclasses ValueError, so it must be handled before that branch or
+        # it surfaces as a misleading 400 "Invalid JSON".
+        logger.warning(f"SSRF guard rejected preview fetch: {e}")
+        raise HTTPException(status_code=403, detail=str(e))
     except ValueError as e:
         logger.warning(f"Invalid JSON for task {task_id}: {str(e)}")
         raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(e)}")
@@ -506,7 +512,8 @@ async def preview_fields_standalone(request: PreviewFieldsRequest):
 
             # SSRF guard (C4): reject private/loopback/link-local targets.
             try:
-                validate_url(request.url)
+                # Async variant: DNS resolution must not block the event loop.
+                await validate_url_async(request.url)
             except SSRFBlockedError as e:
                 logger.warning(f"SSRF guard rejected standalone preview URL: {e}")
                 raise HTTPException(status_code=403, detail=str(e))
@@ -592,6 +599,12 @@ async def preview_fields_standalone(request: PreviewFieldsRequest):
         # Re-raise deliberate HTTP errors (400 missing sample_json, 403 SSRF
         # block) instead of letting the broad handler rewrite them to 400.
         raise
+    except SSRFBlockedError as e:
+        # Raised by the fetch-time re-validation inside fetch_json. It
+        # subclasses ValueError, so it must be handled before that branch or
+        # it surfaces as a misleading 400 "Invalid JSON".
+        logger.warning(f"SSRF guard rejected preview fetch: {e}")
+        raise HTTPException(status_code=403, detail=str(e))
     except ValueError as e:
         logger.warning(f"Invalid JSON: {str(e)}")
         raise HTTPException(status_code=400, detail=f"Invalid JSON: {str(e)}")
diff --git a/backend/app/api/v1/routes/tasks.py b/backend/app/api/v1/routes/tasks.py
index c4b032f..5aac852 100644
--- a/backend/app/api/v1/routes/tasks.py
+++ b/backend/app/api/v1/routes/tasks.py
@@ -197,6 +197,20 @@ def update_task(task_id: int, payload: TaskUpdate, db: Session = Depends(get_db)
     if "connection_id" in update_data:
         _require_existing_connection(update_data["connection_id"])
 
+    # Upsert invariant on the EFFECTIVE state (request merged over stored
+    # values): {"upsert_enabled": true} alone must not leave a task with no
+    # keys, or the runner silently degrades to plain inserts. Only enforced
+    # when the request touches the upsert fields, so an unrelated partial
+    # update of a legacy row is not blocked.
+    if "upsert_enabled" in update_data or "upsert_keys" in update_data:
+        effective_upsert = update_data.get("upsert_enabled", task.upsert_enabled)
+        effective_keys = update_data.get("upsert_keys", task.upsert_keys)
+        if effective_upsert and not effective_keys:
+            raise HTTPException(
+                status_code=400,
+                detail="upsert_enabled requires at least one column in upsert_keys",
+            )
+
     # Encrypt secrets. Empty string = explicit clear; None (explicit null)
     # also clears; omitted = untouched (key not in update_data).
     if "api_key" in update_data:
@@ -573,7 +587,11 @@ def get_task_stats(task_id: int, db: Session = Depends(get_db)):
     else:
         pairs = (
             db.query(TaskRun.started_at, TaskRun.ended_at)
-            .filter(TaskRun.task_id == task_id, TaskRun.ended_at.isnot(None))
+            .filter(
+                TaskRun.task_id == task_id,
+                TaskRun.started_at.isnot(None),
+                TaskRun.ended_at.isnot(None),
+            )
             .all()
         )
         durations = [(p.ended_at - p.started_at).total_seconds() for p in pairs]
diff --git a/backend/app/core/config.py b/backend/app/core/config.py
index bd96af8..d2eea2f 100644
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -1,4 +1,4 @@
-from pydantic import field_validator
+from pydantic import Field, field_validator
 from pydantic_settings import BaseSettings
 
 
@@ -45,7 +45,9 @@ class Settings(BaseSettings):
     ALLOWED_SOURCE_HOSTS: str = ""
 
     # Auto-pause a schedule after this many consecutive dispatch failures.
-    SCHEDULE_MAX_CONSECUTIVE_FAILURES: int = 5
+    # Auto-pause a schedule after this many consecutive dispatch failures.
+    # 0 or negative would pause on the very first failure, so require >= 1.
+    SCHEDULE_MAX_CONSECUTIVE_FAILURES: int = Field(default=5, ge=1)
 
     @property
     def destination_sqlalchemy_url(self) -> str:
diff --git a/backend/app/core/url_guard.py b/backend/app/core/url_guard.py
index 0c40961..1d09c3c 100644
--- a/backend/app/core/url_guard.py
+++ b/backend/app/core/url_guard.py
@@ -61,6 +61,15 @@ def validate_url(url: str, *, resolve: bool = True) -> str:
     if not host:
         raise SSRFBlockedError("URL must include a host")
 
+    # urlparse raises ValueError on attribute access for out-of-range ports
+    # (http://example.com:99999/). Check it up front so literal-IP URLs and
+    # config-time validation (resolve=False) reject it too, as an
+    # SSRFBlockedError rather than a 500.
+    try:
+        port = parsed.port
+    except ValueError as e:
+        raise SSRFBlockedError(f"URL has an invalid port: {e}") from e
+
     # Literal IPs are checked directly; hostnames are resolved.
     try:
         candidate_addrs = [ipaddress.ip_address(host)]
@@ -71,13 +80,6 @@ def validate_url(url: str, *, resolve: bool = True) -> str:
     host_allowed = host.lower() in allowed
 
     if not candidate_addrs and resolve:
-        # urlparse raises ValueError on attribute access for out-of-range
-        # ports (http://example.com:99999/) — surface it as SSRFBlockedError,
-        # not a 500.
-        try:
-            port = parsed.port
-        except ValueError as e:
-            raise SSRFBlockedError(f"URL has an invalid port: {e}") from e
         try:
             infos = socket.getaddrinfo(host, port or None)
             candidate_addrs = [ipaddress.ip_address(info[4][0]) for info in infos]
diff --git a/backend/app/db/schemas/task.py b/backend/app/db/schemas/task.py
index 73ef7bd..dc06c08 100644
--- a/backend/app/db/schemas/task.py
+++ b/backend/app/db/schemas/task.py
@@ -1,6 +1,7 @@
 import re
 from datetime import datetime
 from typing import Any, Literal
+from urllib.parse import urlparse
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
@@ -45,6 +46,10 @@ def _validate_source_url(url: str | None) -> str | None:
     """
     if url is None or "://" not in url:
         return url
+    if not urlparse(url.strip()).scheme:
+        # Relative endpoint path whose query value happens to contain "://"
+        # (e.g. /api/items?redirect=https://cb.example.com); resolved elsewhere.
+        return url
     try:
         return validate_url(url, resolve=False)
     except SSRFBlockedError as e:
@@ -246,16 +251,25 @@ def validate_dest_table_identifier(cls, v: str):
 
     @field_validator("upsert_keys")
     @classmethod
-    def validate_upsert_key_identifiers(cls, v: list[str] | None, info):
-        """Normalize entries to safe SQL identifiers and require at least one
-        key when upsert is enabled."""
+    def validate_upsert_key_identifiers(cls, v: list[str] | None):
+        """Normalize entries to safe SQL identifiers."""
         if v:
             v = [_validate_sql_identifier(k, "upsert_keys entry") for k in v]
-        upsert_enabled = info.data.get("upsert_enabled")
-        if upsert_enabled and (not v or len(v) == 0):
-            raise ValueError("upsert_enabled requires at least one column in upsert_keys")
         return v
 
+    @model_validator(mode="after")
+    def validate_upsert_requires_keys(self):
+        """upsert_enabled needs at least one key.
+
+        A model validator (not a field validator on upsert_keys) so the rule
+        also runs when upsert_keys is omitted entirely — otherwise
+        {"upsert_enabled": true} alone was accepted and the runner silently
+        fell back to plain inserts.
+        """
+        if self.upsert_enabled and not self.upsert_keys:
+            raise ValueError("upsert_enabled requires at least one column in upsert_keys")
+        return self
+
     @field_validator("skip_column")
     @classmethod
     def validate_skip_column_identifier(cls, v: str | None):
@@ -327,14 +341,17 @@ def validate_skip_column_identifier_update(cls, v: str | None):
             _validate_sql_identifier(v, "skip_column")
         return v
 
-    @field_validator("upsert_keys")
-    @classmethod
-    def validate_upsert_keys(cls, v: list[str] | None, info):
-        """When enabling upsert, keys must be present in the same request."""
-        upsert_enabled = info.data.get("upsert_enabled")
-        if upsert_enabled and not v:
+    @model_validator(mode="after")
+    def validate_upsert_keys_when_supplied(self):
+        """Enabling upsert while explicitly sending an empty key list is invalid.
+
+        Runs as a model validator so it also fires when upsert_keys is set to
+        null/[] together with upsert_enabled. When upsert_keys is omitted the
+        stored keys apply; update_task() validates that effective state.
+        """
+        if self.upsert_enabled and "upsert_keys" in self.model_fields_set and not self.upsert_keys:
             raise ValueError("upsert_enabled=true requires at least one column in upsert_keys")
-        return v
+        return self
 
     @model_validator(mode="after")
     def reject_explicit_nulls_for_required_fields(self):
diff --git a/backend/app/services/api_connector.py b/backend/app/services/api_connector.py
index 8b3f68c..9f97645 100644
--- a/backend/app/services/api_connector.py
+++ b/backend/app/services/api_connector.py
@@ -2,7 +2,9 @@
 import base64
 import datetime as _dt
 import email.utils
+import re
 from typing import Any
+from urllib.parse import urlsplit, urlunsplit
 
 import httpx
 from loguru import logger
@@ -49,16 +51,25 @@ def mask_headers(headers: dict) -> dict:
     return masked
 
 
+_LOG_CONTROL_CHARS_RE = re.compile(r"[\x00-\x1f\x7f]")
+
+
 def _redact_url_for_log(url: str) -> str:
-    """Strip query strings AND userinfo (may embed api keys / tokens) for log
-    output. All URL logging must go through this helper."""
-    base = url.split("?", 1)[0]
-    scheme, sep, rest = base.partition("://")
-    if sep and "@" in rest.split("/", 1)[0]:
-        # https://user:secret@host/path -> https://host/path
-        _, _, tail = rest.partition("@")
-        return f"{scheme}://{tail}"
-    return base
+    """Reduce a URL to scheme://host[:port]/path for log output.
+
+    Drops the query string and fragment (may embed api keys / tokens) and any
+    userinfo (user:secret@), then strips control characters so a crafted URL
+    cannot forge log lines. All URL logging must go through this helper.
+    """
+    try:
+        parts = urlsplit(str(url))
+    except ValueError:
+        return "<unparseable-url>"
+    netloc = parts.netloc.rpartition("@")[2]  # drop user:secret@
+    # Drop ;param=value path parameters segment by segment (e.g. ;jsessionid=...).
+    path = "/".join(segment.split(";", 1)[0] for segment in parts.path.split("/"))
+    redacted = urlunsplit((parts.scheme, netloc, path, "", ""))
+    return _LOG_CONTROL_CHARS_RE.sub("", redacted)
 
 
 def _parse_retry_after(header_value: str | None) -> float | None:
@@ -564,6 +575,10 @@ async def fetch_sample_response(
         logger.info(f"Fetched sample API response: {type(response_data).__name__}")
         return response_data
 
+    except SSRFBlockedError:
+        # Preserve the SSRF signal: callers map it to 403. The catch-all below
+        # would flatten it into a generic ValueError (a misleading 400).
+        raise
     except httpx.HTTPStatusError as e:
         # Status + reason only: upstream bodies can contain PII, so neither
         # the log line nor the client-facing message includes an excerpt.
diff --git a/backend/app/services/runner.py b/backend/app/services/runner.py
index f09b576..4f85211 100644
--- a/backend/app/services/runner.py
+++ b/backend/app/services/runner.py
@@ -24,6 +24,10 @@
 _SAFE_SQL_IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_$#]{0,127}$")
 
 
+# Row-error staging is flushed to the DB every N entries (see run_import step 8).
+ROW_ERROR_FLUSH_SIZE = 1000
+
+
 def _redact_cursor(value):
     """Same redaction policy as workers.tasks._redact_cursor — see that
     function for rationale. Duplicated here to avoid a circular import
@@ -268,13 +272,16 @@ async def run_import(
         # Debug: Log source field names and sample values
         if flattened_records:
             sample_record = flattened_records[0]
+            # Names and types only: upstream rows can carry PII and DEBUG logs
+            # may be retained or shipped outside the database boundary.
             logger.debug(f"Sample flattened record keys: {list(sample_record.keys())}")
-            logger.debug(f"Sample flattened record: {sample_record}")
             for mapping in column_mappings:
                 source_val = sample_record.get(mapping.source_field)
                 logger.debug(
                     f"Mapping: {mapping.source_field} -> {mapping.dest_column} "
-                    f"| transforms={mapping.transform_rules} | source_value={repr(source_val)}"
+                    f"| transforms={mapping.transform_rules} "
+                    f"| source_present={source_val is not None} "
+                    f"| source_type={type(source_val).__name__}"
                 )
 
         mapped_records = mapper.map_rows(flattened_records, column_mappings)
@@ -282,7 +289,7 @@ async def run_import(
         # Debug: Log mapped results
         if mapped_records:
             sample_mapped = mapped_records[0]
-            logger.debug(f"Sample mapped record: {sample_mapped}")
+            logger.debug(f"Sample mapped record columns: {list(sample_mapped.keys())}")
 
         # Step 7: Validate rows
         log_step(db, task_run_id, "VALIDATE", f"Validating {len(mapped_records)} records")
@@ -294,7 +301,10 @@ async def run_import(
 
         logger.info(f"Validation complete: {len(valid_rows)} valid, {len(invalid_rows)} invalid")
 
-        # Step 8: Log validation errors
+        # Step 8: Log validation errors. log_row_error() only stages rows; flush
+        # in chunks so a run with thousands of invalid rows doesn't hold every
+        # TaskRunLog in the session until the single commit below.
+        staged_errors = 0
         for idx, invalid_item in enumerate(invalid_rows):
             errors = invalid_item["errors"]
             for error in errors:
@@ -307,6 +317,9 @@ async def run_import(
                     error_message=error.message,
                     source_value=str(error.value) if error.value is not None else None,
                 )
+                staged_errors += 1
+                if staged_errors % ROW_ERROR_FLUSH_SIZE == 0:
+                    db.flush()
 
         task_run.error_count = len(invalid_rows)
         db.commit()
@@ -535,7 +548,7 @@ def _build_insert_statement(
 
     # Debug: Log sample row
     if sample_row:
-        logger.debug(f"Sample row for date detection: {sample_row}")
+        logger.debug(f"Sample row columns for date detection: {list(sample_row.keys())}")
 
     # Build placeholders, wrapping date strings with TO_DATE()
     placeholders = []
@@ -548,7 +561,7 @@ def _build_insert_statement(
             value = sample_row[column]
             if isinstance(value, str) and re.match(r"^\d{4}-\d{2}-\d{2}$", value):
                 is_date_string = True
-                logger.debug(f"Detected date column: {column} = {value}")
+                logger.debug(f"Detected date column: {column}")
 
         if is_date_string:
             # Wrap with TO_DATE() for Oracle DATE columns
@@ -751,10 +764,17 @@ def _process_upsert_batch(
     # clauses silently dropped by SQL) or both to_insert (constraint failure
     # or duplicate rows). First occurrence wins; duplicates are counted as
     # skipped with an attribution entry.
+    # Rows are carried as (absolute_row_index, row) from here on so error
+    # attribution keeps pointing at the caller's row numbers after duplicates
+    # are dropped. Rows missing any key value cannot collide on a key: they
+    # skip deduplication and take the normal route (no match -> insert).
     seen_keys = set()
-    deduped_batch = []
+    indexed_batch: list[tuple[int, dict]] = []
     for idx, row in enumerate(batch):
         key_tuple = tuple(row.get(key) for key in upsert_keys)
+        if any(value is None for value in key_tuple):
+            indexed_batch.append((batch_offset + idx, row))
+            continue
         if key_tuple in seen_keys:
             results["skipped"] += 1
             results["error_details"].append(
@@ -766,8 +786,8 @@ def _process_upsert_batch(
             )
             continue
         seen_keys.add(key_tuple)
-        deduped_batch.append(row)
-    batch = deduped_batch
+        indexed_batch.append((batch_offset + idx, row))
+    batch = [row for _, row in indexed_batch]
 
     if not batch:
         return results
@@ -825,20 +845,20 @@ def _process_upsert_batch(
         to_update = []
         to_skip = []
 
-        for idx, row in enumerate(batch):
+        for abs_idx, row in indexed_batch:
             row_key_tuple = tuple(row.get(key) for key in upsert_keys)
             existing = existing_rows.get(row_key_tuple)
 
             if existing is None:
                 # New record
-                to_insert.append((batch_offset + idx, row))
+                to_insert.append((abs_idx, row))
             elif task.skip_column and task.skip_value is not None and _should_skip(task, existing):
                 # Record exists and matches the configured skip condition —
                 # third parties may have marked this row processed; never
                 # overwrite it.
-                to_skip.append((batch_offset + idx, row))
+                to_skip.append((abs_idx, row))
             else:
-                to_update.append((batch_offset + idx, row))
+                to_update.append((abs_idx, row))
 
         # Steps 3+4: Bulk UPDATE + INSERT inside a savepoint, so a partially
         # applied bulk statement is rolled back completely before falling back
diff --git a/backend/app/services/scheduler.py b/backend/app/services/scheduler.py
index 0a631f7..5949e66 100644
--- a/backend/app/services/scheduler.py
+++ b/backend/app/services/scheduler.py
@@ -176,20 +176,33 @@ def _handle_dispatch_failure(self, task_id: int, exc: Exception):
                 current = task_schedule.consecutive_failures or 0
                 task_schedule.consecutive_failures = current + 1
 
-                if current + 1 >= threshold and task_schedule.is_active:
+                paused = current + 1 >= threshold and task_schedule.is_active
+                if paused:
                     task_schedule.is_active = False
-                    self.remove_schedule(task_id)
-                    logger.error(
-                        f"Auto-paused schedule for task {task_id}: "
-                        f"{current + 1} consecutive dispatch failures "
-                        f"(threshold={threshold}). Use the resume endpoint to reactivate."
-                    )
 
+                # Persist FIRST. If the commit fails, the rollback below leaves
+                # the schedule active in the DB while its in-memory job would
+                # already be gone — dispatch would silently stop until a restart.
                 self.db.commit()
                 logger.error(
                     f"Scheduled task {task_id} dispatch failure "
                     f"(consecutive_failures={current + 1})"
                 )
+
+                if paused:
+                    try:
+                        self.remove_schedule(task_id)
+                    except Exception as remove_exc:
+                        # DB says paused; the job will not be reloaded on restart.
+                        logger.error(
+                            f"Auto-paused task {task_id} in the database but failed to "
+                            f"remove its scheduler job: {remove_exc}"
+                        )
+                    logger.error(
+                        f"Auto-paused schedule for task {task_id}: "
+                        f"{current + 1} consecutive dispatch failures "
+                        f"(threshold={threshold}). Use the resume endpoint to reactivate."
+                    )
         except Exception as counter_exc:
             logger.error(
                 f"Failed to persist consecutive_failures for task {task_id}: {counter_exc}"
diff --git a/backend/requirements.txt b/backend/requirements.txt
index 8872b0e..4fb4157 100644
--- a/backend/requirements.txt
+++ b/backend/requirements.txt
@@ -7,6 +7,8 @@ SQLAlchemy==2.0.36
 alembic==1.13.3
 oracledb==3.4.2
 httpx==0.27.2
+# httpx leaves idna unconstrained; <3.15 is vulnerable (PYSEC-2026-215 / GHSA-65pc-fj4g-8rjx)
+idna>=3.15
 jsonpath-ng==1.6.1
 celery==5.4.0
 redis==5.0.8
diff --git a/backend/tests/integration/test_api_endpoints.py b/backend/tests/integration/test_api_endpoints.py
index 91d9cf4..f9dd0d4 100644
--- a/backend/tests/integration/test_api_endpoints.py
+++ b/backend/tests/integration/test_api_endpoints.py
@@ -451,3 +451,39 @@ def test_root_endpoint(client: TestClient):
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
+
+
+# ============================================================================
+# Standalone field preview: fetch-time SSRF block must surface as 403
+# ============================================================================
+
+
+def test_standalone_preview_fetch_time_ssrf_block_is_403(client: TestClient):
+    """SSRFBlockedError subclasses ValueError; before the dedicated handler it
+    came back as 400 "Invalid JSON", hiding that the URL was blocked.
+
+    The block is raised from fetch_json, i.e. underneath fetch_sample_response,
+    so it must survive that helper's catch-all wrapper to reach the route.
+    """
+    from unittest.mock import AsyncMock, patch
+
+    from app.core.url_guard import SSRFBlockedError
+
+    path = "/api/v1/preview-fields-standalone"
+    with (
+        # The pre-fetch check must pass so the fetch-time path is what is exercised.
+        patch(
+            "app.api.v1.routes.column_mappings.validate_url_async",
+            new=AsyncMock(return_value="https://api.example.com/x"),
+        ),
+        patch(
+            "app.services.api_connector.fetch_json",
+            side_effect=SSRFBlockedError("Host 'api.example.com' resolves to a non-public address"),
+        ),
+    ):
+        resp = client.post(
+            path,
+            json={"use_auto_fetch": True, "method": "GET", "url": "https://api.example.com/x"},
+        )
+    assert resp.status_code == 403
+    assert "non-public" in resp.json()["detail"]
diff --git a/backend/tests/unit/test_config_env.py b/backend/tests/unit/test_config_env.py
index 45c6589..6909153 100644
--- a/backend/tests/unit/test_config_env.py
+++ b/backend/tests/unit/test_config_env.py
@@ -30,7 +30,12 @@ def test_missing_app_env_defaults_to_dev(self):
         from pathlib import Path
 
         backend = Path(__file__).resolve().parents[2]
-        code = "from app.core.config import Settings;print(Settings(SECRET_KEY='x').APP_ENV)"
+        # _env_file=None: a developer's backend/.env must not be able to supply APP_ENV
+        # and mask the built-in default this test exists to exercise.
+        code = (
+            "from app.core.config import Settings;"
+            "print(Settings(_env_file=None, SECRET_KEY='x').APP_ENV)"
+        )
         out = subprocess.run(
             [sys.executable, "-c", code],
             env=env,
@@ -61,6 +66,10 @@ def _make_app(self, app_env):
         finally:
             settings.APP_ENV = original
             settings.API_TOKEN = original_token
+            # Reload again so app.main.app reflects the restored settings for
+            # any later test that imports it; the caller already holds the
+            # instance built above.
+            importlib.reload(main_module)
 
     def test_docs_enabled_outside_production(self):
         from fastapi.testclient import TestClient
@@ -77,3 +86,16 @@ def test_docs_disabled_in_production_canonical_value(self):
         client = TestClient(app)
         assert client.get("/docs").status_code == 404
         assert client.get("/openapi.json").status_code == 404
+
+
+class TestScheduleFailureThreshold:
+    def test_default_is_five(self, monkeypatch):
+        monkeypatch.delenv("SCHEDULE_MAX_CONSECUTIVE_FAILURES", raising=False)
+        settings = Settings(_env_file=None, SECRET_KEY="x")
+        assert settings.SCHEDULE_MAX_CONSECUTIVE_FAILURES == 5
+
+    @pytest.mark.parametrize("bad", [0, -1])
+    def test_non_positive_threshold_rejected(self, bad):
+        # 0 or negative would auto-pause a schedule on its very first failure.
+        with pytest.raises(ValueError):
+            Settings(SECRET_KEY="x", SCHEDULE_MAX_CONSECUTIVE_FAILURES=bad)
diff --git a/backend/tests/unit/test_connection_storage_atomic.py b/backend/tests/unit/test_connection_storage_atomic.py
index 6e9170d..a3f94fa 100644
--- a/backend/tests/unit/test_connection_storage_atomic.py
+++ b/backend/tests/unit/test_connection_storage_atomic.py
@@ -1,6 +1,7 @@
 """Tests for atomic, locked connections-file storage (v1.4 H4)."""
 
 import os
+from pathlib import Path
 
 import pytest
 
@@ -104,6 +105,9 @@ def test_interleaved_read_modify_write_preserves_both(self, tmp_path):
         from app.core.config import settings as app_settings
 
         parent_key = app_settings.ENCRYPTION_KEY or os.environ.get("ENCRYPTION_KEY", "")
+        # Workers `import app`, so run them from the backend dir regardless of
+        # where pytest was launched.
+        backend_dir = str(Path(__file__).resolve().parents[2])
         procs = []
         for i in range(n):
             procs.append(
@@ -117,13 +121,22 @@ def test_interleaved_read_modify_write_preserves_both(self, tmp_path):
                         str(ready_dir),
                         str(n),
                     ],
-                    cwd=os.getcwd(),
+                    cwd=backend_dir,
                     stdout=subprocess.PIPE,
                     stderr=subprocess.PIPE,
                     env={**os.environ, "ENCRYPTION_KEY": parent_key},
                 )
             )
-        outs = [p.communicate(timeout=60) for p in procs]
+        outs = []
+        for p in procs:
+            try:
+                outs.append(p.communicate(timeout=60))
+            except subprocess.TimeoutExpired:
+                for q in procs:
+                    if q.poll() is None:
+                        q.kill()
+                        q.communicate()  # drain pipes, populate returncode
+                raise
         failed = [
             (p.returncode, err.decode()[-500:])
             for p, (_, err) in zip(procs, outs)
diff --git a/backend/tests/unit/test_scheduler.py b/backend/tests/unit/test_scheduler.py
index fced82d..15bf914 100644
--- a/backend/tests/unit/test_scheduler.py
+++ b/backend/tests/unit/test_scheduler.py
@@ -164,6 +164,44 @@ def test_below_threshold_does_not_pause(self, scheduler):
         assert schedule.is_active is True
         mock_remove.assert_not_called()
 
+    def test_auto_pause_is_persisted_before_the_job_is_removed(self, scheduler):
+        """If the commit failed after the job was gone, the DB would still say
+        active while nothing dispatches — so the order must be commit, then remove."""
+        sched, db = scheduler
+        schedule = _make_schedule(consecutive_failures=4, is_active=True)
+        db.query.return_value.filter.return_value.first.return_value = schedule
+        order: list[str] = []
+        db.commit.side_effect = lambda: order.append("commit")
+
+        with (
+            patch("app.services.scheduler.enqueue_run", side_effect=RuntimeError("down")),
+            patch.object(sched, "remove_schedule", side_effect=lambda tid: order.append("remove")),
+            patch("app.core.config.settings") as mock_settings,
+        ):
+            mock_settings.SCHEDULE_MAX_CONSECUTIVE_FAILURES = 5
+            sched._execute_scheduled_task(7, "Nightly import")
+
+        assert order == ["commit", "remove"]
+        assert schedule.is_active is False
+
+    def test_job_removal_failure_keeps_the_persisted_pause(self, scheduler):
+        sched, db = scheduler
+        schedule = _make_schedule(consecutive_failures=4, is_active=True)
+        db.query.return_value.filter.return_value.first.return_value = schedule
+
+        with (
+            patch("app.services.scheduler.enqueue_run", side_effect=RuntimeError("down")),
+            patch.object(sched, "remove_schedule", side_effect=RuntimeError("apscheduler gone")),
+            patch("app.core.config.settings") as mock_settings,
+        ):
+            mock_settings.SCHEDULE_MAX_CONSECUTIVE_FAILURES = 5
+            sched._execute_scheduled_task(7, "Nightly import")  # must not raise
+
+        assert schedule.is_active is False
+        db.commit.assert_called_once()
+        # Only the initial rollback that precedes the counter transaction.
+        db.rollback.assert_called_once()
+
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
diff --git a/backend/tests/unit/test_tasks_partial_update.py b/backend/tests/unit/test_tasks_partial_update.py
index 834bf02..73d8b67 100644
--- a/backend/tests/unit/test_tasks_partial_update.py
+++ b/backend/tests/unit/test_tasks_partial_update.py
@@ -160,3 +160,57 @@ def test_create_rejects_non_http_scheme(self, client):
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
+
+
+class TestUpsertInvariant:
+    """upsert_enabled without any key must be rejected, at create and update time."""
+
+    def test_create_enabled_without_keys_is_422(self, client):
+        resp = client.post(
+            "/api/v1/tasks/",
+            json={
+                "name": "No Keys",
+                "connection_id": "conn-1",
+                "endpoint_path": "https://api.example.com/users",
+                "dest_table": "users",
+                "upsert_enabled": True,
+            },
+        )
+        assert resp.status_code == 422
+
+    def test_enable_without_stored_or_supplied_keys_is_400(self, client):
+        task_id = _create_task(client, "Enable No Keys")["id"]
+        resp = client.put(f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True})
+        assert resp.status_code == 400
+        assert "upsert_keys" in resp.json()["detail"]
+
+    def test_enable_with_supplied_keys_then_toggle_uses_stored_keys(self, client, test_db):
+        task_id = _create_task(client, "Enable With Keys")["id"]
+        resp = client.put(
+            f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True, "upsert_keys": ["id"]}
+        )
+        assert resp.status_code == 200, resp.text
+        resp = client.put(f"/api/v1/tasks/{task_id}", json={"upsert_enabled": False})
+        assert resp.status_code == 200, resp.text
+        # Stored keys satisfy the invariant when re-enabling without resending them.
+        resp = client.put(f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True})
+        assert resp.status_code == 200, resp.text
+        row = test_db.query(Task).filter(Task.id == task_id).first()
+        assert row.upsert_enabled is True
+
+    def test_enable_with_explicit_empty_keys_is_422(self, client):
+        task_id = _create_task(client, "Enable Empty Keys")["id"]
+        resp = client.put(
+            f"/api/v1/tasks/{task_id}", json={"upsert_enabled": True, "upsert_keys": []}
+        )
+        assert resp.status_code == 422
+
+
+class TestRelativeEndpointPath:
+    def test_relative_path_with_absolute_url_in_query_is_kept(self, client):
+        """A relative endpoint whose query embeds an absolute URL is not an SSRF target."""
+        task_id = _create_task(client, "Relative Path")["id"]
+        path = "/api/items?redirect=https://cb.example.com/done"
+        resp = client.put(f"/api/v1/tasks/{task_id}", json={"endpoint_path": path})
+        assert resp.status_code == 200, resp.text
+        assert resp.json()["endpoint_path"] == path
diff --git a/backend/tests/unit/test_upsert_logic.py b/backend/tests/unit/test_upsert_logic.py
index 01867e5..2daf152 100644
--- a/backend/tests/unit/test_upsert_logic.py
+++ b/backend/tests/unit/test_upsert_logic.py
@@ -497,6 +497,34 @@ def test_duplicate_keys_first_row_wins(self, task):
             "duplicate upsert key" in str(d.get("error", "")) for d in results["error_details"]
         )
 
+    def test_rows_without_key_values_are_not_collapsed(self, task):
+        """Rows lacking the upsert key cannot collide on it; they must not be
+        folded into one 'duplicate' — the pre-dedupe behaviour (insert) stands."""
+        db = MagicMock()
+        db.execute.return_value.fetchall.return_value = []
+
+        rows = [{"name": "A"}, {"name": "B"}]
+        results = process_rows_with_upsert(db, task, 1, rows)
+
+        assert results["skipped"] == 0
+        assert results["inserted"] == 2
+        assert not any("duplicate" in str(d.get("error", "")) for d in results["error_details"])
+
+    def test_duplicate_report_uses_the_callers_row_index(self, task):
+        db = MagicMock()
+        db.execute.return_value.fetchall.return_value = []
+
+        rows = [
+            {"employee_id": 1, "name": "First"},
+            {"employee_id": 1, "name": "Dup"},  # index 1
+            {"employee_id": 2, "name": "Other"},  # must still be attributed as index 2
+        ]
+        results = process_rows_with_upsert(db, task, 1, rows)
+
+        dups = [d for d in results["error_details"] if "duplicate" in str(d.get("error", ""))]
+        assert [d["row_index"] for d in dups] == [1]
+        assert results["inserted"] == 2
+
 
 class TestBatchSkipCondition:
     """Regression tests for the unimplemented skip condition in the batch
@@ -558,3 +586,38 @@ def test_row_marked_processed_is_skipped(self, sqlite_dest, task_with_skip):
 
 if __name__ == "__main__":
     pytest.main([__file__, "-v"])
+
+
+def test_none_column_preserved_while_sibling_column_updates(sqlite_dest_factory):
+    """Exercise the ``ELSE <col>`` branch of the bulk CASE update.
+
+    A row with one None column and one non-None column must emit a SET clause
+    (so the CASE actually executes) and still keep the stored value for the
+    None column.
+    """
+    from unittest.mock import MagicMock
+
+    from sqlalchemy import text
+
+    from app.services.runner import _bulk_update_rows
+
+    session = sqlite_dest_factory(
+        "CREATE TABLE EMPLOYEES (employee_id INTEGER PRIMARY KEY, name TEXT, dept TEXT)"
+    )
+    session.execute(
+        text("INSERT INTO EMPLOYEES (employee_id, name, dept) VALUES (1, 'Original', 'OldDept')")
+    )
+    session.commit()
+
+    task = MagicMock()
+    task.dest_table = "EMPLOYEES"
+    task.upsert_keys = ["employee_id"]
+
+    count = _bulk_update_rows(
+        session, task, [(0, {"employee_id": 1, "name": None, "dept": "NewDept"})], ["employee_id"]
+    )
+    session.commit()
+
+    row = session.execute(text("SELECT name, dept FROM EMPLOYEES WHERE employee_id = 1")).fetchone()
+    assert count == 1
+    assert tuple(row) == ("Original", "NewDept")
diff --git a/backend/tests/unit/test_url_guard.py b/backend/tests/unit/test_url_guard.py
index 99f27f4..ad2d0fc 100644
--- a/backend/tests/unit/test_url_guard.py
+++ b/backend/tests/unit/test_url_guard.py
@@ -90,3 +90,29 @@ async def test_fetch_json_rejects_invalid_port_as_ssrf_error(self):
         # validation failure), not escape as ValueError -> HTTP 500.
         with pytest.raises(SSRFBlockedError):
             await fetch_json("GET", "http://example.com:99999/")
+
+
+class TestInvalidPort:
+    @pytest.mark.parametrize("url", ["http://8.8.8.8:99999/", "https://example.com:70000/x"])
+    def test_out_of_range_port_rejected_without_resolving(self, url):
+        # Must be an SSRFBlockedError on every path, including literal IPs and
+        # config-time validation (resolve=False), not a ValueError/500.
+        with pytest.raises(SSRFBlockedError, match="invalid port"):
+            validate_url(url, resolve=False)
+
+
+class TestFetchSampleResponsePreservesSsrfError:
+    @pytest.mark.asyncio
+    async def test_ssrf_block_is_not_flattened_into_value_error(self):
+        from unittest.mock import patch
+
+        from app.services.api_connector import fetch_sample_response
+
+        with patch(
+            "app.services.api_connector.fetch_json",
+            side_effect=SSRFBlockedError("Host 'x' resolves to a non-public address"),
+        ):
+            with pytest.raises(SSRFBlockedError):
+                await fetch_sample_response(
+                    method="GET", url="https://api.example.com/x", auth_type="none"
+                )
diff --git a/backend/tests/unit/test_url_redaction.py b/backend/tests/unit/test_url_redaction.py
new file mode 100644
index 0000000..d678948
--- /dev/null
+++ b/backend/tests/unit/test_url_redaction.py
@@ -0,0 +1,30 @@
+"""Tests for the log-safe URL redaction helper."""
+
+import pytest
+
+from app.services.api_connector import _redact_url_for_log
+
+
+@pytest.mark.parametrize(
+    "raw,expected",
+    [
+        ("https://api.example.com/v1/users?api_key=SECRET", "https://api.example.com/v1/users"),
+        ("https://user:s3cret@api.example.com/v1/users", "https://api.example.com/v1/users"),
+        ("https://api.example.com:8443/v1/users#frag", "https://api.example.com:8443/v1/users"),
+        ("https://user:s3cret@[::1]:8080/p?x=1", "https://[::1]:8080/p"),
+        ("https://api.example.com/v1/users", "https://api.example.com/v1/users"),
+        (
+            "https://api.example.com/v1;jsessionid=SECRET/users;v=2?x=1",
+            "https://api.example.com/v1/users",
+        ),
+    ],
+)
+def test_query_userinfo_and_fragment_are_dropped(raw, expected):
+    assert _redact_url_for_log(raw) == expected
+
+
+def test_control_characters_cannot_forge_log_lines():
+    forged = "https://api.example.com/v1\r\n2026-01-01 ERROR fake line?token=abc"
+    redacted = _redact_url_for_log(forged)
+    assert "\r" not in redacted and "\n" not in redacted
+    assert "token" not in redacted
diff --git a/docker-compose.yml b/docker-compose.yml
index 54fccae..ac36d19 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -13,7 +13,7 @@ services:
       # For the authenticated healthcheck ($$REDIS_PASSWORD at runtime)
       REDIS_PASSWORD: ${REDIS_PASSWORD:?REDIS_PASSWORD required in .env}
     healthcheck:
-      test: ["CMD-SHELL", "redis-cli -a $$REDIS_PASSWORD ping | grep PONG"]
+      test: ["CMD-SHELL", "redis-cli -a \"$$REDIS_PASSWORD\" ping | grep PONG"]
       interval: 10s
       timeout: 5s
       retries: 5
diff --git a/frontend/package.json b/frontend/package.json
index a2a1b11..13d7824 100644
--- a/frontend/package.json
+++ b/frontend/package.json
@@ -3,6 +3,9 @@
   "private": true,
   "version": "0.1.0",
   "type": "module",
+  "engines": {
+    "node": "^20.19.0 || >=22.12.0"
+  },
   "scripts": {
     "dev": "vite",
     "build": "tsc -b && vite build",
diff --git a/frontend/src/pages/RunDetail.tsx b/frontend/src/pages/RunDetail.tsx
index 88b19c3..841a928 100644
--- a/frontend/src/pages/RunDetail.tsx
+++ b/frontend/src/pages/RunDetail.tsx
@@ -157,7 +157,13 @@ export function RunDetail() {
 
       {/* Row Errors */}
       {run.row_errors && run.row_errors.length > 0 && (
-        <Card title={`Row Errors (${run.row_errors_total ?? run.row_errors.length})`}>
+        <Card
+          title={
+            run.row_errors_total != null && run.row_errors_total > run.row_errors.length
+              ? `Row Errors (showing ${run.row_errors.length} of ${run.row_errors_total})`
+              : `Row Errors (${run.row_errors.length})`
+          }
+        >
           <Table
             columns={errorColumns}
             dataSource={run.row_errors.map((err, idx) => ({ ...err, key: idx }))}
diff --git a/frontend/src/types/index.ts b/frontend/src/types/index.ts
index 1d9f77b..4c6d1eb 100644
--- a/frontend/src/types/index.ts
+++ b/frontend/src/types/index.ts
@@ -266,7 +266,8 @@ export interface TaskRun {
   replay_of_run_id?: number | null
   execution_logs?: TaskLog[]
   row_errors?: TaskRunLog[]
-  // Uncapped count of row errors (row_errors list is capped at 500 by the API)
+  // Uncapped count of row errors. row_errors itself is capped by the
+  // row_errors_limit query parameter (default 500, max 5000).
   row_errors_total?: number
 }