1+ import logging
2+ import re
3+ from typing import Iterable , List , Optional
4+
5+ from v3 .models .models import MPlan , MStep
6+
7+ logger = logging .getLogger (__name__ )
8+
9+
10+ class PlanToMPlanConverter :
11+ """
12+ Convert a free-form, bullet-style plan string into an MPlan object.
13+
14+ Bullet parsing rules:
15+ 1. Recognizes lines starting (optionally with indentation) followed by -, *, or •
16+ 2. Attempts to resolve the agent in priority order:
17+ a. First bolded token (**AgentName**) if within detection window and in team
18+ b. Any team agent name appearing (case-insensitive) within the first detection window chars
19+ c. Fallback agent name (default 'MagenticAgent')
20+ 3. Removes the matched agent token from the action text
21+ 4. Ignores bullet lines whose remaining action is blank
22+
23+ Notes:
24+ - This does not mutate MPlan.user_id (caller can assign after parsing).
25+ - You can supply task text (becomes user_request) and facts text.
26+ - Optionally detect sub-bullets (indent > 0). If enabled, a `level` integer is
27+ returned alongside each MStep in an auxiliary `step_levels` list (since the
28+ current MStep model doesn’t have a level field).
29+
30+ Example:
31+ converter = PlanToMPlanConverter(team=["ResearchAgent","AnalysisAgent"])
32+ mplan = converter.parse(plan_text=raw, task="Analyze Q4", facts="Some facts")
33+
34+ """
35+
36+ BULLET_RE = re .compile (r'^(?P<indent>\s*)[-•*]\s+(?P<body>.+)$' )
37+ BOLD_AGENT_RE = re .compile (r'\*\*([A-Za-z0-9_]+)\*\*' )
38+ STRIP_BULLET_MARKER_RE = re .compile (r'^[-•*]\s+' )
39+
40+ def __init__ (
41+ self ,
42+ team : Iterable [str ],
43+ task : str = "" ,
44+ facts : str = "" ,
45+ detection_window : int = 25 ,
46+ fallback_agent : str = "MagenticAgent" ,
47+ enable_sub_bullets : bool = False ,
48+ trim_actions : bool = True ,
49+ collapse_internal_whitespace : bool = True ,
50+ ):
51+ self .team : List [str ] = list (team )
52+ self .task = task
53+ self .facts = facts
54+ self .detection_window = detection_window
55+ self .fallback_agent = fallback_agent
56+ self .enable_sub_bullets = enable_sub_bullets
57+ self .trim_actions = trim_actions
58+ self .collapse_internal_whitespace = collapse_internal_whitespace
59+
60+ # Map for faster case-insensitive lookups while preserving canonical form
61+ self ._team_lookup = {t .lower (): t for t in self .team }
62+
63+ # ---------------- Public API ---------------- #
64+
65+ def parse (self , plan_text : str ) -> MPlan :
66+ """
67+ Parse the supplied bullet-style plan text into an MPlan.
68+
69+ Returns:
70+ MPlan with team, user_request, facts, steps populated.
71+
72+ Side channel (if sub-bullets enabled):
73+ self.last_step_levels: List[int] parallel to steps (0 = top, 1 = sub, etc.)
74+ """
75+ mplan = MPlan ()
76+ mplan .team = self .team .copy ()
77+ mplan .user_request = self .task or mplan .user_request
78+ mplan .facts = self .facts or mplan .facts
79+
80+ lines = self ._preprocess_lines (plan_text )
81+
82+ step_levels : List [int ] = []
83+ for raw_line in lines :
84+ bullet_match = self .BULLET_RE .match (raw_line )
85+ if not bullet_match :
86+ continue # ignore non-bullet lines entirely
87+
88+ indent = bullet_match .group ("indent" ) or ""
89+ body = bullet_match .group ("body" ).strip ()
90+
91+ level = 0
92+ if self .enable_sub_bullets and indent :
93+ # Simple heuristic: any indentation => level 1 (could extend to deeper)
94+ level = 1
95+
96+ agent , action = self ._extract_agent_and_action (body )
97+
98+ if not action :
99+ continue
100+
101+ mplan .steps .append (MStep (agent = agent , action = action ))
102+ if self .enable_sub_bullets :
103+ step_levels .append (level )
104+
105+ if self .enable_sub_bullets :
106+ # Expose levels so caller can correlate (parallel list)
107+ self .last_step_levels = step_levels # type: ignore[attr-defined]
108+
109+ return mplan
110+
111+ # ---------------- Internal Helpers ---------------- #
112+
113+ def _preprocess_lines (self , plan_text : str ) -> List [str ]:
114+ lines = plan_text .splitlines ()
115+ cleaned : List [str ] = []
116+ for line in lines :
117+ stripped = line .rstrip ()
118+ if stripped :
119+ cleaned .append (stripped )
120+ return cleaned
121+
122+ def _extract_agent_and_action (self , body : str ) -> (str , str ):
123+ """
124+ Apply bold-first strategy, then window scan fallback.
125+ Returns (agent, action_text).
126+ """
127+ original = body
128+
129+ # 1. Try bold token
130+ agent , body_after = self ._try_bold_agent (original )
131+ if agent :
132+ action = self ._finalize_action (body_after )
133+ return agent , action
134+
135+ # 2. Try window scan
136+ agent2 , body_after2 = self ._try_window_agent (original )
137+ if agent2 :
138+ action = self ._finalize_action (body_after2 )
139+ return agent2 , action
140+
141+ # 3. Fallback
142+ action = self ._finalize_action (original )
143+ return self .fallback_agent , action
144+
145+ def _try_bold_agent (self , text : str ) -> (Optional [str ], str ):
146+ m = self .BOLD_AGENT_RE .search (text )
147+ if not m :
148+ return None , text
149+ if m .start () <= self .detection_window :
150+ candidate = m .group (1 )
151+ canonical = self ._team_lookup .get (candidate .lower ())
152+ if canonical : # valid agent
153+ cleaned = text [:m .start ()] + text [m .end ():]
154+ return canonical , cleaned .strip ()
155+ return None , text
156+
157+ def _try_window_agent (self , text : str ) -> (Optional [str ], str ):
158+ head_segment = text [: self .detection_window ].lower ()
159+ for canonical in self .team :
160+ if canonical .lower () in head_segment :
161+ # Remove first occurrence (case-insensitive)
162+ pattern = re .compile (re .escape (canonical ), re .IGNORECASE )
163+ cleaned = pattern .sub ("" , text , count = 1 )
164+ cleaned = cleaned .replace ("*" , "" )
165+ return canonical , cleaned .strip ()
166+ return None , text
167+
168+ def _finalize_action (self , action : str ) -> str :
169+ if self .trim_actions :
170+ action = action .strip ()
171+ if self .collapse_internal_whitespace :
172+ action = re .sub (r'\s+' , ' ' , action )
173+ return action
174+
175+ # --------------- Convenience (static) --------------- #
176+
177+ @staticmethod
178+ def convert (
179+ plan_text : str ,
180+ team : Iterable [str ],
181+ task : str = "" ,
182+ facts : str = "" ,
183+ ** kwargs ,
184+ ) -> MPlan :
185+ """
186+ One-shot convenience method:
187+ mplan = PlanToMPlanConverter.convert(plan_text, team, task="X")
188+ """
189+ return PlanToMPlanConverter (
190+ team = team ,
191+ task = task ,
192+ facts = facts ,
193+ ** kwargs ,
194+ ).parse (plan_text )
0 commit comments