diff --git a/apps/api/scripts/backfill_map_unit_indexes.py b/apps/api/scripts/backfill_map_unit_indexes.py
index 78a88f6c..c0301d38 100644
--- a/apps/api/scripts/backfill_map_unit_indexes.py
+++ b/apps/api/scripts/backfill_map_unit_indexes.py
@@ -15,10 +15,22 @@
 from pathlib import Path
 
 
+def _resolve_shared_root(api_root: Path) -> Path:
+    """Resolve the shared package in source checkouts and runtime images."""
+    runtime_shared_root = api_root / "packages" / "shared-python"
+    if runtime_shared_root.is_dir():
+        return runtime_shared_root
+
+    repository_shared_root = api_root.parents[1] / "packages" / "shared-python"
+    if repository_shared_root.is_dir():
+        return repository_shared_root
+
+    raise RuntimeError(f"Could not locate shared-python package from {api_root}")
+
+
 def _bootstrap_python_path() -> None:
     api_root = Path(__file__).resolve().parents[1]
-    repo_root = api_root.parents[1]
-    shared_root = repo_root / "packages" / "shared-python"
+    shared_root = _resolve_shared_root(api_root)
     for path in (api_root, shared_root):
         value = os.fspath(path)
         if value not in sys.path:
@@ -44,7 +56,9 @@ def _build_parser() -> argparse.ArgumentParser:
         action="store_true",
         help="Build and commit each current revision index.",
     )
-    parser.add_argument("--document-id", default="", help="Limit the backfill to one document.")
+    parser.add_argument(
+        "--document-id", default="", help="Limit the backfill to one document."
+    )
     return parser
 
 
@@ -62,7 +76,9 @@ def backfill_map_unit_indexes(*, apply: bool, document_id: str = "") -> int:
     documents = _load_documents(document_id)
     if not apply:
         for document in documents:
-            print(f"would backfill document={document.document_id} revision={document.current_job_result_id}")
+            print(
+                f"would backfill document={document.document_id} revision={document.current_job_result_id}"
+            )
         return len(documents)
 
     session_factory = get_sync_session_factory()
diff --git a/apps/api/tests/contract/test_backfill_map_unit_indexes_contract.py b/apps/api/tests/contract/test_backfill_map_unit_indexes_contract.py
new file mode 100644
index 00000000..72c61f91
--- /dev/null
+++ b/apps/api/tests/contract/test_backfill_map_unit_indexes_contract.py
@@ -0,0 +1,28 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+
+def test_backfill_script_resolves_shared_package_from_runtime_image_layout(
+    tmp_path: Path,
+) -> None:
+    from scripts.backfill_map_unit_indexes import _resolve_shared_root
+
+    api_root = tmp_path / "app"
+    shared_root = api_root / "packages" / "shared-python"
+    shared_root.mkdir(parents=True)
+
+    assert _resolve_shared_root(api_root) == shared_root
+
+
+def test_backfill_script_resolves_shared_package_from_source_checkout_layout(
+    tmp_path: Path,
+) -> None:
+    from scripts.backfill_map_unit_indexes import _resolve_shared_root
+
+    repository_root = tmp_path / "repository"
+    api_root = repository_root / "apps" / "api"
+    shared_root = repository_root / "packages" / "shared-python"
+    shared_root.mkdir(parents=True)
+
+    assert _resolve_shared_root(api_root) == shared_root
diff --git a/apps/api/tests/contract/test_logging_security_contract.py b/apps/api/tests/contract/test_logging_security_contract.py
new file mode 100644
index 00000000..fbe5a1e8
--- /dev/null
+++ b/apps/api/tests/contract/test_logging_security_contract.py
@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class _FakeLogfireExceptionHelper:
+    exception: BaseException
+    is_recording_exception: bool = True
+
+    def no_record_exception(self) -> None:
+        self.is_recording_exception = False
+
+
+def test_redact_sensitive_text_masks_postgresql_url_credentials() -> None:
+    from shared.core.logging import redact_sensitive_text
+
+    message = (
+        "invalid dsn after "
+        "postgresql+psycopg2://postgres:super-secret@database.example:5432/knowhere"
+    )
+
+    redacted = redact_sensitive_text(message)
+
+    assert "super-secret" not in redacted
+    assert "postgresql+psycopg2://[REDACTED]@database.example:5432/knowhere" in redacted
+
+
+def test_logfire_callback_does_not_export_exception_with_database_credentials() -> None:
+    from shared.core.logging import _downgrade_expected_logfire_exception
+
+    helper = _FakeLogfireExceptionHelper(
+        exception=RuntimeError(
+            "invalid dsn: postgresql://postgres:super-secret@database.example/knowhere"
+        )
+    )
+
+    _downgrade_expected_logfire_exception(helper)  # pyright: ignore[reportArgumentType]
+
+    assert helper.is_recording_exception is False
diff --git a/packages/shared-python/shared/core/logging.py b/packages/shared-python/shared/core/logging.py
index bb246f18..b4be391c 100644
--- a/packages/shared-python/shared/core/logging.py
+++ b/packages/shared-python/shared/core/logging.py
@@ -1,4 +1,5 @@
 import logging
