diff --git a/docs/engineering/von_workflow_language_manual.md b/docs/engineering/von_workflow_language_manual.md
index 812da758d..f4666d578 100644
--- a/docs/engineering/von_workflow_language_manual.md
+++ b/docs/engineering/von_workflow_language_manual.md
@@ -240,6 +240,13 @@ Runtime semantics:
   transport facts under `mcp_transport`, including execution identity,
   queue/handler/transport timing, deadline, timeout phase, and the explicit
   `discard_from_turn` late-result policy;
+- the transport applies the same remaining hard-deadline budget as a PyMongo
+  client-side operation timeout around the handler. Nested Vontology database
+  operations therefore release their bounded isolation worker at the deadline
+  instead of continuing under a succession of independent driver timeouts. A
+  small bounded part of that budget is reserved for caught database exceptions
+  to unwind into the typed terminal outcome. Non-database handlers remain
+  bounded by the fixed worker pool and cooperative cancellation scope;
 - a late handler completion cannot rewrite the workflow action outcome. Reads
   expose represented recovery affordances such as bounded retry or alternate
   path selection; writes report an indeterminate mutation outcome and require
diff --git a/src/backend/integrations/internal_mcp/transport.py b/src/backend/integrations/internal_mcp/transport.py
index a42391e56..8796501bc 100644
--- a/src/backend/integrations/internal_mcp/transport.py
+++ b/src/backend/integrations/internal_mcp/transport.py
@@ -38,6 +38,9 @@
 from datetime import datetime, timezone
 from typing import Any, Callable, Dict, Mapping
 
+from pymongo import timeout as pymongo_timeout
+from pymongo.errors import PyMongoError
+
 logger = logging.getLogger(__name__)
 
 _DEFAULT_READ_ADVISORY_TIMEOUT_SEC = 6.0
@@ -51,6 +54,17 @@
 # scheduling tolerance applied at handler start; materially queued work must
 # still retain its configured minimum.
 _EFFECT_ADMISSION_START_TOLERANCE_SEC = 0.005
+# Leave a small bounded interval for a database deadline to unwind through
+# legacy handlers that catch and return exceptions before the transport itself
+# reaches its terminal deadline.  This prevents a caught CSOT expiry from
+# becoming an apparently completed error payload or a misleading late result.
+_MONGO_DEADLINE_COMPLETION_RESERVE_SEC = 0.05
+_MONGO_DEADLINE_COMPLETION_RESERVE_FRACTION = 0.1
+_MONGO_DEADLINE_CLASSIFICATION_TOLERANCE_SEC = 0.1
+_MONGO_CSOT_ADMISSION_REFUSAL_MARKER = (
+    "operation would exceed time limit, remaining timeout:"
+)
+_MONGO_CSOT_CONFIGURED_TIMEOUT_MARKER = "configured timeouts: timeoutms:"
 _LATE_COMPLETION_HISTORY_LIMIT = 50
 _LATE_COMPLETION_MAX_PAYLOAD_CHARS = 32_000
 _LATE_COMPLETION_MAX_RECEIPT_FIELD_CHARS = 4_000
@@ -162,6 +176,60 @@ class InternalMCPHandlerCancelled(RuntimeError):
     """Raised by cooperative handlers after the transport requests cancellation."""
 
 
+class InternalMCPHandlerDeadlineExceeded(InternalMCPHandlerCancelled):
+    """Raised when a handler's database work consumes its transport deadline."""
+
+
+def _mongo_timeout_consumed_transport_deadline(
+    exc: PyMongoError,
+    *,
+    scope: InternalMCPExecutionScope,
+) -> bool:
+    if not bool(getattr(exc, "timeout", False)):
+        return False
+    details = getattr(exc, "details", None)
+    detail_message = (
+        str(details.get("errmsg") or "")
+        if isinstance(details, Mapping)
+        else ""
+    )
+    if _MONGO_CSOT_ADMISSION_REFUSAL_MARKER in (
+        f"{exc} {detail_message}".lower()
+    ):
+        return True
+    return (
+        scope.remaining_seconds
+        <= _MONGO_DEADLINE_CLASSIFICATION_TOLERANCE_SEC
+    )
+
+
+def _handler_payload_reports_mongo_deadline(
+    payload: Any,
+    *,
+    scope: InternalMCPExecutionScope,
+) -> bool:
+    if not isinstance(payload, Mapping):
+        return False
+    messages = [
+        payload.get("error"),
+        payload.get("message"),
+        payload.get("detail"),
+    ]
+    details = payload.get("details")
+    if isinstance(details, Mapping):
+        messages.extend((details.get("error"), details.get("errmsg")))
+    combined_message = " ".join(
+        str(message or "") for message in messages
+    ).lower()
+    if _MONGO_CSOT_ADMISSION_REFUSAL_MARKER in combined_message:
+        return True
+    return bool(
+        _MONGO_CSOT_CONFIGURED_TIMEOUT_MARKER in combined_message
+        and scope.remaining_seconds
+        <= _MONGO_DEADLINE_CLASSIFICATION_TOLERANCE_SEC
+    )
+
+
 def get_internal_mcp_execution_scope() -> InternalMCPExecutionScope | None:
     """Return the active handler deadline/cancellation scope, if any."""
 
