Add blocklist to Blackboard 'text_snapshot()' (#13)

* Tests fot Blackboard text_snapshot

Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com>

* Add blocked keys to text_snapshot blackboard

Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com>

* Print with proper method output

Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com>

* Clean comments

Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com>

* Review

Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com>

---------

Signed-off-by: Carlos Ferreira González <carlosferreira@eprosima.com>

Author: cferreiragonz

src/vulcanai/console/console.py

@@ -348,8 +348,9 @@ def worker(user_input: str = "") -> None:
 
                 # Print the backboard state
                 bb_ret = result.get("blackboard", None)
-                bb_ret = str(bb_ret).replace("<", "'").replace(">", "'")
-                self.logger.log_console(f"Output of plan: {bb_ret}")
+                if bb_ret:
+                    bb_ret = bb_ret.text_snapshot().replace("<", "'").replace(">", "'")
+                    self.logger.log_console(f"Output of plan: {bb_ret}")
 
             except KeyboardInterrupt:
                 self.logger.log_msg("<yellow>Exiting...</yellow>")

src/vulcanai/console/utils.py

@@ -216,7 +216,6 @@ def _on_done(task: asyncio.Task) -> None:
         asyncio.get_running_loop()
     except RuntimeError:
         # No loop here → probably ROS thread. Bounce into Textual thread.
-        # `console.app` is your Textual App instance.
         console.app.call_from_thread(_launcher)
     else:
         # We *are* in the loop → just launch directly.
@@ -289,7 +288,6 @@ def _get_suggestions(real_string_list_comp: list[str], string_comp: str) -> tupl
         return most_topic_similar, ret_list
 
     if input_string not in real_string_list:
-        # console.add_line(f"{tool_header_str} {string_name}: \"{input_string}\" does not exists")
         console.logger.log_tool(f'{string_name}: "{input_string}" does not exists', tool_name=tool_name)
 
         # Get the suggestions list sorted by similitud value

src/vulcanai/core/executor.py

@@ -33,15 +33,17 @@
 class Blackboard(dict):
     """Shared memory for passing outputs between steps."""
 
+    _default_blocked_keys = {"main_node", "console"}
+
     def text_snapshot(self, keys: Optional[List[str]] = None) -> str:
         """
         Return a string representing the blackboard.
 
-        :param keys: Optional list of keys to include. If None, include all.
+        :param keys: Optional list of keys to include. If None, include all except default blocked keys.
         :return: A string representation of the blackboard entries.
         """
         snapshot = {}
-        keyset: Set[str] = set(keys) if keys is not None else set(self.keys())
+        keyset: Set[str] = set(keys) if keys is not None else set(self.keys() - self._default_blocked_keys)
         filtered = {k: self.get(k) for k in keyset if k in self}
 
         for k, v in filtered.items():

tests/unittest/test_executor.py

@@ -801,6 +801,50 @@ def test_types_substitution_is_respected(self):
         self.assertEqual(bb["add"]["result"], 8.5)
         self.assertTrue(isinstance(bb["add"]["result"], float))
 
+    def test_default_blocked_keys_in_output(self):
+        """
+        Test that default keys are blocked and not printed.
+        """
+        executor_mod = importlib.import_module("vulcanai.core.executor")
+        blackboard_class = executor_mod.Blackboard
+
+        test_bb = blackboard_class()
+
+        test_bb["speak"] = True
+        test_bb["main_node"] = "This should not be printed"
+        test_bb["console"] = "This should not be printed either"
+        self.assertIn("speak", test_bb)
+        self.assertIn("main_node", test_bb)
+        self.assertIn("console", test_bb)
+        string_output = test_bb.text_snapshot()
+        self.assertIn("speak", string_output)
+        self.assertNotIn("main_node", string_output)
+        self.assertNotIn("console", string_output)
+
+    def test_text_snapshot_select_keys_when_given(self):
+        """
+        Test that the text_snapshot method of the blackboard properly filters by the provided keys.
+        """
+        executor_mod = importlib.import_module("vulcanai.core.executor")
+        blackboard_class = executor_mod.Blackboard
+
+        test_bb = blackboard_class()
+
+        test_bb["speak"] = True
+        test_bb["another_potential_tool"] = "Random text"
+        test_bb["other_tool"] = "Whatever output"
+        self.assertIn("speak", test_bb)
+        self.assertIn("another_potential_tool", test_bb)
+        self.assertIn("other_tool", test_bb)
+        string_output = test_bb.text_snapshot(keys=["speak"])
+        self.assertIn("speak", string_output)
+        self.assertNotIn("another_potential_tool", string_output)
+        self.assertNotIn("other_tool", string_output)
+        string_output = test_bb.text_snapshot(keys=["other_tool", "another_potential_tool"])
+        self.assertNotIn("speak", string_output)
+        self.assertIn("another_potential_tool", string_output)
+        self.assertIn("other_tool", string_output)
+
 
 if __name__ == "__main__":
     unittest.main()