@@ -349,6 +349,29 @@ def _iter_effective_handlers(logger: logging.Logger) -> Iterator[logging.Handler
         current = current.parent
 
 
+def _iter_namespace_handlers(logger: logging.Logger) -> Iterator[logging.Handler]:
+    """Yield effective and descendant-owned handlers for one logger namespace."""
+    seen: set[int] = set()
+
+    for handler in _iter_effective_handlers(logger):
+        seen.add(id(handler))
+        yield handler
+
+    root_logger = logging.getLogger()
+    prefix = None if logger is root_logger else f"{logger.name}."
+    for candidate in tuple(logging.Logger.manager.loggerDict.values()):
+        if not isinstance(candidate, logging.Logger) or candidate is logger:
+            continue
+        if prefix is not None and not candidate.name.startswith(prefix):
+            continue
+
+        for handler in candidate.handlers:
+            handler_id = id(handler)
+            if handler_id not in seen:
+                seen.add(handler_id)
+                yield handler
+
+
 def install_handler_filters(
     logger: logging.Logger,
     *,
@@ -359,7 +382,7 @@ def install_handler_filters(
     root_logger = logging.getLogger()
     logger_prefix = None if logger is root_logger else logger.name
 
-    for handler in _iter_effective_handlers(logger):
+    for handler in _iter_namespace_handlers(logger):
         existing = next(
             (
                 candidate
 
@@ -16,7 +16,7 @@
 from collections.abc import Callable
 from re import Pattern
 from types import FrameType
-from typing import Any, BinaryIO, cast
+from typing import IO, Any, cast
 
 from logprivacy.exceptions import InputLimitExceededError
 
@@ -106,8 +106,8 @@ def __init__(self) -> None:
         assert stdout_pipe is not None
 
         self._process: subprocess.Popen[bytes] = process
-        self._stdin: BinaryIO = stdin_pipe
-        self._stdout: BinaryIO = stdout_pipe
+        self._stdin: IO[bytes] = stdin_pipe
+        self._stdout: IO[bytes] = stdout_pipe
         self._responses: queue.Queue[bytes | None] = queue.Queue()
         self._request_lock = threading.Lock()
         self._ready = threading.Event()
@@ -298,7 +298,7 @@ def _shutdown_worker() -> None:
         worker.close()
 
 
-def _read_exact(stream: BinaryIO, size: int) -> bytes | None:
+def _read_exact(stream: IO[bytes], size: int) -> bytes | None:
     chunks: list[bytes] = []
     remaining = size
     while remaining:
 
@@ -249,14 +249,15 @@ def scan_jsonl(
     adapters: AdapterRegistry | None = None,
     on_error: _OnError = "raise",
 ) -> Iterator[JSONLScanRecord]:
-    """Yield scan findings from a JSON Lines source, line by line.
+    """Yield scan findings and incomplete audits from a JSON Lines source.
 
     Each non-empty line is parsed as JSON, then audited for sensitive values.
-    Yields ``JSONLScanRecord`` for lines that contain findings.
-    Lines with no findings are not yielded.
+    A ``JSONLScanRecord`` is yielded when the line contains findings or when the
+    audit could not inspect the complete value. Complete lines with no findings
+    are omitted.
 
     ``"raise"`` raises ``JSONLProcessingError`` for invalid JSON or UTF-8.
-    Because scan records only represent findings, ``"skip"`` and
+    Because scan records represent parsed audits, ``"skip"`` and
     ``"placeholder"`` both omit invalid lines instead of yielding a synthetic
     finding. The original line content is never stored or included in errors.
     """
@@ -286,8 +287,13 @@ def scan_jsonl(
             continue
 
         report = cleaner.audit(parsed)
-        if report.findings:
-            yield JSONLScanRecord(line_number=line_number, findings=report.findings)
+        if report.findings or not report.complete:
+            yield JSONLScanRecord(
+                line_number=line_number,
+                findings=report.findings,
+                complete=report.complete,
+                limitations=report.limitations,
+            )
 
 
 def _validate_on_error(on_error: str) -> None:
 
@@ -277,3 +277,5 @@ class JSONLScanRecord:
 
     line_number: int
     findings: tuple[Finding, ...]
+    complete: bool = True
+    limitations: tuple[str, ...] = ()
@@ -0,0 +1,198 @@
+from __future__ import annotations
+
+import io
+import logging
+
+from logprivacy import LogPrivacyFilter, get_safe_logger
+
+
+def _restore_logger(
+    logger: logging.Logger,
+    *,
+    handlers: list[logging.Handler],
+    filters: list[logging.Filter],
+    level: int,
+    propagate: bool,
+) -> None:
+    logger.handlers[:] = handlers
+    logger.filters[:] = filters
+    logger.setLevel(level)
+    logger.propagate = propagate
+
+
+def test_get_safe_logger_protects_existing_child_handler_without_propagation() -> None:
+    parent = logging.getLogger("child-handler-secure")
+    child = logging.getLogger("child-handler-secure.worker")
+    parent_state = {
+        "handlers": list(parent.handlers),
+        "filters": list(parent.filters),
+        "level": parent.level,
+        "propagate": parent.propagate,
+    }
+    child_state = {
+        "handlers": list(child.handlers),
+        "filters": list(child.filters),
+        "level": child.level,
+        "propagate": child.propagate,
+    }
+    stream = io.StringIO()
+    handler = logging.StreamHandler(stream)
+
+    try:
+        parent.handlers[:] = []
+        parent.propagate = False
+        child.handlers[:] = [handler]
+        child.filters[:] = []
+        child.setLevel(logging.DEBUG)
+        child.propagate = False
+
+        get_safe_logger("child-handler-secure", level=logging.DEBUG)
+        child.error("email=%s password=%s", "john@example.com", "secret123")
+    finally:
+        _restore_logger(parent, **parent_state)
+        _restore_logger(child, **child_state)
+
+    rendered = stream.getvalue()
+    assert "john@example.com" not in rendered
+    assert "secret123" not in rendered
+    assert "email=[EMAIL] password=[SECRET]" in rendered
+
+
+def test_get_safe_logger_deduplicates_filters_on_existing_child_handler() -> None:
+    parent = logging.getLogger("child-handler-deduplicated")
+    child = logging.getLogger("child-handler-deduplicated.worker")
+    parent_state = {
+        "handlers": list(parent.handlers),
+        "filters": list(parent.filters),
+        "level": parent.level,
+        "propagate": parent.propagate,
+    }
+    child_state = {
+        "handlers": list(child.handlers),
+        "filters": list(child.filters),
+        "level": child.level,
+        "propagate": child.propagate,
+    }
+    handler = logging.StreamHandler(io.StringIO())
+
+    try:
+        parent.handlers[:] = []
+        parent.propagate = False
+        child.handlers[:] = [handler]
+        child.propagate = False
+
+        get_safe_logger("child-handler-deduplicated")
+        get_safe_logger("child-handler-deduplicated")
+
+        scoped = [
+            candidate
+            for candidate in handler.filters
+            if isinstance(candidate, LogPrivacyFilter)
+            and candidate.logger_prefix == "child-handler-deduplicated"
+        ]
+        assert len(scoped) == 1
+    finally:
+        _restore_logger(parent, **parent_state)
+        _restore_logger(child, **child_state)
+
+
+def test_get_safe_logger_does_not_install_filters_on_unrelated_child_handlers() -> None:
+    parent = logging.getLogger("child-handler-scope")
+    unrelated = logging.getLogger("other-child-handler.worker")
+    parent_state = {
+        "handlers": list(parent.handlers),
+        "filters": list(parent.filters),
+        "level": parent.level,
+        "propagate": parent.propagate,
+    }
+    unrelated_state = {
+        "handlers": list(unrelated.handlers),
+        "filters": list(unrelated.filters),
+        "level": unrelated.level,
+        "propagate": unrelated.propagate,
+    }
+    handler = logging.StreamHandler(io.StringIO())
+
+    try:
+        parent.handlers[:] = []
+        parent.propagate = False
+        unrelated.handlers[:] = [handler]
+        unrelated.propagate = False
+
+        get_safe_logger("child-handler-scope")
+
+        scoped = [
+            candidate for candidate in handler.filters if isinstance(candidate, LogPrivacyFilter)
+        ]
+        assert scoped == []
+    finally:
+        _restore_logger(parent, **parent_state)
+        _restore_logger(unrelated, **unrelated_state)
+
+
+def test_get_safe_logger_does_not_match_similar_namespace_prefix() -> None:
+    parent = logging.getLogger("app")
+    unrelated = logging.getLogger("application.worker")
+    parent_state = {
+        "handlers": list(parent.handlers),
+        "filters": list(parent.filters),
+        "level": parent.level,
+        "propagate": parent.propagate,
+    }
+    unrelated_state = {
+        "handlers": list(unrelated.handlers),
+        "filters": list(unrelated.filters),
+        "level": unrelated.level,
+        "propagate": unrelated.propagate,
+    }
+    handler = logging.StreamHandler(io.StringIO())
+
+    try:
+        parent.handlers[:] = []
+        parent.propagate = False
+        unrelated.handlers[:] = [handler]
+        unrelated.propagate = False
+
+        get_safe_logger("app")
+
+        assert not any(isinstance(candidate, LogPrivacyFilter) for candidate in handler.filters)
+    finally:
+        _restore_logger(parent, **parent_state)
+        _restore_logger(unrelated, **unrelated_state)
+
+
+def test_shared_handler_receives_one_scoped_filter() -> None:
+    parent = logging.getLogger("shared-handler")
+    child = logging.getLogger("shared-handler.worker")
+    parent_state = {
+        "handlers": list(parent.handlers),
+        "filters": list(parent.filters),
+        "level": parent.level,
+        "propagate": parent.propagate,
+    }
+    child_state = {
+        "handlers": list(child.handlers),
+        "filters": list(child.filters),
+        "level": child.level,
+        "propagate": child.propagate,
+    }
+    handler = logging.StreamHandler(io.StringIO())
+
+    try:
+        parent.handlers[:] = [handler]
+        parent.propagate = False
+        child.handlers[:] = [handler]
+        child.propagate = False
+
+        get_safe_logger("shared-handler")
+
+        scoped = [
+            candidate
+            for candidate in handler.filters
+            if isinstance(candidate, LogPrivacyFilter)
+            and candidate.logger_prefix == "shared-handler"
+        ]
+        assert len(scoped) == 1
+    finally:
+        _restore_logger(parent, **parent_state)
+        _restore_logger(child, **child_state)
@@ -0,0 +1,52 @@
+from __future__ import annotations
+
+from dataclasses import replace
+from pathlib import Path
+
+from logprivacy.jsonl import scan_jsonl
+from logprivacy.policy import CleanerPolicy
+from logprivacy.result import JSONLScanRecord
+
+
+def test_scan_jsonl_yields_incomplete_audit_without_findings(tmp_path: Path) -> None:
+    source = tmp_path / "input.jsonl"
+    source.write_text('{"outer":{"status":"ok"}}\n', encoding="utf-8")
+    policy = replace(CleanerPolicy.default(), max_depth=0)
+
+    records = list(scan_jsonl(source, policy=policy))
+
+    assert len(records) == 1
+    assert records[0].line_number == 1
+    assert records[0].findings == ()
+    assert records[0].complete is False
+    assert records[0].limitations == ("max_depth",)
+
+
+def test_scan_jsonl_preserves_findings_when_audit_is_incomplete(tmp_path: Path) -> None:
+    source = tmp_path / "input.jsonl"
+    source.write_text(
+        '{"email":"user@example.com","nested":{"status":"ok"}}\n',
+        encoding="utf-8",
+    )
+    policy = replace(CleanerPolicy.default(), max_items=1)
+
+    records = list(scan_jsonl(source, policy=policy))
+
+    assert len(records) == 1
+    assert records[0].complete is False
+    assert records[0].limitations == ("max_items",)
+    assert {finding.category for finding in records[0].findings} == {"email"}
+
+
+def test_scan_jsonl_still_omits_complete_lines_without_findings(tmp_path: Path) -> None:
+    source = tmp_path / "input.jsonl"
+    source.write_text('{"status":"ok"}\n', encoding="utf-8")
+
+    assert list(scan_jsonl(source)) == []
+
+
+def test_jsonl_scan_record_defaults_preserve_existing_construction() -> None:
+    record = JSONLScanRecord(line_number=3, findings=())
+
+    assert record.complete is True
+    assert record.limitations == ()