diff --git a/chess/engine.py b/chess/engine.py
index c66bc0c45..1d96d8c11 100644
--- a/chess/engine.py
+++ b/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:
diff --git a/test.py b/test.py
index a481c2adb..8da2ca4dd 100755
--- a/test.py
+++ b/test.py
@@ -3041,6 +3041,46 @@ 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_uci_option_map_equality(self):
         a = chess.engine.UciOptionMap()
         b = chess.engine.UciOptionMap()