Ignore engine lines received after the command is finished

chess/engine.py

@@ -1222,11 +1222,16 @@ def _handle_exception(self, exc: Exception) -> None:
             self._dispatch_finished()
 
     def set_finished(self) -> None:
-        assert self.state in [CommandState.ACTIVE, CommandState.CANCELLING], self.state
+        # A queued command may be finished while still NEW (replaced by a
+        # newer command in Protocol.communicate, or cancelled before it
+        # could start). Nothing was sent to the engine in that case, so
+        # there is no active phase to unwind.
+        assert self.state in [CommandState.NEW, CommandState.ACTIVE, CommandState.CANCELLING], self.state
         if not self.result.done():
             self.result.set_exception(EngineError(f"engine command finished before returning result: {self!r}"))
         self.state = CommandState.DONE
-        self.finished.set_result(None)
+        if not self.finished.done():
+            self.finished.set_result(None)
         self._dispatch_finished()
 
     def _cancel(self) -> None:
@@ -1245,6 +1250,11 @@ def _start(self) -> None:
             self._handle_exception(err)
 
     def _line_received(self, line: str) -> None:
+        if self.state == CommandState.DONE:
+            # Some engines send trailing output (e.g. info lines around
+            # bestmove). The command is already finished, so there is
+            # nothing left to feed the line into - ignore it.
+            return
         assert self.state in [CommandState.ACTIVE, CommandState.CANCELLING], self.state
         try:
             self.line_received(line)

test.py

@@ -3041,6 +3041,59 @@ def test_utf8_bom(self):
 @unittest.skipIf(sys.platform == "win32" and (3, 8, 0) <= sys.version_info < (3, 8, 1), "https://bugs.python.org/issue34679")
 class EngineTestCase(unittest.TestCase):
 
+    @staticmethod
+    def _with_event_loop(fn):
+        # BaseCommand builds asyncio Futures in __init__; Python 3.14
+        # removed implicit event-loop creation, so provide one explicitly.
+        loop = asyncio.new_event_loop()
+        asyncio.set_event_loop(loop)
+        try:
+            fn()
+        finally:
+            asyncio.set_event_loop(None)
+            loop.close()
+
+    def test_command_set_finished_while_queued_cancelled(self):
+        # Regression test for issue #1116: a queued command whose task was
+        # cancelled before it could start (result and finished futures
+        # already cancelled by Protocol.communicate) must finish cleanly
+        # instead of raising AssertionError on CommandState.NEW.
+        def scenario():
+            cmd = chess.engine.BaseCommand(None)
+            cmd.result.cancel()
+            cmd.finished.cancel()
+            cmd.set_finished()
+            self.assertEqual(cmd.state, chess.engine.CommandState.DONE)
+
+        self._with_event_loop(scenario)
+
+    def test_command_set_finished_while_queued_pending(self):
+        # Replacing a still-queued command that was never awaited should
+        # surface an EngineError on its result rather than crash.
+        def scenario():
+            cmd = chess.engine.BaseCommand(None)
+            cmd.set_finished()
+            self.assertEqual(cmd.state, chess.engine.CommandState.DONE)
+            self.assertTrue(cmd.finished.done())
+            self.assertIsInstance(
+                cmd.result.exception(), chess.engine.EngineError
+            )
+
+        self._with_event_loop(scenario)
+
+    def test_command_line_received_after_done_is_ignored(self):
+        # Regression test for issue #1161: engines may send trailing
+        # output (e.g. info lines around bestmove) after the finishing
+        # line was processed. Feeding a line to a DONE command used to
+        # raise AssertionError; it should be ignored instead.
+        def scenario():
+            cmd = chess.engine.BaseCommand(None)
+            cmd.set_finished()
+            self.assertEqual(cmd.state, chess.engine.CommandState.DONE)
+            cmd._line_received("info depth 1 nodes 42")
+
+        self._with_event_loop(scenario)
+
     def test_uci_option_map_equality(self):
         a = chess.engine.UciOptionMap()
         b = chess.engine.UciOptionMap()