@@ -255,6 +323,7 @@ class _HandlerTask:
     completed_monotonic: float | None = None
     result: Any = None
     exception: BaseException | None = None
+    deadline_exceeded_during_handler: bool = False
     terminal_returned: bool = False
     cancelled_before_start: bool = False
     admission_denied_before_start: bool = False
@@ -270,7 +339,51 @@ def _invoke(self) -> Any:
         )
         token = _ACTIVE_EXECUTION_SCOPE.set(scope)
         try:
-            return self.handler(**self.payload)
+            remaining_seconds = scope.remaining_seconds
+            if remaining_seconds <= 0.0:
+                raise InternalMCPHandlerDeadlineExceeded(
+                    f"Internal MCP execution {scope.execution_id} reached its "
+                    "deadline before the handler started."
+                )
+            database_completion_reserve_sec = min(
+                _MONGO_DEADLINE_COMPLETION_RESERVE_SEC,
+                remaining_seconds
+                * _MONGO_DEADLINE_COMPLETION_RESERVE_FRACTION,
+            )
+            database_timeout_sec = (
+                remaining_seconds - database_completion_reserve_sec
+            )
+            try:
+                # PyMongo CSOT is context-local, applies the remaining budget
+                # across all nested database operations, and leaves only a
+                # small bounded interval for caught exceptions to unwind before
+                # the transport deadline. Non-Mongo handlers are unaffected
+                # and remain bounded by the transport worker pool.
+                with pymongo_timeout(database_timeout_sec):
+                    result = self.handler(**self.payload)
+                # Some legacy support handlers convert every exception to an
+                # error mapping.  Preserve their compatibility behaviour for
+                # ordinary errors, but do not let an explicit PyMongo CSOT
+                # admission refusal masquerade as a completed transport call.
+                if _handler_payload_reports_mongo_deadline(
+                    result,
+                    scope=scope,
+                ):
+                    raise InternalMCPHandlerDeadlineExceeded(
+                        f"Internal MCP execution {scope.execution_id} consumed "
+                        "its database operation deadline."
+                    )
+                return result
+            except PyMongoError as exc:
+                if _mongo_timeout_consumed_transport_deadline(
+                    exc,
+                    scope=scope,
+                ):
+                    raise InternalMCPHandlerDeadlineExceeded(
+                        f"Internal MCP execution {scope.execution_id} consumed "
+                        "its database operation deadline."
+                    ) from exc
+                raise
         finally:
             _ACTIVE_EXECUTION_SCOPE.reset(token)
 
@@ -316,11 +429,18 @@ def run(self) -> None:
         with self.lock:
             self.result = result
             self.exception = exception
+            self.deadline_exceeded_during_handler = isinstance(
+                exception,
+                InternalMCPHandlerDeadlineExceeded,
+            )
             self.completed_at = completed_at
             self.completed_monotonic = completed_monotonic
             was_late = bool(
-                self.terminal_returned
-                or completed_monotonic > self.deadline_monotonic
+                not self.deadline_exceeded_during_handler
+                and (
+                    self.terminal_returned
+                    or completed_monotonic > self.deadline_monotonic
+                )
             )
             # Wake the caller before any potentially blocking durable observer.
             self.done_event.set()
@@ -652,6 +772,7 @@ def _timeout_payload(
             "handler_elapsed_ms": handler_elapsed_ms,
             "cancellation_requested": True,
             "handler_isolation": "bounded_worker_pool",
+            "database_deadline_propagation": "pymongo_csot",
             "outcome_finality": "terminal_for_turn",
             "late_result_policy": late_result_policy,
         }
@@ -943,6 +1064,7 @@ def execute(
                 (completed or task.done_event.is_set())
                 and task.completed_monotonic is not None
                 and task.completed_monotonic <= handler_deadline_monotonic
+                and not task.deadline_exceeded_during_handler
             )
             if completed_within_deadline:
                 task_completed = True
