fix(mma): Change self.engine to self.engines dict for concurrent track support
- self.engine was a single ConductorEngine reference that got overwritten when multiple tracks ran concurrently, orphaning the first track's engine - Now uses self.engines: Dict[str, ConductorEngine] keyed by track.id - Updated _spawn_worker, kill_worker, pause_mma, resume_mma, approve_ticket, _load_active_tickets, and _update_ticket_depends_on to use engines.get(track_id) Fixes concurrent MMA track execution bug where only one worker ever appeared.
This commit is contained in:
@@ -98,5 +98,13 @@
|
||||
"C:\\projects\\manual_slop\\conductor\\tracks\\data_oriented_optimization_20260312\\spec.md": {
|
||||
"hash": "8d64d4ed23a19cef973bb639dee2953492445f5598c08e7dd7c272183ad9848b",
|
||||
"summary": "This specification outlines a data-oriented optimization pass for Python code in `./src` and `./simulation`, aiming to minimize Python overhead by treating it as a procedural semantic definer. Key requirements include updating product guidelines, auditing and refactoring code, expanding profiling, and evaluating C extensions as a last resort, while maintaining test coverage and GUI responsiveness.\n\n**Outline:**\n**Markdown** \u2014 35 lines\nheadings:\n Specification: Data-Oriented Python Optimization Pass\n Overview\n Functional Requirements\n Non-Functional Requirements\n Acceptance Criteria\n Out of Scope"
|
||||
},
|
||||
"C:\\Users\\Ed\\AppData\\Local\\Temp\\pytest-of-Ed\\pytest-849\\test_auto_aggregate_skip0\\file1.txt": {
|
||||
"hash": "d0b425e00e15a0d36b9b361f02bab63563aed6cb4665083905386c55d5b679fa",
|
||||
"summary": "This document contains a single line of text, \"content1\". Its purpose and key takeaways are not discernible from the provided content.\n\n**Outline:**\n**TXT** \u2014 1 lines\npreview:\n```\ncontent1\n```"
|
||||
},
|
||||
"C:\\Users\\Ed\\AppData\\Local\\Temp\\pytest-of-Ed\\pytest-849\\test_force_full0\\other.txt": {
|
||||
"hash": "04d61c0832f9cbc2a210334352425d2519890a0a5945da96ccc5bd9ff101c4d3",
|
||||
"summary": "This document is a plain text file containing ten lines of content. The preview provided shows the first eight lines, indicating the file's straightforward, sequential nature.\n\n**Outline:**\n**TXT** \u2014 10 lines\npreview:\n```\nline1\nline2\nline3\nline4\nline5\nline6\nline7\nline8\n```"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
with open('src/app_controller.py', 'rb') as f:
|
||||
content = f.read()
|
||||
|
||||
# Fix _update_ticket_depends_on
|
||||
old = b' if self.engine:\r\n from src.dag_engine import TrackDAG, ExecutionEngine\r\n self.engine.dag = TrackDAG(self.active_track.tickets)\r\n self.engine.engine = ExecutionEngine(self.engine.dag, auto_queue=self.engine.engine.auto_queue)'
|
||||
new = b' engine = self.engines.get(self.active_track.id if self.active_track else None)\r\n if engine:\r\n from src.dag_engine import TrackDAG, ExecutionEngine\r\n engine.dag = TrackDAG(self.active_track.tickets)\r\n engine.engine = ExecutionEngine(engine.dag, auto_queue=engine.engine.auto_queue)'
|
||||
content = content.replace(old, new)
|
||||
|
||||
with open('src/app_controller.py', 'wb') as f:
|
||||
f.write(content)
|
||||
|
||||
print("Done")
|
||||
+16
-16
@@ -102,26 +102,26 @@ Collapsed=0
|
||||
DockId=0x0000000D,0
|
||||
|
||||
[Window][Discussion Hub]
|
||||
Pos=87,24
|
||||
Size=1593,1176
|
||||
Pos=1168,24
|
||||
Size=1593,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000006,0
|
||||
|
||||
[Window][Operations Hub]
|
||||
Pos=0,24
|
||||
Size=85,1176
|
||||
Size=1166,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000005,2
|
||||
|
||||
[Window][Files & Media]
|
||||
Pos=87,24
|
||||
Size=1593,1176
|
||||
Pos=1168,24
|
||||
Size=1593,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000006,1
|
||||
|
||||
[Window][AI Settings]
|
||||
Pos=0,24
|
||||
Size=85,1176
|
||||
Size=1166,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000005,0
|
||||
|
||||
@@ -131,14 +131,14 @@ Size=416,325
|
||||
Collapsed=0
|
||||
|
||||
[Window][MMA Dashboard]
|
||||
Pos=87,24
|
||||
Size=1593,1176
|
||||
Pos=1168,24
|
||||
Size=1593,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000006,2
|
||||
|
||||
[Window][Log Management]
|
||||
Pos=87,24
|
||||
Size=1593,1176
|
||||
Pos=1168,24
|
||||
Size=1593,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000006,3
|
||||
|
||||
@@ -407,13 +407,13 @@ DockId=0x00000006,1
|
||||
|
||||
[Window][Project Settings]
|
||||
Pos=0,24
|
||||
Size=85,1176
|
||||
Size=1166,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000005,1
|
||||
|
||||
[Window][Undo/Redo History]
|
||||
Pos=1268,24
|
||||
Size=1593,1754
|
||||
Pos=1168,24
|
||||
Size=1593,1564
|
||||
Collapsed=0
|
||||
DockId=0x00000006,4
|
||||
|
||||
@@ -551,12 +551,12 @@ Column 2 Width=150
|
||||
DockNode ID=0x00000008 Pos=3125,170 Size=593,1157 Split=Y
|
||||
DockNode ID=0x00000009 Parent=0x00000008 SizeRef=1029,147 Selected=0x0469CA7A
|
||||
DockNode ID=0x0000000A Parent=0x00000008 SizeRef=1029,145 Selected=0xDF822E02
|
||||
DockSpace ID=0xAFC85805 Window=0x079D3A04 Pos=0,24 Size=1680,1176 Split=X
|
||||
DockSpace ID=0xAFC85805 Window=0x079D3A04 Pos=0,24 Size=2761,1564 Split=X
|
||||
DockNode ID=0x00000003 Parent=0xAFC85805 SizeRef=2175,1183 Split=X
|
||||
DockNode ID=0x0000000B Parent=0x00000003 SizeRef=404,1186 Split=X Selected=0xF4139CA2
|
||||
DockNode ID=0x00000007 Parent=0x0000000B SizeRef=1512,858 Split=X Selected=0x8CA2375C
|
||||
DockNode ID=0x00000005 Parent=0x00000007 SizeRef=1266,1681 CentralNode=1 Selected=0x418C7449
|
||||
DockNode ID=0x00000006 Parent=0x00000007 SizeRef=1593,1681 Selected=0x2C0206CE
|
||||
DockNode ID=0x00000005 Parent=0x00000007 SizeRef=1266,1681 CentralNode=1 Selected=0x7BD57D6A
|
||||
DockNode ID=0x00000006 Parent=0x00000007 SizeRef=1593,1681 Selected=0x6F2B5B04
|
||||
DockNode ID=0x0000000E Parent=0x0000000B SizeRef=1777,858 Selected=0x418C7449
|
||||
DockNode ID=0x0000000D Parent=0x00000003 SizeRef=435,1186 Selected=0x363E93D6
|
||||
DockNode ID=0x00000004 Parent=0xAFC85805 SizeRef=1162,1183 Split=X Selected=0x3AEC3498
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -3,6 +3,7 @@ name = "project"
|
||||
git_dir = ""
|
||||
system_prompt = ""
|
||||
main_context = ""
|
||||
execution_mode = "native"
|
||||
|
||||
[output]
|
||||
output_dir = "./md_gen"
|
||||
|
||||
@@ -9,5 +9,5 @@ active = "main"
|
||||
|
||||
[discussions.main]
|
||||
git_commit = ""
|
||||
last_updated = "2026-05-02T14:52:30"
|
||||
last_updated = "2026-05-06T21:54:25"
|
||||
history = []
|
||||
|
||||
+22
-17
@@ -160,8 +160,7 @@ class AppController:
|
||||
self._loop_thread: Optional[threading.Thread] = None
|
||||
self.tracks: List[Dict[str, Any]] = []
|
||||
self.active_track: Optional[models.Track] = None
|
||||
self.engine: Optional[multi_agent_conductor.ConductorEngine] = None
|
||||
self.active_tickets: List[Dict[str, Any]] = []
|
||||
self.engines: Dict[str, multi_agent_conductor.ConductorEngine] = {}
|
||||
self.mma_streams: Dict[str, str] = {}
|
||||
self._worker_status: Dict[str, str] = {} # stream_id -> "running" | "completed" | "failed" | "killed"
|
||||
self.MAX_STREAM_SIZE: int = 10 * 1024 # 10KB max per stream
|
||||
@@ -2917,7 +2916,7 @@ class AppController:
|
||||
# Use the active track object directly to start execution
|
||||
self._set_mma_status("running")
|
||||
engine = multi_agent_conductor.ConductorEngine(self.active_track, self.event_queue, auto_queue=not self.mma_step_mode)
|
||||
self.engine = engine
|
||||
self.engines[self.active_track.id] = engine
|
||||
flat = project_manager.flat_config(self.project, self.active_discussion, track_id=self.active_track.id)
|
||||
full_md, _, _ = aggregate.run(flat)
|
||||
threading.Thread(target=engine.run, kwargs={"md_content": full_md}, daemon=True).start()
|
||||
@@ -2991,7 +2990,7 @@ class AppController:
|
||||
self._pending_gui_tasks.append({'action': 'refresh_from_project'})
|
||||
# 4. Initialize ConductorEngine and run loop
|
||||
engine = multi_agent_conductor.ConductorEngine(track, self.event_queue, auto_queue=not self.mma_step_mode)
|
||||
self.engine = engine
|
||||
self.engines[self.active_track.id] = engine
|
||||
# Use current full markdown context for the track execution
|
||||
track_id_param = track.id
|
||||
flat = project_manager.flat_config(self.project, self.active_discussion, track_id=track_id_param)
|
||||
@@ -3018,31 +3017,35 @@ class AppController:
|
||||
|
||||
def _spawn_worker(self, ticket_id: str, data: dict = None) -> None:
|
||||
"""Manually initiates a sub-agent execution for a ticket."""
|
||||
if self.engine:
|
||||
engine = self.engines.get(self.active_track.id if self.active_track else None)
|
||||
if engine:
|
||||
for t in self.active_track.tickets:
|
||||
if t.id == ticket_id:
|
||||
t.status = "todo"
|
||||
t.step_mode = False
|
||||
break
|
||||
self.engine.engine.auto_queue = True
|
||||
engine.engine.auto_queue = True
|
||||
self.event_queue.put("mma_retry", {"ticket_id": ticket_id})
|
||||
|
||||
def kill_worker(self, worker_id: str) -> None:
|
||||
"""Aborts a running worker."""
|
||||
if self.engine:
|
||||
self.engine.kill_worker(worker_id)
|
||||
engine = self.engines.get(self.active_track.id if self.active_track else None)
|
||||
if engine:
|
||||
engine.kill_worker(worker_id)
|
||||
|
||||
def pause_mma(self) -> None:
|
||||
"""Pauses the global MMA loop."""
|
||||
self.mma_step_mode = True
|
||||
if self.engine:
|
||||
self.engine.pause()
|
||||
engine = self.engines.get(self.active_track.id if self.active_track else None)
|
||||
if engine:
|
||||
engine.pause()
|
||||
|
||||
def resume_mma(self) -> None:
|
||||
"""Resumes the global MMA loop."""
|
||||
self.mma_step_mode = False
|
||||
if self.engine:
|
||||
self.engine.resume()
|
||||
engine = self.engines.get(self.active_track.id if self.active_track else None)
|
||||
if engine:
|
||||
engine.resume()
|
||||
|
||||
def inject_context(self, data: dict) -> None:
|
||||
"""Programmatic context injection."""
|
||||
@@ -3058,8 +3061,9 @@ class AppController:
|
||||
|
||||
def approve_ticket(self, ticket_id: str) -> None:
|
||||
"""Manually approves a ticket for execution."""
|
||||
if self.engine and self.engine.engine:
|
||||
self.engine.engine.approve_task(ticket_id)
|
||||
engine = self.engines.get(self.active_track.id if self.active_track else None)
|
||||
if engine and engine.engine:
|
||||
engine.engine.approve_task(ticket_id)
|
||||
else:
|
||||
# Fallback if engine not running
|
||||
for t in self.active_tickets:
|
||||
@@ -3082,10 +3086,11 @@ class AppController:
|
||||
if t.id == ticket_id:
|
||||
t.depends_on = depends_on
|
||||
break
|
||||
if self.engine:
|
||||
engine = self.engines.get(self.active_track.id if self.active_track else None)
|
||||
if engine:
|
||||
from src.dag_engine import TrackDAG, ExecutionEngine
|
||||
self.engine.dag = TrackDAG(self.active_track.tickets)
|
||||
self.engine.engine = ExecutionEngine(self.engine.dag, auto_queue=self.engine.engine.auto_queue)
|
||||
engine.dag = TrackDAG(self.active_track.tickets)
|
||||
engine.engine = ExecutionEngine(engine.dag, auto_queue=engine.engine.auto_queue)
|
||||
self._push_mma_state_update()
|
||||
|
||||
def _cb_run_conductor_setup(self) -> None:
|
||||
|
||||
Reference in New Issue
Block a user