Implement force action priority

docs/source/api.rst

@@ -502,12 +502,6 @@ The API module handles various error conditions:
       except Exception as e:
           print(f"Unexpected error: {e}")
 
-Deprecated Methods
-------------------
-
-.. deprecated:: 2.1.0
-   ``read_raw_message`` has been deprecated. Use ``read_raw_server_message`` from the parent class instead.
-
 Dependencies
 ------------
 

src/neuro_api/api.py

@@ -23,14 +23,14 @@
 
 __title__ = "api"
 __author__ = "CoolCat467"
-__version__ = "2.3.0"
+__version__ = "3.0.0"
 __license__ = "GNU Lesser General Public License Version 3"
 
 
 from abc import abstractmethod
 from typing import TYPE_CHECKING, NamedTuple
 
-from neuro_api import _deprecate, command
+from neuro_api import command
 from neuro_api.client import AbstractNeuroAPIClient
 
 if TYPE_CHECKING:
@@ -245,6 +245,7 @@ async def send_force_action(
         query: str,
         action_names: Sequence[str],
         ephemeral_context: bool = False,
+        priority: command.ForcePriority = command.ForcePriority.LOW,
     ) -> None:
         """Force Neuro to execute an action with specific context.
 
@@ -268,6 +269,20 @@ async def send_force_action(
                   be remembered after actions force completion.
                 - If True: Neuro will only remember the context during the
                   actions force.
+            priority (command.ForcePriority):
+                Determines how urgently Neuro should respond to the
+                action force when she is speaking. If Neuro is not
+                speaking, this setting has no effect. The default is
+                `command.ForcePriority.LOW`, which will cause Neuro to
+                wait until she finishes speaking before responding.
+                `command.ForcePriority.MEDIUM` causes her to finish her
+                current utterance sooner. `command.ForcePriority.HIGH`
+                prompts her to process the action force immediately,
+                shortening her utterance and then responding.
+                `command.ForcePriority.CRITICAL` will interrupt her
+                speech and make her respond at once. Use
+                `command.ForcePriority.CRITICAL` with caution, as it may
+                lead to abrupt and potentially jarring interruptions.
 
         Raises:
             ValueError: If any specified action name is not currently
@@ -294,6 +309,7 @@ async def send_force_action(
                 query,
                 action_names,
                 ephemeral_context,
+                priority,
             ),
         )
 
@@ -447,13 +463,6 @@ async def handle_immediate_shutdown(self) -> None:
         """
         await self.send_shutdown_ready()
 
-    read_raw_message = _deprecate.deprecated_async_alias(
-        "read_raw_message",
-        AbstractNeuroAPIClient.read_raw_server_message,
-        "2.1.0",
-        issue=None,
-    )
-
     async def read_message(self) -> None:
         """Read message from Neuro websocket.
 

src/neuro_api/command.py

@@ -27,6 +27,7 @@
 
 
 import sys
+from enum import Enum
 from types import GenericAlias, UnionType
 from typing import (
     TYPE_CHECKING,
@@ -42,8 +43,6 @@
 import orjson
 from typing_extensions import NotRequired, is_typeddict
 
-from neuro_api._deprecate import deprecated_alias
-
 if TYPE_CHECKING:
     from collections.abc import Mapping, Sequence
 
@@ -337,12 +336,22 @@ def actions_unregister_command(
     )
 
 
+class ForcePriority(str, Enum):
+    """`actions/force` `priority` field values."""
+
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
 def actions_force_command(
     game: str,
     state: str,
     query: str,
     action_names: Sequence[str],
     ephemeral_context: bool = False,
+    priority: ForcePriority = ForcePriority.LOW,
 ) -> bytes:
     """Return formatted actions/force command.
 
