gh-127495: Append to history file after every statement in PyREPL (GH-132294)

Lib/_pyrepl/readline.py

@@ -90,6 +90,7 @@ __all__ = [
     # "set_pre_input_hook",
     "set_startup_hook",
     "write_history_file",
+    "append_history_file",
     # ---- multiline extensions ----
     "multiline_input",
 ]
@@ -453,6 +454,7 @@ class _ReadlineWrapper:
                         del buffer[:]
                     if line:
                         history.append(line)
+        self.set_history_length(self.get_current_history_length())
 
     def write_history_file(self, filename: str = gethistoryfile()) -> None:
         maxlength = self.saved_history_length
@@ -464,6 +466,19 @@ class _ReadlineWrapper:
                 entry = entry.replace("\n", "\r\n")  # multiline history support
                 f.write(entry + "\n")
 
+    def append_history_file(self, filename: str = gethistoryfile()) -> None:
+        reader = self.get_reader()
+        saved_length = self.get_history_length()
+        length = self.get_current_history_length() - saved_length
+        history = reader.get_trimmed_history(length)
+        f = open(os.path.expanduser(filename), "a",
+                 encoding="utf-8", newline="\n")
+        with f:
+            for entry in history:
+                entry = entry.replace("\n", "\r\n")  # multiline history support
+                f.write(entry + "\n")
+        self.set_history_length(saved_length + length)
+
     def clear_history(self) -> None:
         del self.get_reader().history[:]
 
@@ -533,6 +548,7 @@ set_history_length = _wrapper.set_history_length
 get_current_history_length = _wrapper.get_current_history_length
 read_history_file = _wrapper.read_history_file
 write_history_file = _wrapper.write_history_file
+append_history_file = _wrapper.append_history_file
 clear_history = _wrapper.clear_history
 get_history_item = _wrapper.get_history_item
 remove_history_item = _wrapper.remove_history_item

Lib/_pyrepl/simple_interact.py

@@ -30,8 +30,9 @@ import functools
 import os
 import sys
 import code
+import warnings
 
-from .readline import _get_reader, multiline_input
+from .readline import _get_reader, multiline_input, append_history_file
 
 
 _error: tuple[type[Exception], ...] | type[Exception]
@@ -144,6 +145,10 @@ def run_multiline_interactive_console(
             input_name = f"<python-input-{input_n}>"
             more = console.push(_strip_final_indent(statement), filename=input_name, _symbol="single")  # type: ignore[call-arg]
             assert not more
+            try:
+                append_history_file()
+            except (FileNotFoundError, PermissionError, OSError) as e:
+                warnings.warn(f"failed to open the history file for writing: {e}")
             input_n += 1
         except KeyboardInterrupt:
             r = _get_reader()

Lib/test/test_pyrepl/test_pyrepl.py

@@ -112,6 +112,7 @@ class ReplTestCase(TestCase):
         else:
             os.close(master_fd)
             process.kill()
+            process.wait(timeout=SHORT_TIMEOUT)
             self.fail(f"Timeout while waiting for output, got: {''.join(output)}")
 
         os.close(master_fd)
@@ -1564,6 +1565,27 @@ class TestMain(ReplTestCase):
             self.assertEqual(exit_code, 0)
             self.assertNotIn("\\040", pathlib.Path(hfile.name).read_text())
 
+    def test_history_survive_crash(self):
+        env = os.environ.copy()
+        commands = "1\nexit()\n"
+        output, exit_code = self.run_repl(commands, env=env)
+        if "can't use pyrepl" in output:
+            self.skipTest("pyrepl not available")
+
+        with tempfile.NamedTemporaryFile() as hfile:
+            env["PYTHON_HISTORY"] = hfile.name
+            commands = "spam\nimport time\ntime.sleep(1000)\npreved\n"
+            try:
+                self.run_repl(commands, env=env)
+            except AssertionError:
+                pass
+
+            history = pathlib.Path(hfile.name).read_text()
+            self.assertIn("spam", history)
+            self.assertIn("time", history)
+            self.assertNotIn("sleep", history)
+            self.assertNotIn("preved", history)
+
     def test_keyboard_interrupt_after_isearch(self):
         output, exit_code = self.run_repl(["\x12", "\x03", "exit"])
         self.assertEqual(exit_code, 0)

Misc/NEWS.d/next/Library/2025-04-08-14-50-39.gh-issue-127495.Q0V0bS.rst

@@ -0,0 +1,3 @@
+In PyREPL, append a new entry to the ``PYTHON_HISTORY`` file *after* every
+statement.  This should preserve command-line history after interpreter is
+terminated.  Patch by Sergey B Kirpichev.