@@ -1114,6 +1236,7 @@ def get_diagnostics(self) -> Dict[str, Any]:
                 "late_completion_count": self._late_completion_count,
                 "late_completions": list(self._late_completions),
                 "late_result_policy": "discard_from_turn",
+                "database_deadline_propagation": "pymongo_csot",
             }
         local["handler_pool"] = self._executor.diagnostics()
         return local
diff --git a/tests/backend/test_internal_mcp_transport_deadlines.py b/tests/backend/test_internal_mcp_transport_deadlines.py
index e2968e88f..acb3cffcd 100644
--- a/tests/backend/test_internal_mcp_transport_deadlines.py
+++ b/tests/backend/test_internal_mcp_transport_deadlines.py
@@ -1,9 +1,14 @@
 from __future__ import annotations
 
+from contextlib import contextmanager
 import sys
 import time
 from threading import Event, Thread
 
+from pymongo.errors import ExecutionTimeout
+import pytest
+
+from src.backend.integrations.internal_mcp import transport as transport_mod
 from src.backend.integrations.internal_mcp.gateway import (
     InternalMCPGateway,
     MethodCatalogue,
@@ -128,6 +133,7 @@ def _handler():
     assert result.payload["outcome_finality"] == "terminal_for_turn"
     assert result.payload["late_result_policy"] == "discard_from_turn"
     assert result.payload["handler_isolation"] == "bounded_worker_pool"
+    assert result.payload["database_deadline_propagation"] == "pymongo_csot"
     assert result.payload["timeout_phase"] == "handler"
     assert result.handler_duration_ms is None
     assert result.handler_elapsed_ms is not None
@@ -148,6 +154,181 @@ def _handler():
     assert method_metrics["last_error"] == "tool_timeout"
 
 
+def test_handler_database_operations_receive_the_remaining_transport_budget(
+    monkeypatch,
+) -> None:
+    observed_timeouts: list[float] = []
+
+    @contextmanager
+    def _capture_timeout(seconds: float):
+        observed_timeouts.append(seconds)
+        yield
+
+    monkeypatch.setattr(transport_mod, "pymongo_timeout", _capture_timeout)
+    transport = InternalMCPTransport(
+        read_timeout_sec=0.2,
+        read_advisory_timeout_sec=0.01,
+    )
+    gateway = _gateway_for(
+        method_name="synthetic_database_read",
+        handler=lambda: {"success": True},
+        transport=transport,
+    )
+
+    result = gateway.invoke("synthetic_database_read", {})
+
+    assert result.outcome == "completed"
+    assert len(observed_timeouts) == 1
+    assert 0.0 < observed_timeouts[0] < 0.2
+
+
+def test_database_deadline_timeout_releases_worker_for_follow_up() -> None:
+    executor = _BoundedHandlerExecutor(worker_count=1, queue_capacity=1)
+    transport = InternalMCPTransport(
+        read_timeout_sec=0.08,
+        read_advisory_timeout_sec=0.01,
+        handler_executor=executor,
+    )
+
+    def _database_read() -> None:
+        scope = get_internal_mcp_execution_scope()
+        assert scope is not None
+        while scope.remaining_seconds > 0.0:
+            time.sleep(0.001)
+        raise ExecutionTimeout("simulated MongoDB CSOT expiry")
+
+    gateway = _gateway_for(
+        method_name="synthetic_database_deadline",
+        handler=_database_read,
+        transport=transport,
+    )
+
+    result = gateway.invoke("synthetic_database_deadline", {})
+
+    assert result.outcome == "timed_out"
+    assert result.payload["error_code"] == "tool_timeout"
+    assert result.payload["timeout_phase"] == "handler"
+    assert result.payload["late_result_policy"] == "discard_from_turn"
+    assert result.payload["database_deadline_propagation"] == "pymongo_csot"
+    assert transport.get_diagnostics()["late_completion_count"] == 0
+
+    follow_up_gateway = _gateway_for(
+        method_name="synthetic_read_after_database_deadline",
+        handler=lambda: {"success": True},
+        transport=transport,
+    )
+    follow_up = follow_up_gateway.invoke(
+        "synthetic_read_after_database_deadline",
+        {},
+    )
+    assert follow_up.outcome == "completed"
+    assert follow_up.payload == {"success": True}
+    assert executor.diagnostics()["active_worker_count"] == 0
+    assert executor.diagnostics()["completed_count"] == 2
+
+
+def test_pymongo_csot_admission_refusal_is_typed_transport_timeout() -> None:
+    message = (
+        "operation would exceed time limit, remaining timeout:0.10516 "
+        "<= network round trip time:0.29659"
+    )
+    transport = InternalMCPTransport(
+        read_timeout_sec=0.5,
+        read_advisory_timeout_sec=0.1,
+    )
+    gateway = _gateway_for(
+        method_name="synthetic_csot_admission_refusal",
+        handler=lambda: (_ for _ in ()).throw(
+            ExecutionTimeout(
+                message,
+                50,
+                {"ok": 0, "errmsg": message, "code": 50},
+            )
+        ),
+        transport=transport,
+    )
+
+    result = gateway.invoke("synthetic_csot_admission_refusal", {})
+
+    assert result.outcome == "timed_out"
+    assert result.duration_ms < 100.0
+    assert result.payload["error_code"] == "tool_timeout"
+    assert result.payload["database_deadline_propagation"] == "pymongo_csot"
+
+
+def test_swallowed_pymongo_csot_error_cannot_masquerade_as_success() -> None:
+    message = (
+        "operation would exceed time limit, remaining timeout:0.19221 "
+        "<= network round trip time:0.28518"
+    )
+    transport = InternalMCPTransport(
+        read_timeout_sec=0.5,
+        read_advisory_timeout_sec=0.1,
+    )
+    gateway = _gateway_for(
+        method_name="synthetic_swallowed_csot_error",
+        handler=lambda: {"error": message, "tree": []},
+        transport=transport,
+    )
+
+    result = gateway.invoke("synthetic_swallowed_csot_error", {})
+
+    assert result.outcome == "timed_out"
+    assert result.payload["success"] is False
+    assert result.payload["error_code"] == "tool_timeout"
+    assert result.payload["database_deadline_propagation"] == "pymongo_csot"
+    assert result.payload.get("tree") is None
+    assert transport.get_diagnostics()["late_completion_count"] == 0
+
+
+def test_swallowed_pymongo_network_timeout_at_deadline_is_not_late_success() -> None:
+    transport = InternalMCPTransport(
+        read_timeout_sec=0.08,
+        read_advisory_timeout_sec=0.01,
+    )
+
+    def _handler() -> dict[str, object]:
+        time.sleep(0.075)
+        return {
+            "error": (
+                "The read operation timed out "
+                "(configured timeouts: timeoutMS: 71.9ms, "
+                "connectTimeoutMS: 5000.0ms)"
+            ),
+            "tree": [],
+        }
+
+    gateway = _gateway_for(
+        method_name="synthetic_swallowed_network_timeout",
+        handler=_handler,
+        transport=transport,
+    )
+
+    result = gateway.invoke("synthetic_swallowed_network_timeout", {})
+
+    assert result.outcome == "timed_out"
+    assert result.payload["error_code"] == "tool_timeout"
+    assert result.duration_ms < 100.0
+    assert transport.get_diagnostics()["late_completion_count"] == 0
+
+
+def test_database_timeout_before_transport_deadline_keeps_handler_error() -> None:
+    transport = InternalMCPTransport(
+        read_timeout_sec=0.5,
+        read_advisory_timeout_sec=0.1,
+    )
+    gateway = _gateway_for(
+        method_name="synthetic_early_database_timeout",
+        handler=lambda: (_ for _ in ()).throw(
+            ExecutionTimeout("database operation failed before transport deadline")
+        ),
+        transport=transport,
+    )
+
+    with pytest.raises(ExecutionTimeout):
+        gateway.invoke("synthetic_early_database_timeout", {})
+
+
 def test_caller_deadline_shortens_the_registered_method_timeout() -> None:
     cancellation_seen = Event()
     handler_deadlines: list[float] = []
diff --git a/tests/backend/test_orchestrator_tool_call_validation.py b/tests/backend/test_orchestrator_tool_call_validation.py
index 669ad38fa..10f55bd8d 100644
--- a/tests/backend/test_orchestrator_tool_call_validation.py
+++ b/tests/backend/test_orchestrator_tool_call_validation.py
@@ -149,13 +149,14 @@ def test_tool_calling_loop_records_terminal_timeout_not_late_success(
         },
     )
 
-    started_at = time.perf_counter()
     result = orchestrator._action_tool_calling_execute(request)
-    elapsed = time.perf_counter() - started_at
 
-    assert elapsed < 0.2
     assert result.outputs["tool_execution_complete"] is True
     invocation = request.data["invocations"][0]
+    # Keep the deadline assertion on the transport span.  The first
+    # orchestration action in a fresh process may also initialise unrelated
+    # evidence/workflow support modules, which is not handler monopolisation.
+    assert invocation["transport"]["duration_ms"] < 200.0
     assert invocation["status"] == "timeout"
     assert invocation["error_code"] == "tool_timeout"
     assert invocation["effective_payload"]["success"] is False