+import re
 import sys
 from contextlib import contextmanager
 from contextvars import ContextVar
@@ -15,6 +16,10 @@
     "{time:YYYY-MM-DD HH:mm:ss.SSS} | {level:<8} | {extra[event]} | {message}"
 )
 _DEVELOPMENT_CONSOLE_FORMAT = "<green>{time:YYYY-MM-DD HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level> <cyan>{extra}</cyan>"
+_DATABASE_URL_WITH_CREDENTIALS_PATTERN = re.compile(
+    r"(?P<scheme>postgres(?:ql)?(?:\+[^:/\s]+)?://)(?P<credentials>[^/\s@]+@)",
+    re.IGNORECASE,
+)
 
 
 class LogEvent(Enum):
@@ -79,6 +84,20 @@ def get_log_context() -> Dict[str, Any]:
     return _log_context.get().copy()
 
 
+def redact_sensitive_text(value: object) -> str:
+    """Redact credentials embedded in PostgreSQL URLs before they are logged."""
+    text = str(value)
+    return _DATABASE_URL_WITH_CREDENTIALS_PATTERN.sub(
+        r"\g<scheme>[REDACTED]@",
+        text,
+    )
+
+
+def contains_database_credentials(value: object) -> bool:
+    """Return whether text contains a PostgreSQL URL user-info component."""
+    return _DATABASE_URL_WITH_CREDENTIALS_PATTERN.search(str(value)) is not None
+
+
 def _is_expected_client_exception(exception: BaseException) -> bool:
     """Identify handled 4xx exceptions that should stay warnings in Logfire."""
     from shared.core.exceptions.knowhere_exception import KnowhereException
@@ -105,6 +124,13 @@ def _downgrade_expected_logfire_exception(
     helper: "ExceptionCallbackHelper",
 ) -> None:
     """Prevent handled client errors from creating Logfire exception issues."""
+    if contains_database_credentials(helper.exception):
+        # Logfire keeps exception.message and exception.stacktrace as safe keys,
+        # so its normal scrubber does not redact credentials embedded in them.
+        # Do not export the exception object when it contains a database URL.
+        helper.no_record_exception()
+        return
+
     if not _is_expected_client_exception(helper.exception):
         return
 
@@ -280,6 +306,11 @@ def emit(self, record: logging.LogRecord) -> None:
             frame = frame.f_back
             depth += 1
 
-        logger.opt(depth=depth, exception=record.exc_info).log(
-            level, record.getMessage()
-        )
+        message = redact_sensitive_text(record.getMessage())
+        exception = record.exc_info
+        if exception is not None and contains_database_credentials(exception[1]):
+            # The exception object would otherwise be serialized by Logfire with
+            # its raw message and stacktrace, bypassing message scrubbing.
+            exception = None
+
+        logger.opt(depth=depth, exception=exception).log(level, message)