Merge pull request #2 from OmniJax/chore/vibe-safety

chore(vibe): non-core safety and maintainability improvements

Author: AckerlyLau

masfactory/components/vibe/vibe_graph.py

@@ -107,6 +107,10 @@ def build(self):
                         "system_advice": "",
                     }
                 )
+                if not isinstance(output, dict):
+                    raise TypeError(
+                        f"Vibe build workflow must return dict output, got {type(output).__name__}"
+                    )
                 raw_graph_design = output.get("graph_design", {})
                 graph_design = normalize_graph_design(raw_graph_design)
         else:

masfactory/components/vibe/vibe_workflow/planner/planner_diagnose_loop.py

@@ -1,8 +1,23 @@
-from masfactory import NodeTemplate,Loop
+from masfactory import NodeTemplate, Loop
 from .planner_agent import PlannerAgent
 from .diagnose_node import DiagnoseNode
+
+
+def _as_bool(value: object) -> bool:
+    """Best-effort bool coercion used by loop termination checks."""
+    if isinstance(value, bool):
+        return value
+    if isinstance(value, str):
+        lowered = value.strip().lower()
+        if lowered in {"true", "1", "yes", "y", "on"}:
+            return True
+        if lowered in {"false", "0", "no", "n", "off", ""}:
+            return False
+    return bool(value)
+
+
 def terminate_check(messages: dict, _attributes: dict | None = None) -> bool:
-    return not bool(messages.get("diagnose_has_issues", False))
+    return not _as_bool(messages.get("diagnose_has_issues", False))
 
 
 PlannerDiagnoseLoop = NodeTemplate(

masfactory/components/vibe/vibe_workflow/planner/planner_graph.py

@@ -1,92 +1,63 @@
-from masfactory import Loop,NodeTemplate,HumanFileEditVisual
+from masfactory import Loop, NodeTemplate, HumanFileEditVisual
 from .planner_diagnose_loop import PlannerDiagnoseLoop
-def terminate_check(messages:dict):
-   user_advice:str = messages.get("user_advice","")
-   if len(user_advice.strip()) == 0 or "agree" in user_advice.lower():
-      return True
-   else:
-      return False
+
+
+def terminate_check(messages: dict) -> bool:
+    """Stop when user gives no feedback or explicitly agrees."""
+    user_advice = messages.get("user_advice", "")
+    if not isinstance(user_advice, str):
+        user_advice = str(user_advice)
+    return (not user_advice.strip()) or ("agree" in user_advice.lower())
 
 PlannerHuman = NodeTemplate(
-   HumanFileEditVisual,
-   pull_keys={},
-   push_keys={},
+    HumanFileEditVisual,
+    pull_keys={},
+    push_keys={},
 )
 
-# PlannerGraph = NodeTemplate(
-#    Loop,
-#    pull_keys={},
-#    push_keys={},
-#    terminate_condition_function=terminate_check,
-#    nodes=[
-#       ("planner",Planner),
-#       ("planner-human",PlannerHuman)
-#    ],
-#    edges=[
-#       ("CONTROLLER","planner",
-#          {
-#             "user_demand":"",
-#             "role_list":"",
-#             "user_advice":"No advice yet."
-#          }
-#       ),
-#       ("planner","planner-human",
-#          {
-#             "graph_design":"The generated graph design accroding to the user's demand and the roles."
-#          }
-#       ),
-#       ("planner","CONTROLLER",
-#          {
-#             "graph_design":"The generated graph design accroding to the user's demand and the roles."
-#          }
-#       ),
-#       ("planner-human","CONTROLLER",
-#          {
-#             "user_advice":"Do you agree the plan? If you agree, enter AGREE. If you have any comments, please enter your comments."
-#          }
-#       )
-#    ]
-# )
-
 PlannerGraph = NodeTemplate(
     Loop,
     terminate_condition_function=terminate_check,
-    pull_keys={"cache_file_path":""},
+    pull_keys={"cache_file_path": ""},
     push_keys={},
-   nodes=[
-      ("planner-diagnose-loop",PlannerDiagnoseLoop),
-      ("planner-human",PlannerHuman)
-   ],
+    nodes=[
+        ("planner-diagnose-loop", PlannerDiagnoseLoop),
+        ("planner-human", PlannerHuman),
+    ],
     edges=[
         (
-            "CONTROLLER","planner-diagnose-loop",
+            "CONTROLLER",
+            "planner-diagnose-loop",
             {
                 "user_demand": "User demand", 
                 "role_list": "Role list",
-                "user_advice":"User's feedback"
+                "user_advice": "User's feedback",
             },
         ),
         (
-            "planner-diagnose-loop","planner-human",
+            "planner-diagnose-loop",
+            "planner-human",
             {
-                "graph_design":"The generated graph design accroding to the user's demand and the roles."
+                "graph_design": "The generated graph design accroding to the user's demand and the roles."
             },
         ),
         (
-            "planner-human","CONTROLLER",
+            "planner-human",
+            "CONTROLLER",
             {
-                "user_advice":"Do you agree the plan? If you agree, enter AGREE. If you have any comments, please enter your comments."
+                "user_advice": "Do you agree the plan? If you agree, enter AGREE. If you have any comments, please enter your comments."
             }
         ),
         (
-            "planner-diagnose-loop","CONTROLLER",
+            "planner-diagnose-loop",
+            "CONTROLLER",
             {
-                "graph_design":"The generated graph design accroding to the user's demand and the roles."
+                "graph_design": "The generated graph design accroding to the user's demand and the roles."
             }
-        )
+        ),
     ],
 )
 
-__ALL__=[
-    'PlannerGraph'
-]
+__ALL__ = [
+    "PlannerGraph",
+]

masfactory/components/vibe/vibe_workflow/role_assigner.py

@@ -48,12 +48,14 @@
    pull_keys={},
    push_keys={},
 )
-def terminate_check(messages:dict):
-   user_advice:str = messages.get("user_advice","")
-   if len(user_advice.strip()) == 0 or "agree" in user_advice.lower():
-      return True
-   else:
-      return False
+
+
+def terminate_check(messages: dict) -> bool:
+   """Stop when user gives no feedback or explicitly agrees."""
+   user_advice = messages.get("user_advice", "")
+   if not isinstance(user_advice, str):
+      user_advice = str(user_advice)
+   return (not user_advice.strip()) or ("agree" in user_advice.lower())
 
 RoleAssignerGraph = NodeTemplate(
    Loop,