Improve observation queue handling and add scene-aware event delivery.

MakeObservation improvements:
- Extract ObservationQueue into standalone class with thread-safe operations
- Add external_queue parameter to share queue across multiple GMs
- Add allow_llm_fallback parameter to control LLM observation generation
- Simplify internal implementation by always using ObservationQueue

SendEventToRelevantPlayers improvements:
- Add player_filter callback parameter to filter which players receive events
- Decouples the component from scene-specific logic; callers provide
  any filtering function (e.g., scene_tracker.get_participants)

Prefab updates (dialogic_and_dramaturgic, game_theoretic_and_dramaturgic,
physically_situated_and_dramaturgic):
- Add external_queue parameter for cross-GM observation persistence
- Use player_filter callback for scene-aware event delivery

PiperOrigin-RevId: 863371470
Change-Id: Ida87d5a917cf2316ebac66d989d9bc49b73459cf

Author: vezhnick

concordia/components/game_master/event_resolution.py

@@ -353,6 +353,7 @@ def __init__(
       player_names: Sequence[str],
       make_observation_component_key: str,
       components: Sequence[str] = (),
+      player_filter: Callable[[], Sequence[str]] | None = None,
       pre_act_label: str = DEFAULT_SEND_PRE_ACT_VALUES_TO_PLAYERS_PRE_ACT_LABEL,
   ):
     """Initializes a component that sends component pre-act values to players.
@@ -363,11 +364,15 @@ def __init__(
 
     Args:
       model: The language model to use for the component.
-      player_names: Names of players.
+      player_names: Names of players (used if no player_filter is supplied).
       make_observation_component_key: the key for a MakeObservation component to
         add the pre-act values to the queue of events to observe.
       components: Keys of components to condition whether the event is relevant.
         If empty, all events are relevant.
+      player_filter: Optional callback that returns the list of player names
+        that should receive observations. When provided, this function is called
+        during post_act to determine which players get the event. This can be
+        used to filter by scene participants or any other criteria.
       pre_act_label: Prefix to add to the output of the component when called in
         `pre_act`.
 
@@ -380,6 +385,7 @@ def __init__(
     self._player_names = player_names
     self._pre_act_label = pre_act_label
     self._components = components
+    self._player_filter = player_filter
     self._queue = {}
     self._last_action_spec = None
     self._optional_make_observation_component_key = (
@@ -437,7 +443,12 @@ def post_act(
       prompt = interactive_document.InteractiveDocument(self._model)
       proceed = True
 
-      for active_entity_name in self._player_names:
+      # Use player_filter if provided, otherwise all players
+      relevant_players = self._player_names
+      if self._player_filter:
+        relevant_players = self._player_filter()
+
+      for active_entity_name in relevant_players:
         if self._components:
           component_states = '\n'.join(
               [self._component_pre_act_display(key) for key in self._components]

concordia/components/game_master/make_observation.py

@@ -17,6 +17,7 @@
 from collections.abc import Sequence
 import copy
 import threading
+from typing import Any
 
 from concordia.components.agent import action_spec_ignored
 from concordia.document import interactive_document
@@ -37,6 +38,50 @@
 )
 
 
+class ObservationQueue:
+  """A shared queue for observations that can be used across multiple GMs.
+
+  This allows observations queued in one GM to be delivered by another GM,
+  preventing observation loss during GM transitions.
+  """
+
+  def __init__(self):
+    self._queue = {}
+    self._lock = threading.Lock()
+
+  def add(self, entity_name: str, event: str, player_names: Sequence[str]):
+    """Adds an event to the queue for the given entity."""
+    with self._lock:
+      if entity_name.lower().strip() == 'all':
+        for player in player_names:
+          if player not in self._queue:
+            self._queue[player] = []
+          self._queue[player].append(event)
+      else:
+        if entity_name not in self._queue:
+          self._queue[entity_name] = []
+        self._queue[entity_name].append(event)
+
+  def get_and_clear(self, entity_name: str) -> list[str]:
+    """Gets and clears the queue for the given entity."""
+    with self._lock:
+      if entity_name in self._queue and self._queue[entity_name]:
+        events = self._queue[entity_name]
+        self._queue[entity_name] = []
+        return events
+      return []
+
+  def get_all(self) -> dict[str, list[str]]:
+    """Returns a deep copy of the entire queue state."""
+    with self._lock:
+      return copy.deepcopy(self._queue)
+
+  def set_all(self, queue_state: Any):
+    """Sets the entire queue state from a deep copy."""
+    with self._lock:
+      self._queue = copy.deepcopy(queue_state)
+
+
 class MakeObservation(entity_component.ContextComponent,
                       entity_component.ComponentWithLogging):
   """A component that generates observations to send to players."""
@@ -49,14 +94,16 @@ def __init__(
       call_to_make_observation: str = DEFAULT_CALL_TO_MAKE_OBSERVATION,
       reformat_observations_in_specified_style: str = '',
       pre_act_label: str = DEFAULT_MAKE_OBSERVATION_PRE_ACT_LABEL,
+      external_queue: ObservationQueue | None = None,
+      allow_llm_fallback: bool = True,
   ):
     """Initializes the component.
 
     Args:
       model: The language model to use for the component.
       player_names: Names of players.
       components: Keys of components to condition the observation on.
