feat(conductor): Implement abort checks in worker lifecycle and kill_worker method

src/multi_agent_conductor.py

@@ -98,6 +98,21 @@ class ConductorEngine:
     self.tier_usage[tier]["input"] += input_tokens
     self.tier_usage[tier]["output"] += output_tokens
 
+ def kill_worker(self, ticket_id: str) -> None:
+  """Sets the abort event for a worker and attempts to join its thread."""
+  if ticket_id in self._abort_events:
+   print(f"[MMA] Setting abort event for {ticket_id}")
+   self._abort_events[ticket_id].set()
+  
+  with self._workers_lock:
+   thread = self._active_workers.get(ticket_id)
+  
+  if thread:
+   print(f"[MMA] Joining thread for {ticket_id}")
+   thread.join(timeout=1.0)
+   with self._workers_lock:
+    self._active_workers.pop(ticket_id, None)
+
  def _push_state(self, status: str = "running", active_tier: str = None) -> None:
   if not self.event_queue:
    return
@@ -221,6 +236,8 @@ class ConductorEngine:
       )
       
       if spawned:
+       with self._workers_lock:
+        self._active_workers[ticket.id] = spawned
        ticket.status = "in_progress"
        _queue_put(self.event_queue, "ticket_started", {"ticket_id": ticket.id, "timestamp": time.time()})
        print(f"Executing ticket {ticket.id}: {ticket.description}")
@@ -317,6 +334,17 @@ def run_worker_lifecycle(ticket: Ticket, context: WorkerContext, context_files:
  # Enforce Context Amnesia: each ticket starts with a clean slate.
  ai_client.reset_session()
  ai_client.set_provider(ai_client.get_provider(), context.model_name)
+
+ # Check for abort BEFORE any major work
+ if engine and hasattr(engine, "_abort_events"):
+  abort_event = engine._abort_events.get(ticket.id)
+  if abort_event and abort_event.is_set():
+   print(f"[MMA] Ticket {ticket.id} aborted early.")
+   ticket.status = "killed"
+   if event_queue:
+    _queue_put(event_queue, "ticket_completed", {"ticket_id": ticket.id, "timestamp": time.time()})
+   return "ABORTED"
+
  context_injection = ""
  tokens_before = 0
  tokens_after = 0
@@ -383,6 +411,12 @@ def run_worker_lifecycle(ticket: Ticket, context: WorkerContext, context_files:
  def clutch_callback(payload: str) -> bool:
   if not event_queue:
    return True
+  # SECONDARY CHECK: Before executing any tool, check abort
+  if engine and hasattr(engine, "_abort_events"):
+   abort_event = engine._abort_events.get(ticket.id)
+   if abort_event and abort_event.is_set():
+    print(f"[MMA] Ticket {ticket.id} aborted during clutch_callback.")
+    return False # Reject tool execution
   return confirm_execution(payload, event_queue, ticket.id)
 
  def stream_callback(chunk: str) -> None:
@@ -423,6 +457,17 @@ def run_worker_lifecycle(ticket: Ticket, context: WorkerContext, context_files:
  finally:
   ai_client.comms_log_callback = old_comms_cb
   ai_client.set_current_tier(None)
+
+ # THIRD CHECK: After blocking send() returns
+ if engine and hasattr(engine, "_abort_events"):
+  abort_event = engine._abort_events.get(ticket.id)
+  if abort_event and abort_event.is_set():
+   print(f"[MMA] Ticket {ticket.id} aborted after AI call.")
+   ticket.status = "killed"
+   if event_queue:
+    _queue_put(event_queue, "ticket_completed", {"ticket_id": ticket.id, "timestamp": time.time()})
+   return "ABORTED"
+
  if event_queue:
   # Push via "response" event type — _process_event_queue wraps this
   # as {"action": "handle_ai_response", "payload": ...} for the GUI.