@@ -32,6 +32,8 @@
 from pydantic import BaseModel, Field, TypeAdapter
 from uuid6 import uuid7
 
+from airflow import settings
+from airflow.configuration import conf
 from airflow.dag_processing.importer_routing import get_claiming_coordinator
 from airflow.dag_processing.processor import (
     BaseDagFileProcessorProcess,
@@ -78,12 +80,22 @@ class StartLangSDKRuntime(BaseModel):
 
 
 class LangSDKRuntimeSchemaVersion(BaseModel):
-    """The supervisor schema version of the runtime the parse child is about to exec."""
+    """The schema version and the import timeout of the runtime the parse child is about to exec."""
 
     schema_version: str | None
+    import_timeout: float | None = None
+    """Seconds from the start of the parse; ``None`` means no timeout."""
     type: Literal["LangSDKRuntimeSchemaVersion"] = "LangSDKRuntimeSchemaVersion"
 
 
+def _get_import_timeout(path: str) -> float | None:
+    """Return the ``get_dagbag_import_timeout`` policy's timeout for *path*; ``None`` means none."""
+    timeout = settings.get_dagbag_import_timeout(path)
+    if not isinstance(timeout, (int, float)):
+        raise TypeError(f"Value ({timeout}) from get_dagbag_import_timeout must be int or float")
+    return timeout if timeout > 0 else None
+
+
 def _start_runtime_entrypoint() -> None:
     """Exec the runtime that parses the file named by the start request, or report why it cannot start."""
     os.environ["_AIRFLOW_PROCESS_CONTEXT"] = "client"
@@ -101,9 +113,11 @@ def _start_runtime_entrypoint() -> None:
         raise RuntimeError(f"Required first message to be a StartLangSDKRuntime, it was {msg}")
 
     def report_schema_version(schema_version: str | None) -> None:
-        comms.send(LangSDKRuntimeSchemaVersion(schema_version=schema_version))
+        comms.send(LangSDKRuntimeSchemaVersion(schema_version=schema_version, import_timeout=import_timeout))
 
     try:
+        # The policy is user code: it runs in this child, where a failure is only this file's import error.
+        import_timeout = _get_import_timeout(msg.file)
         coordinator = get_claiming_coordinator(msg.file, msg.bundle_name)
         if coordinator is None:
             raise RuntimeError(f"No coordinator's Dag importer claims {msg.file}")
@@ -149,6 +163,7 @@ class LangSDKDagFileProcessorProcess(BaseDagFileProcessorProcess):
     _listeners: dict[_Channel, socket]
     _parse_request: DagFileParseRequest
     _runtime_schema_version: str | None = attrs.field(default=None, init=False)
+    _import_timeout: float | None = attrs.field(default=None, init=False)
     _schema_version_reported: bool = attrs.field(default=False, init=False)
     _parsing_result_monotonic: float | None = attrs.field(default=None, init=False)
     _unverified_connections: list[tuple[socket, _Channel]] = attrs.field(factory=list, init=False)
@@ -212,17 +227,16 @@ def run(
         bundle_path: Path,
         bundle_name: str,
         dag_file_rel_path: str,
-        timeout: float | None,
         logger: FilteringBoundLogger,
     ) -> DagFileParsingResult:
         """
         Parse *path* outside the Dag processor and wait for the result.
 
-        There is no API client, so each request of the runtime that needs one gets an error.
-
-        :raises TimeoutError: if the parse does not finish within *timeout* seconds. The runtime is
-            killed.
+        There is no API client, so each request of the runtime that needs one gets an error. The file's import
+        timeout bounds the parse, and ``[dag_processor] dag_file_processor_timeout`` until the parse child
+        reports it.
         """
+        processor_timeout = conf.getfloat("dag_processor", "dag_file_processor_timeout")
         with selectors.DefaultSelector() as selector:
             proc = cls.start(
                 id=uuid7(),
@@ -235,14 +249,13 @@ def run(
             )
             try:
                 while not proc.is_ready:
-                    wait = 0.1
-                    if timeout is not None:
-                        if (remaining := proc.start_time + timeout - time.monotonic()) <= 0:
-                            raise TimeoutError(
-                                f"The Lang-SDK runtime did not parse {os.fspath(path)} within {timeout}s"
-                            )
-                        wait = min(wait, remaining)
-                    proc._service_subprocess(max_wait_time=wait)
+                    timeout = proc._import_timeout if proc._schema_version_reported else processor_timeout
+                    if timeout is not None and time.monotonic() - proc.start_time > timeout:
+                        # Unlike is_ready, this does not wait for an exited runtime's leftover processes,
+                        # which can hold its sockets open. close() closes them.
+                        proc._time_out(timeout)
+                        break
+                    proc._service_subprocess(max_wait_time=0.1)
             except BaseException:
                 proc._kill_runtime()
                 raise
@@ -376,6 +389,7 @@ def _handle_runtime_schema_version(
             self._reject_request(msg, log, req_id)
             return ResponseSent.ALREADY_SENT
         self._runtime_schema_version = msg.schema_version
+        self._import_timeout = msg.import_timeout
         self._schema_version_reported = True
         return None, {}
 
@@ -438,6 +452,13 @@ def is_ready(self) -> bool:
         ):
             self.process_log.warning("The Lang-SDK runtime did not exit after its parse result; killing it")
             self._kill_runtime()
+        if (
+            self._import_timeout is not None
+            and self.parsing_result is None
+            and self._exit_code is None
+            and time.monotonic() - self.start_time > self._import_timeout
+        ):
+            self._time_out(self._import_timeout)
         if self._check_subprocess_exit() is None:
             return False
         self._close_listeners()
@@ -449,6 +470,13 @@ def is_ready(self) -> bool:
             )
         return True
 
+    def _time_out(self, timeout: float) -> None:
+        if self.parsing_result is None:
+            self._set_import_error(
+                f"The Lang-SDK runtime did not parse {self._parse_request.file} within {timeout}s"
+            )
+        self._kill_runtime()
+
     def _kill_runtime(self) -> None:
         """Kill the runtime and wait for it, without servicing its sockets, whose handler may have failed."""
         if self._exit_code is not None:
 
@@ -36,6 +36,7 @@
 from airflow.dag_processing.lang_sdk_processor import (
     LangSDKDagFileProcessorProcess,
     LangSDKRuntimeSchemaVersion,
+    _get_import_timeout,
 )
 from airflow.dag_processing.processor import DagFileParseRequest, DagFileParsingResult
 from airflow.sdk import DAG, BaseOperator
@@ -332,6 +333,27 @@ def reply(request, comms):
         assert proc._exit_code == -signal.SIGKILL
         assert "The Lang-SDK runtime did not exit after its parse result; killing it" in cap_structlog
 
+    @pytest.mark.parametrize(
+        ("policy", "error"),
+        [
+            pytest.param(
+                {"side_effect": RuntimeError("policy bug")}, "RuntimeError: policy bug", id="raises"
+            ),
+            pytest.param(
+                {"return_value": "30"},
+                "TypeError: Value (30) from get_dagbag_import_timeout must be int or float",
+                id="not-a-number",
+            ),
+        ],
+    )
+    def test_a_failing_import_timeout_policy_is_an_import_error(self, parse, policy, error):
+        with patch("airflow.settings.get_dagbag_import_timeout", autospec=True, **policy):
+            proc = parse()
+
+        assert proc.parsing_result.import_errors == {
+            "dag.native": f"Cannot start the Lang-SDK runtime: {error}"
+        }
+
     @pytest.mark.skipif(not Path("/proc/self/fd").is_dir(), reason="reads /proc")
     @pytest.mark.parametrize("use_exec", [False, True], ids=["fork", "spawn"])
     def test_the_runtime_inherits_only_its_standard_streams(self, monkeypatch, tmp_path, use_exec):
@@ -357,13 +379,12 @@ def test_the_runtime_inherits_only_its_standard_streams(self, monkeypatch, tmp_p
 
 class TestRun:
     @staticmethod
-    def _run(tmp_path, *, timeout: float | None = 30) -> DagFileParsingResult:
+    def _run(tmp_path, **spec) -> DagFileParsingResult:
         return LangSDKDagFileProcessorProcess.run(
-            path=write_native_file(tmp_path / "dag.native"),
+            path=write_native_file(tmp_path / "dag.native", **spec),
             bundle_path=tmp_path,
             bundle_name="testing",
             dag_file_rel_path="dag.native",
-            timeout=timeout,
             logger=structlog.get_logger(),
         )
 
@@ -398,9 +419,9 @@ def reply(request, comms):
         mock_mask_secret.assert_called_once_with("native-secret", "native_conn")
 
     @pytest.mark.parametrize("connected", [False, True], ids=["before-connecting", "after-connecting"])
-    def test_a_parse_past_its_timeout_is_killed(self, tmp_path, connected):
+    @patch("airflow.settings.get_dagbag_import_timeout", autospec=True, return_value=1)
+    def test_a_parse_past_the_import_timeout_is_killed(self, mock_timeout, tmp_path, connected):
         # A runtime that never connects leaves both listeners open when it is killed.
-        write_native_file(tmp_path / "dag.native", argv=["/bin/sh", "-c", "exec sleep 60"])
         fds_before = _get_open_fds()
 
         with (
@@ -418,22 +439,61 @@ def test_a_parse_past_its_timeout_is_killed(self, tmp_path, connected):
                 autospec=True,
                 side_effect=LangSDKDagFileProcessorProcess.close,
             ) as mock_close,
-            pytest.raises(TimeoutError, match=r"did not parse .*dag\.native within 1s"),
         ):
-            LangSDKDagFileProcessorProcess.run(
-                path=tmp_path / "dag.native",
-                bundle_path=tmp_path,
-                bundle_name="testing",
-                dag_file_rel_path="dag.native",
-                timeout=1,
-                logger=structlog.get_logger(),
-            )
+            result = self._run(tmp_path, argv=["/bin/sh", "-c", "exec sleep 60"])
 
+        assert result.import_errors == {
+            "dag.native": f"The Lang-SDK runtime did not parse {tmp_path / 'dag.native'} within 1.0s"
+        }
         [proc] = [c.args[0] for c in mock_close.call_args_list]
         assert proc._exit_code == -9
         assert not proc._open_sockets
         assert _get_open_fds() <= fds_before
 
+    @patch("airflow.settings.get_dagbag_import_timeout", autospec=True, return_value=1)
+    def test_the_import_timeout_holds_after_the_runtime_exits(self, mock_timeout, tmp_path):
+        with patch.object(
+            LangSDKDagFileProcessorProcess,
+            "close",
+            autospec=True,
+            side_effect=LangSDKDagFileProcessorProcess.close,
+        ) as mock_close:
+            # The runtime exits, and the process it leaves behind keeps its output open.
+            result = self._run(tmp_path, argv=["/bin/sh", "-c", "sleep 30 & exit 0"])
+        [proc] = [c.args[0] for c in mock_close.call_args_list]
+        os.killpg(proc.pid, signal.SIGKILL)
+
+        assert result.import_errors == {
+            "dag.native": f"The Lang-SDK runtime did not parse {tmp_path / 'dag.native'} within 1.0s"
+        }
+        assert proc._exit_code == 0
+        assert not proc._open_sockets
+
+    @conf_vars({("dag_processor", "dag_file_processor_timeout"): "1"})
+    @patch.object(
+        FakeCoordinator,
+        "_build_parse_dag_command",
+        autospec=True,
+        side_effect=lambda self, *, path: time.sleep(60),
+    )
+    def test_the_dag_file_processor_timeout_applies_until_the_import_timeout_is_reported(
+        self, mock_build_parse_dag_command, tmp_path
+    ):
+        result = self._run(tmp_path)
+
+        assert result.import_errors == {
+            "dag.native": f"The Lang-SDK runtime did not parse {tmp_path / 'dag.native'} within 1.0s"
+        }
+
+
+@pytest.mark.parametrize(("configured", "expected"), [(30, 30), (0.5, 0.5), (0, None), (-1, None)])
+@patch("airflow.settings.get_dagbag_import_timeout", autospec=True)
+def test_only_a_positive_import_timeout_applies(mock_timeout, configured, expected):
+    mock_timeout.return_value = configured
+
+    assert _get_import_timeout("/b/dag.native") == expected
+    mock_timeout.assert_called_once_with("/b/dag.native")
+
 
 def _make_process(**kwargs) -> LangSDKDagFileProcessorProcess:
     return LangSDKDagFileProcessorProcess(
 
@@ -1768,6 +1768,54 @@ def reply(request, comms):
         assert import_error.stacktrace.startswith("The Lang-SDK runtime sent an invalid frame: ")
         assert manager.selector.get_map() == {}
 
+    @mock.patch("airflow.settings.get_dagbag_import_timeout", autospec=True, return_value=1)
+    @mock.patch.object(FakeCoordinator, "parse_dag", autospec=True)
+    def test_a_coordinator_file_past_the_import_timeout_is_an_import_error(
+        self, mock_parse_dag, mock_timeout, tmp_path, configure_testing_dag_bundle
+    ):
+        mock_parse_dag.side_effect = play_runtime(lambda request, comms: time.sleep(60))
+        slow = write_native_file(tmp_path / "slow.native")
+
+        with fake_coordinator(), configure_testing_dag_bundle(tmp_path):
+            manager = DagFileProcessorManager(max_runs=1, processor_timeout=60)
+            manager.run()
+
+        with create_session() as session:
+            [import_error] = session.scalars(select(ParseImportError)).all()
+
+        assert import_error.filename == "slow.native"
+        assert import_error.stacktrace == f"The Lang-SDK runtime did not parse {slow} within 1.0s"
+        assert manager.selector.get_map() == {}
+
+    @mock.patch("airflow.settings.get_dagbag_import_timeout", autospec=True)
+    def test_a_failing_import_timeout_policy_fails_only_its_file(
+        self, mock_timeout, tmp_path, configure_testing_dag_bundle
+    ):
+        def get_dagbag_import_timeout(dag_file_path):
+            if dag_file_path.endswith(".native"):
+                raise RuntimeError("policy bug")
+            return 30
+
+        mock_timeout.side_effect = get_dagbag_import_timeout
+        write_native_file(tmp_path / "broken.native")
+        (tmp_path / "python_dag.py").write_text(
+            "from airflow.sdk import DAG\nfrom airflow.sdk.bases.operator import BaseOperator\n\n"
+            'with DAG("python_dag", schedule=None):\n    BaseOperator(task_id="task")\n'
+        )
+
+        with fake_coordinator(), configure_testing_dag_bundle(tmp_path):
+            DagFileProcessorManager(max_runs=1, processor_timeout=60).run()
+
+        with create_session() as session:
+            dag_ids = session.scalars(select(SerializedDagModel.dag_id)).all()
+            [import_error] = session.scalars(select(ParseImportError)).all()
+
+        assert dag_ids == ["python_dag"]
+        assert (import_error.filename, import_error.stacktrace) == (
+            "broken.native",
+            "Cannot start the Lang-SDK runtime: RuntimeError: policy bug",
+        )
+
     def test_terminate_orphan_processes_kills_then_closes_processor(self):
         manager = DagFileProcessorManager(max_runs=1)
         processor, _ = self.mock_processor()
 
@@ -23,7 +23,6 @@
 
 import structlog
 
-from airflow.sdk.exceptions import AirflowConfigException
 from airflow.sdk.importers.base import (
     AbstractDagImporter,
     DagImportError,
@@ -45,19 +44,6 @@
 log: FilteringBoundLogger = structlog.get_logger(logger_name="coordinators.dag_importer")
 
 
-def _get_import_timeout(definition: FilesystemDagDefinition) -> float | None:
-    """Return the parse timeout for *definition*, as ``PythonDagImporter`` does; ``None`` means none."""
-    try:
-        from airflow import settings  # noqa: SDK002
-
-        timeout = settings.get_dagbag_import_timeout(repr(definition))
-    except (ImportError, AttributeError):
-        timeout = 30.0
-    if not isinstance(timeout, (int, float)):
-        raise AirflowConfigException(f"Value ({timeout}) from get_dagbag_import_timeout must be int or float")
-    return timeout if timeout > 0 else None
-
-
 class CoordinatorDagImporter(AbstractDagImporter[FilesystemDagDefinition]):
     """
     Import the native Dags of a coordinator's artifacts by running the coordinator's runtime.
@@ -104,23 +90,15 @@ def import_definition(
 
         source_reference = repr(definition)
         bundle_path = bundle.path or definition.path.parent
-        result = DagImportResult(definition=definition)
-        try:
-            parsing_result = LangSDKDagFileProcessorProcess.run(
-                path=definition.path,
-                bundle_path=bundle_path,
-                bundle_name=bundle.name,
-                dag_file_rel_path=definition.get_relative_loc(bundle_path),
-                timeout=_get_import_timeout(definition),
-                logger=log,
-            )
-        except TimeoutError as e:
-            result.errors.append(
-                DagImportError(source_reference=source_reference, message=str(e), error_type="timeout")
-            )
-            return result
-
         relative_loc = definition.get_relative_loc(bundle_path)
+        result = DagImportResult(definition=definition)
+        parsing_result = LangSDKDagFileProcessorProcess.run(
+            path=definition.path,
+            bundle_path=bundle_path,
+            bundle_name=bundle.name,
+            dag_file_rel_path=relative_loc,
+            logger=log,
+        )
         for key, message in (parsing_result.import_errors or {}).items():
             result.errors.append(
                 DagImportError(