-      call_to_make_observation: The call to action to make the observation. 
+      call_to_make_observation: The call to action to make the observation.
         Needed to extract the name of the active entity.
       reformat_observations_in_specified_style: If non-empty, the component will
         ask the model to reformat the observation to fit the style specified in
@@ -68,6 +115,12 @@ def __init__(
         description"."
       pre_act_label: Prefix to add to the output of the component when called in
         `pre_act`.
+      external_queue: Optional shared ObservationQueue. If provided, this
+        component will use the external queue instead of creating its own. This
+        allows observations to persist across GM transitions.
+      allow_llm_fallback: If True, when the queue is empty, the LLM will be used
+        to generate an observation. If False, an empty observation will be
+        returned when the queue is empty. Defaults to True.
 
     Raises:
       ValueError: If the component order is not None and contains duplicate
@@ -82,9 +135,10 @@ def __init__(
     )
     self._call_to_make_observation = call_to_make_observation
     self._pre_act_label = pre_act_label
-    self._lock = threading.Lock()
+    self._allow_llm_fallback = allow_llm_fallback
 
-    self._queue = {}
+    # Use external queue if provided, otherwise create one internally
+    self._queue = external_queue if external_queue else ObservationQueue()
 
   def get_named_component_pre_act_value(self, component_name: str) -> str:
     """Returns the pre-act value of a named component of the parent entity."""
@@ -147,32 +201,25 @@ def pre_act(
       )
 
       log_entry['Active Entity'] = active_entity_name
-      with self._lock:
-        log_entry['queue'] = copy.deepcopy(self._queue)
-
-        if (
-            active_entity_name in self._queue
-            and self._queue[active_entity_name]
-        ):
-          log_entry['queue_active_entity'] = copy.deepcopy(
-              self._queue[active_entity_name]
-          )
-          result = ''
-          for event in self._queue[active_entity_name]:
-            result += event + '\n\n\n'
 
-          self._queue[active_entity_name] = []
-        else:
-          result = prompt.open_question(
-              question=(
-                  f'What does {active_entity_name} observe now? Never '
-                  'repeat information that was already provided to '
-                  f'{active_entity_name} unless absolutely necessary. Keep '
-                  'the story moving forward.'
-              ),
-              max_tokens=1200,
-              terminators=(),
-          )
+      events = self._queue.get_and_clear(active_entity_name)
+      log_entry['queue'] = self._queue.get_all()
+      if events:
+        log_entry['queue_active_entity'] = events
+        result = '\n\n\n'.join(events) + '\n\n\n'
+      elif self._allow_llm_fallback:
+        result = prompt.open_question(
+            question=(
+                f'What does {active_entity_name} observe now? Never '
+                'repeat information that was already provided to '
+                f'{active_entity_name} unless absolutely necessary. Keep '
+                'the story moving forward.'
+            ),
+            max_tokens=1200,
+            terminators=(),
+        )
+      else:
+        result = ''
 
       if self._reformat_observations_in_specified_style:
         prompt.statement(
@@ -208,23 +255,12 @@ def pre_act(
 
   def add_to_queue(self, entity_name: str, event: str):
     """Adds an event to the queue of events to observe."""
-    with self._lock:
-      if entity_name.lower().strip() == 'all':
-        for player in self._player_names:
-          if player not in self._queue:
-            self._queue[player] = []
-          self._queue[player].append(event)
-      else:
-        if entity_name not in self._queue:
-          self._queue[entity_name] = []
-        self._queue[entity_name].append(event)
+    self._queue.add(entity_name, event, self._player_names)
 
   def get_state(self) -> entity_component.ComponentState:
     """Returns the state of the component."""
-    with self._lock:
-      return {'queue': copy.deepcopy(self._queue)}
+    return {'queue': self._queue.get_all()}
 
   def set_state(self, state: entity_component.ComponentState) -> None:
     """Sets the state of the component."""
-    with self._lock:
-      self._queue = copy.deepcopy(state['queue'])
+    self._queue.set_all(state['queue'])

concordia/prefabs/game_master/dialogic_and_dramaturgic.py

@@ -111,6 +111,10 @@ class GameMaster(prefab_lib.Prefab):
           # in the component order. If not specified, the extra components
           # will be inserted at the end of the component order.
           'extra_components_index': {},
+          # Optional shared ObservationQueue for cross-GM observation
+          # persistence. When provided, observations queued by one GM
+          # persist across GM switches.
+          'external_queue': None,
       }
   )
   entities: Sequence[entity_agent_with_logging.EntityAgentWithLogging] = ()
@@ -142,6 +146,7 @@ def build(
         )
 
     player_names = [entity.name for entity in self.entities]
+    external_queue = self.params.get('external_queue', None)
 
     scenes = self.params.get('scenes', _configure_default_scenes(player_names))
     assert isinstance(scenes, Sequence), 'scenes must be a sequence.'
@@ -187,30 +192,34 @@ def build(
             observation_component_key,
             display_events_key,
         ],
+        external_queue=external_queue,
     )
 
