diff --git a/src/xdist/dsession.py b/src/xdist/dsession.py
index 47e1de7d..2a841e03 100644
--- a/src/xdist/dsession.py
+++ b/src/xdist/dsession.py
@@ -199,7 +199,8 @@ def worker_workerfinished(self, node: WorkerController) -> None:
         workerready before shutdown was triggered.
         """
         self.config.hook.pytest_testnodedown(node=node, error=None)
-        if node.workeroutput["exitstatus"] == 2:  # keyboard-interrupt
+        exitstatus = node.workeroutput["exitstatus"]
+        if exitstatus == 2:  # keyboard-interrupt
             self.shouldstop = f"{node} received keyboard-interrupt"
             self.worker_errordown(node, "keyboard-interrupt")
             return
@@ -214,6 +215,22 @@ def worker_workerfinished(self, node: WorkerController) -> None:
             assert self.sched is not None
             if node in self.sched.nodes:
                 crashitem = self.sched.remove_node(node)
+                # pytest.exit() with a custom exitcode can leave pending tests,
+                # causing crashitem to be non-empty. Handle this gracefully
+                # instead of raising INTERNALERROR.
+                if crashitem:
+                    self.shouldstop = (
+                        f"{node} exited with status {exitstatus}, "
+                        f"pending test: {crashitem}"
+                    )
+                    self.worker_errordown(node, f"exit-{exitstatus}")
+                    return
+                # For normal completion, crashitem should be empty.
+                # This assertion catches unexpected states.
+                # Note: exitstatus 0/1/5 are normal outcomes:
+                #   0: all tests passed
+                #   1: tests failed (but all were run)
+                #   5: no tests collected
                 assert not crashitem, (crashitem, node)
         self._active_nodes.remove(node)
 
diff --git a/testing/acceptance_test.py b/testing/acceptance_test.py
index 1b44985d..9b108985 100644
--- a/testing/acceptance_test.py
+++ b/testing/acceptance_test.py
@@ -1699,3 +1699,28 @@ def test():
     )
     result = pytester.runpytest()
     assert result.ret == 0
+
+
+def test_pytest_exit_nonzero_exitcode(pytester: pytest.Pytester) -> None:
+    """Test that pytest.exit() with non-zero exit code is handled properly.
+
+    Issue #1239: pytest.exit causes internal error when exit code is non-zero.
+    """
+    pytester.makepyfile(
+        """
+        import pytest
+        import time
+
+        @pytest.mark.parametrize('i', range(10))
+        def test_me(i):
+            if i == 5:
+                pytest.exit("Something", 1)
+            time.sleep(0.1)
+        """
+    )
+    result = pytester.runpytest("-n2", "--tb=short")
+    # Should not cause INTERNALERROR, should exit cleanly
+    assert "INTERNALERROR" not in result.stdout.str()
+    # The worker exit status is handled, but xdist raises Interrupted
+    # which causes INTERRUPTED (2) exit status
+    assert result.ret == 2  # INTERRUPTED due to shouldstop handling