@@ -374,6 +383,19 @@ def actions_force_command(
             remembered by Neuro after the actions force is completed.
             If True, Neuro will only remember it for the duration of
             the actions force. Defaults to False.
+        priority (ForcePriority):
+            Determines how urgently Neuro should respond to the action
+            force when she is speaking. If Neuro is not speaking, this
+            setting has no effect. The default is `ForcePriority.LOW`,
+            which will cause Neuro to wait until she finishes speaking
+            before responding. `ForcePriority.MEDIUM` causes her to
+            finish her current utterance sooner. `ForcePriority.HIGH`
+            prompts her to process the action force immediately,
+            shortening her utterance and then responding.
+            `ForcePriority.CRITICAL` will interrupt her speech and make
+            her respond at once. Use `ForcePriority.CRITICAL` with
+            caution, as it may lead to abrupt and potentially jarring
+            interruptions.
 
     Returns:
         bytes: A formatted command to force actions for the specified game.
@@ -388,6 +410,7 @@ def actions_force_command(
         "state": state,
         "query": query,
         "action_names": list(action_names),
+        "priority": priority.value,
     }
     if ephemeral_context:
         payload["ephemeral_context"] = True
@@ -675,17 +698,6 @@ def convert_parameterized_generic_union_items(
     return generic
 
 
-# Old name with a typo, TODO remove
-convert_parameterized_generic_union_items = (  # spellcheck: ignore
-    deprecated_alias(
-        "convert_parameterized_generic_union_items",  # spellcheck: ignore
-        convert_parameterized_generic_union_items,
-        "2.3.0",
-        issue=None,
-    )
-)
-
-
 def convert_parameterized_generic(
     generic: GenericAlias | UnionType | T,
 ) -> T | type | tuple[type, ...]:

src/neuro_api/server.py

@@ -30,7 +30,7 @@
 import weakref
 from abc import ABCMeta, abstractmethod
 from functools import partial
-from typing import TYPE_CHECKING, Any, TypedDict, cast
+from typing import TYPE_CHECKING, Any, Literal, TypedDict, cast
 from uuid import UUID
 
 import trio
@@ -44,7 +44,7 @@
 from neuro_api import command, json_schema_types
 from neuro_api.api import __version__
 from neuro_api.client import AbstractNeuroAPIClient
-from neuro_api.command import Action
+from neuro_api.command import Action, ForcePriority
 
 if TYPE_CHECKING:
     from collections.abc import Awaitable, Callable
@@ -118,13 +118,17 @@ class ForceActionsData(TypedDict):
         ephemeral_context (NotRequired[bool], optional):
             Flag for ephemeral context. Defaults to None.
         action_names (list[str]): List of action names to force.
+        priority (ForcePriority):
+            Determines how urgently Neuro should respond to the action
+            force when she is speaking.
 
     """
 
     state: NotRequired[str]
     query: str
     ephemeral_context: NotRequired[bool]
     action_names: list[str]
+    priority: NotRequired[Literal["low" | "medium" | "high" | "critical"]]
 
 
 class ActionResultData(TypedDict):
@@ -428,6 +432,7 @@ async def handle_actions_force(
         query: str,
         ephemeral_context: bool,
         action_names: list[str],
+        priority: ForcePriority,
     ) -> None:
         """Force Neuro to choose and execute actions from a specified list.
 
@@ -449,6 +454,19 @@ async def handle_actions_force(
                   force operation
             action_names (list[str]): Names of actions Neuro MUST choose
                 from when executing the force command.
+            priority (ForcePriority):
+                Determines how urgently Neuro should respond to the
+                action force when she is speaking. If Neuro is not
+                speaking, this setting has no effect. The default is
+                `ForcePriority.LOW`, which will cause Neuro to wait
+                until she finishes speaking before responding.
+                `ForcePriority.MEDIUM` causes her to finish her current
+                utterance sooner. `ForcePriority.HIGH` prompts her to
+                process the action force immediately, shortening her
+                utterance and then responding. `ForcePriority.CRITICAL`
+                will interrupt her speech and make her respond at once.
+                Use `ForcePriority.CRITICAL` with caution, as it may
+                lead to abrupt and potentially jarring interruptions.
 
         """
 
@@ -633,6 +651,7 @@ async def read_message(self) -> None:
                 force_actions_data["query"],
                 force_actions_data.get("ephemeral_context", False),
                 action_names,
+                ForcePriority(force_actions_data.get("priority", "low")),
             )
         elif command_type == "action/result":
             if data is None:
@@ -1040,6 +1059,7 @@ async def perform_actions_force(
         query: str,
         ephemeral_context: bool,
         action_names: list[str],
+        priority: ForcePriority,
     ) -> None:
         """Execute a forced action sequence with automatic retry on failure.
 
@@ -1142,6 +1162,7 @@ async def handle_actions_force(
         query: str,
         ephemeral_context: bool,
         action_names: list[str],
+        priority: ForcePriority,
     ) -> None:
         """Process a force actions command from the game client.
 
@@ -1162,6 +1183,19 @@ async def handle_actions_force(
                 - True: State and query are only used during this operation
             action_names (list[str]): List of action names that Neuro is
                 restricted to choose from during this force sequence.
+            priority (ForcePriority):
+                Determines how urgently Neuro should respond to the
+                action force when she is speaking. If Neuro is not
+                speaking, this setting has no effect. The default is
+                `ForcePriority.LOW`, which will cause Neuro to wait
+                until she finishes speaking before responding.
+                `ForcePriority.MEDIUM` causes her to finish her current
+                utterance sooner. `ForcePriority.HIGH` prompts her to
+                process the action force immediately, shortening her
+                utterance and then responding. `ForcePriority.CRITICAL`
+                will interrupt her speech and make her respond at once.
+                Use `ForcePriority.CRITICAL` with caution, as it may
+                lead to abrupt and potentially jarring interruptions.
 
         Behavior:
             1. Validates the game title matches the current session
@@ -1189,6 +1223,7 @@ async def handle_actions_force(
             query,
             ephemeral_context,
             action_names,
+            priority,
         )
         await self.submit_call_async_soon(do_actions_force)
 

tests/test_command.py

@@ -8,6 +8,7 @@
 
 from neuro_api.command import (
     Action,
+    ForcePriority,
     IncomingActionMessageSchema,
     action_command,
     actions_force_command,
@@ -157,7 +158,7 @@ def test_actions_force_command() -> None:
     query = "Please take your turn."
     action_names = ["test_action"]
 
-    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"]}}'
+    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"],"priority":"low"}}'
     assert (
         actions_force_command(game, state, query, action_names)
         == expected_output
@@ -170,13 +171,94 @@ def test_actions_force_command_ephemeral() -> None:
     query = "Please take your turn."
     action_names = ["test_action"]
 
-    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"],"ephemeral_context":true}}'
+    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"],"priority":"low","ephemeral_context":true}}'
     assert (
         actions_force_command(game, state, query, action_names, True)
         == expected_output
     )
 
 
+def test_actions_force_command_priority_medium() -> None:
+    game = "Test Game"
+    state = "Game is in progress."
+    query = "Please take your turn."
+    action_names = ["test_action"]
+    priority = ForcePriority.MEDIUM
+
+    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"],"priority":"medium"}}'
+    assert (
+        actions_force_command(
+            game,
+            state,
+            query,
+            action_names,
+            priority=priority,
+        )
+        == expected_output
+    )
+
+
+def test_actions_force_command_priority_high() -> None:
+    game = "Test Game"
+    state = "Game is in progress."
+    query = "Please take your turn."
+    action_names = ["test_action"]
+    priority = ForcePriority.HIGH
+
+    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"],"priority":"high"}}'
+    assert (
+        actions_force_command(
+            game,
+            state,
+            query,
+            action_names,
+            priority=priority,
+        )
+        == expected_output
+    )
+
+
+def test_actions_force_command_priority_critical() -> None:
+    game = "Test Game"
+    state = "Game is in progress."
+    query = "Please take your turn."
+    action_names = ["test_action"]
+    priority = ForcePriority.CRITICAL
+
+    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"],"priority":"critical"}}'
+    assert (
+        actions_force_command(
+            game,
+            state,
+            query,
+            action_names,
+            priority=priority,
+        )
+        == expected_output
+    )
+
+
+def test_actions_force_command_priority_explicit_low() -> None:
+    game = "Test Game"
+    state = "Game is in progress."
+    query = "Please take your turn."
+    action_names = ["test_action"]
+    priority = ForcePriority.LOW
+
+    expected_output = b'{"command":"actions/force","game":"Test Game","data":{"state":"Game is in progress.","query":"Please take your turn.","action_names":["test_action"],"priority":"low"}}'
+    assert (
+        actions_force_command(
+            game,
+            state,
+            query,
+            action_names,
+            False,
+            priority,
+        )
+        == expected_output
+    )
+
+
 def test_actions_result_command() -> None:
     game = "Test Game"
     id_ = "12345"