+    scene_tracker = gm_components.scene_tracker.SceneTracker(
+        model=model,
+        scenes=scenes,
+        observation_component_key=(
+            gm_components.make_observation.DEFAULT_MAKE_OBSERVATION_COMPONENT_KEY
+        ),
+    )
+
+    # SendEventToRelevantPlayers handles notifying players about events.
+    # Use scene tracker's get_participants as a filter to limit notifications.
     send_events_to_players_key = (
         gm_components.event_resolution.DEFAULT_SEND_PRE_ACT_VALUES_TO_PLAYERS_PRE_ACT_LABEL
     )
+    scene_tracker_key = (
+        gm_components.next_game_master.DEFAULT_NEXT_GAME_MASTER_COMPONENT_KEY
+    )
     send_events_to_players = (
         gm_components.event_resolution.SendEventToRelevantPlayers(
             model=model,
             player_names=player_names,
             make_observation_component_key=make_observation_key,
+            player_filter=scene_tracker.get_participants,
         )
     )
 
-    scene_tracker_key = (
-        gm_components.next_game_master.DEFAULT_NEXT_GAME_MASTER_COMPONENT_KEY
-    )
-    scene_tracker = gm_components.scene_tracker.SceneTracker(
-        model=model,
-        scenes=scenes,
-        observation_component_key=(
-            gm_components.make_observation.DEFAULT_MAKE_OBSERVATION_COMPONENT_KEY
-        ),
-    )
-
     next_actor_key = gm_components.next_acting.DEFAULT_NEXT_ACTING_COMPONENT_KEY
     next_action_spec_key = (
         gm_components.next_acting.DEFAULT_NEXT_ACTION_SPEC_COMPONENT_KEY
@@ -237,7 +246,6 @@ def build(
     event_resolution = gm_components.event_resolution.EventResolution(
         model=model,
         event_resolution_steps=(identity_without_prefix,),
-        notify_observers=False,
     )
 
     terminator_key = gm_components.terminate.DEFAULT_TERMINATE_COMPONENT_KEY
@@ -252,14 +260,18 @@ def build(
         observation_component_key: observation,
         observation_to_memory_key: observation_to_memory,
         display_events_key: display_events,
-        send_events_to_players_key: send_events_to_players,
         make_observation_key: make_observation,
         memory_component_key: memory,
         scene_tracker_key: scene_tracker,
         next_actor_key: next_actor,
         next_action_spec_key: next_action_spec,
         event_resolution_key: event_resolution,
     }
+    # Only add SendEventToRelevantPlayers when notify_observers is False.
+    if send_events_to_players is not None:
+      components_of_game_master[send_events_to_players_key] = (
+          send_events_to_players
+      )
 
     component_order = list(components_of_game_master.keys())
 

concordia/prefabs/game_master/game_theoretic_and_dramaturgic.py

@@ -129,6 +129,9 @@ class GameMaster(prefab_lib.Prefab):
           'scenes': (),
           'action_to_scores': _default_action_to_scores,
           'scores_to_observation': _default_scores_to_observation,
+          # Optional shared ObservationQueue for cross-GM observation
+          # persistence.
+          'external_queue': None,
       }
   )
   entities: (
@@ -152,6 +155,7 @@ def build(
     name = self.params.get('name', DEFAULT_NAME)
 
     player_names = [entity.name for entity in self.entities]
+    external_queue = self.params.get('external_queue', None)
 
     scenes = self.params.get('scenes', _configure_default_scenes(player_names))
     assert isinstance(scenes, Sequence), 'scenes must be a sequence.'
@@ -196,15 +200,15 @@ def build(
 
     make_observation_key = (
         gm_components.make_observation.DEFAULT_MAKE_OBSERVATION_COMPONENT_KEY)
-    make_observation = (
-        gm_components.make_observation.MakeObservation(
-            model=model,
-            player_names=player_names,
-            components=[
-                observation_component_key,
-                display_events_key,
-            ],
-        )
+    make_observation = gm_components.make_observation.MakeObservation(
+        model=model,
+        player_names=player_names,
+        components=[
+            observation_component_key,
+            display_events_key,
+        ],
+        external_queue=external_queue,
+        allow_llm_fallback=False,
     )
 
     scene_tracker_key = (