From aa62f4cb09e64106718b50c40d9d12f887d1abd8 Mon Sep 17 00:00:00 2001 From: RooberSmoth <96537304+RooberSmoth@users.noreply.github.com> Date: Fri, 7 Aug 2026 10:19:52 +0900 Subject: [PATCH 1/3] Adding print statements near trigger actions --- agent_core/core/impl/trigger/session_queue.py | 14 +++++++++++++- app/triggers/runtime.py | 5 +++++ 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/agent_core/core/impl/trigger/session_queue.py b/agent_core/core/impl/trigger/session_queue.py index 6db2dd97..fe0396c6 100644 --- a/agent_core/core/impl/trigger/session_queue.py +++ b/agent_core/core/impl/trigger/session_queue.py @@ -67,6 +67,7 @@ async def put(self, trig: Trigger) -> None: async with self._cv: if self._closed: raise QueueClosed(self.session_id) + print(f"Trigger added: {trig.source}") heapq.heappush(self._heap, (trig.fire_at, next(self._seq), trig)) self._cv.notify() @@ -120,7 +121,12 @@ async def pop_due_batch(self) -> List[Trigger]: now = time.time() batch: List[tuple] = [] while self._heap and self._heap[0][0] <= now: - batch.append(heapq.heappop(self._heap)) + entry = heapq.heappop(self._heap) #Storing the heappop + trigger = entry[2] #find the trigger + + print(f"popping trigger: {trigger.source}") + print(f"trigger={trigger}") + batch.append(entry) return [entry[2] for entry in batch] async def purge(self, predicate) -> int: @@ -135,11 +141,17 @@ async def purge(self, predicate) -> int: async with self._cv: if self._closed or not self._heap: return 0 + print("\n===== QUEUE BEFORE PURGE =====") + for _, _, trigger in self._heap: + print(trigger) kept = [entry for entry in self._heap if not predicate(entry[2])] removed = [entry[2] for entry in self._heap if predicate(entry[2])] if not removed: return 0 self._heap = kept + print("\n===== QUEUE AFTERRR PURGE =====") + for _, _, trigger in self._heap: + print(trigger) heapq.heapify(self._heap) self._notify_evicted(removed) return len(removed) diff --git a/app/triggers/runtime.py b/app/triggers/runtime.py index dd8893de..6f468623 100644 --- a/app/triggers/runtime.py +++ b/app/triggers/runtime.py @@ -261,7 +261,9 @@ async def request_stop(self, session_id: str) -> bool: (bounded by STOP_SETTLE_TIMEOUT_S) before returning, so the caller can treat the return as "everything is shut". """ + print("Stop requested ()") purged = await self._purge_continuations(session_id) + print("Purging continuations ()") turn = self._turns.get(session_id) turn_inflight = turn is not None and not turn.done() @@ -288,9 +290,11 @@ async def request_stop(self, session_id: str) -> bool: except Exception as e: logger.warning(f"[SessionRuntime] Process kill failed: {e}") + print("Cancelling turn ()") turn.cancel() try: await asyncio.wait_for(signal.settled.wait(), STOP_SETTLE_TIMEOUT_S) + print("Turn settled ()") except asyncio.TimeoutError: # The turn is stuck in a non-cancellable await. Finalize anyway: # the UI must reach idle, and the abandoned coroutine can no @@ -307,6 +311,7 @@ async def request_stop(self, session_id: str) -> bool: async def _purge_continuations(self, session_id: str) -> int: """Drop a session's queued run-continuation triggers.""" + """I need to make a change here""" queue = self._queues.get(session_id) if queue is None: return 0 From e024e725c664be8debb31adcca97f3d2583d5478 Mon Sep 17 00:00:00 2001 From: RooberSmoth <96537304+RooberSmoth@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:15:40 +0900 Subject: [PATCH 2/3] added statements to print into logger --- agent_core/core/impl/llm/interface.py | 4 ++- agent_core/core/impl/trigger/session_queue.py | 31 ++++++++++++++----- app/agent_base.py | 3 ++ app/triggers/runtime.py | 8 ++--- app/ui_layer/controller/ui_controller.py | 2 +- 5 files changed, 35 insertions(+), 13 deletions(-) diff --git a/agent_core/core/impl/llm/interface.py b/agent_core/core/impl/llm/interface.py index 43a89489..c2b3a931 100644 --- a/agent_core/core/impl/llm/interface.py +++ b/agent_core/core/impl/llm/interface.py @@ -643,7 +643,9 @@ def _generate_response_sync( raise LLMConsecutiveFailureError(self._consecutive_failures) if log_response: - logger.info(f"[LLM SEND] system={system_prompt} | user={user_prompt}") + #logger.info(f"[LLM SEND] system={system_prompt} | user={user_prompt}") + #REMOVEEEE + pass try: if self.provider in ( diff --git a/agent_core/core/impl/trigger/session_queue.py b/agent_core/core/impl/trigger/session_queue.py index fe0396c6..2decb343 100644 --- a/agent_core/core/impl/trigger/session_queue.py +++ b/agent_core/core/impl/trigger/session_queue.py @@ -67,7 +67,7 @@ async def put(self, trig: Trigger) -> None: async with self._cv: if self._closed: raise QueueClosed(self.session_id) - print(f"Trigger added: {trig.source}") + logger.warning(f"Trigger added: {trig.source}") heapq.heappush(self._heap, (trig.fire_at, next(self._seq), trig)) self._cv.notify() @@ -120,15 +120,31 @@ async def pop_due_batch(self) -> List[Trigger]: return [] now = time.time() batch: List[tuple] = [] + + + logger.warning("-------------queue before pop") + for _, _, t in self._heap: + logger.warning(t) + + while self._heap and self._heap[0][0] <= now: entry = heapq.heappop(self._heap) #Storing the heappop trigger = entry[2] #find the trigger - print(f"popping trigger: {trigger.source}") - print(f"trigger={trigger}") + logger.warning(f"popping trigger: {trigger.source}") + logger.warning(f"trigger={trigger}") batch.append(entry) + + + logger.warning("---------------queue after pop") + for _, _, t in self._heap: + logger.warning(t) + + return [entry[2] for entry in batch] + + async def purge(self, predicate) -> int: """Remove queued triggers matching ``predicate`` (a Trigger -> bool). @@ -138,20 +154,21 @@ async def purge(self, predicate) -> int: durable rows settle instead of rehydrating next boot. Returns the number of triggers removed. """ + async with self._cv: if self._closed or not self._heap: return 0 - print("\n===== QUEUE BEFORE PURGE =====") + logger.warning("\n===== QUEUE BEFORE PURGE =====") for _, _, trigger in self._heap: - print(trigger) + logger.warning(trigger) kept = [entry for entry in self._heap if not predicate(entry[2])] removed = [entry[2] for entry in self._heap if predicate(entry[2])] if not removed: return 0 self._heap = kept - print("\n===== QUEUE AFTERRR PURGE =====") + logger.warning("\n===== QUEUE AFTERRR PURGE =====") for _, _, trigger in self._heap: - print(trigger) + logger.warning(trigger) heapq.heapify(self._heap) self._notify_evicted(removed) return len(removed) diff --git a/app/agent_base.py b/app/agent_base.py index c6098764..3fcbf964 100644 --- a/app/agent_base.py +++ b/app/agent_base.py @@ -519,6 +519,7 @@ async def react(self, trigger: Trigger) -> None: trigger: The Trigger that wakes the session and describes when and why it should act. """ + session_id = trigger.session_id or MAIN_SESSION_ID try: @@ -1425,6 +1426,7 @@ async def _on_run_stopped(self, session_id: str) -> None: resurrecting it immediately would make the stop button a no-op. """ self._lui_run_writes.pop(session_id, None) + logger.warning("Run stopped, agent base") # A force-stopped memory run must not leave the unprocessed buffer # frozen forever. @@ -1440,6 +1442,7 @@ async def _on_run_stopped(self, session_id: str) -> None: if self.event_stream_manager: msg = "User force-stopped the run. The work in progress was halted." try: + logger.warning("run stopped.") self.event_stream_manager.log( "system", msg, diff --git a/app/triggers/runtime.py b/app/triggers/runtime.py index 6f468623..137bf8e8 100644 --- a/app/triggers/runtime.py +++ b/app/triggers/runtime.py @@ -261,9 +261,9 @@ async def request_stop(self, session_id: str) -> bool: (bounded by STOP_SETTLE_TIMEOUT_S) before returning, so the caller can treat the return as "everything is shut". """ - print("Stop requested ()") + logger.warning("Stop requested ()") purged = await self._purge_continuations(session_id) - print("Purging continuations ()") + logger.warning("Purging continuations ()") turn = self._turns.get(session_id) turn_inflight = turn is not None and not turn.done() @@ -290,11 +290,11 @@ async def request_stop(self, session_id: str) -> bool: except Exception as e: logger.warning(f"[SessionRuntime] Process kill failed: {e}") - print("Cancelling turn ()") + logger.warning("Cancelling turn ()") turn.cancel() try: await asyncio.wait_for(signal.settled.wait(), STOP_SETTLE_TIMEOUT_S) - print("Turn settled ()") + logger.warning("Turn settled ()") except asyncio.TimeoutError: # The turn is stuck in a non-cancellable await. Finalize anyway: # the UI must reach idle, and the abandoned coroutine can no diff --git a/app/ui_layer/controller/ui_controller.py b/app/ui_layer/controller/ui_controller.py index 56ea7946..781cd988 100644 --- a/app/ui_layer/controller/ui_controller.py +++ b/app/ui_layer/controller/ui_controller.py @@ -19,7 +19,6 @@ from app.agent_base import AgentBase from app.ui_layer.adapters.base import InterfaceAdapter - @dataclass class UIControllerConfig: """ @@ -296,6 +295,7 @@ async def stop_run(self, session_id: Optional[str] = None) -> bool: Returns True when a run was actually stopped. Run-state broadcasts ("stopping" then "idle") come from the agent, not from here. """ + logger.warning("stop button pressed.") return await self._agent.request_run_stop(session_id or "main") async def notify_session_updated(self, session_id: str) -> None: From 21777b504bb9eea23cc6fbe83ff98706783c7474 Mon Sep 17 00:00:00 2001 From: ahmad-ajmal Date: Fri, 7 Aug 2026 10:34:03 +0100 Subject: [PATCH 3/3] Fix run-stop: propagate cancellation into in-flight actions and await their settlement before "Run stopped" (wait_for shim, executor threads, action manager re-raise). Adds regression tests; removes debug prints. --- agent_core/core/impl/action/executor.py | 19 ++++-- agent_core/core/impl/action/manager.py | 64 +++++++++++++++++-- agent_core/core/impl/llm/interface.py | 4 +- agent_core/core/impl/trigger/session_queue.py | 31 +-------- app/agent_base.py | 3 - app/triggers/runtime.py | 5 -- app/ui_layer/controller/ui_controller.py | 2 +- 7 files changed, 73 insertions(+), 55 deletions(-) diff --git a/agent_core/core/impl/action/executor.py b/agent_core/core/impl/action/executor.py index de1bb8ca..60888898 100644 --- a/agent_core/core/impl/action/executor.py +++ b/agent_core/core/impl/action/executor.py @@ -627,12 +627,19 @@ async def _atomic_action_internal_async( logger.debug( f"[SYNC] Action '{action_name}' is sync, running in thread pool" ) - loop = asyncio.get_running_loop() - execution_result = await loop.run_in_executor( - THREAD_POOL, - function_to_call, - input_data, - ) + thread_future = THREAD_POOL.submit(function_to_call, input_data) + try: + execution_result = await asyncio.wrap_future(thread_future) + except asyncio.CancelledError: + # A user force-stop cancelled this turn, but a thread cannot + # be interrupted mid-body — and the stop contract (PR #410) + # is that settlement WAITS for in-flight work: the spinner + # runs until the last real action finishes, and only then + # "Run stopped." shows. Without this wait the thread became + # an orphan whose message/file output landed after the stop. + while not thread_future.done(): + await asyncio.sleep(0.05) + raise return execution_result diff --git a/agent_core/core/impl/action/manager.py b/agent_core/core/impl/action/manager.py index e1491700..7fc70416 100644 --- a/agent_core/core/impl/action/manager.py +++ b/agent_core/core/impl/action/manager.py @@ -53,7 +53,26 @@ async def _compat_wait_for(fut, timeout): if timeout is None: return await fut task = asyncio.ensure_future(fut) - _done, pending = await asyncio.wait({task}, timeout=timeout) + try: + _done, pending = await asyncio.wait({task}, timeout=timeout) + except asyncio.CancelledError: + # Real wait_for GUARANTEES the wrapped future is cancelled + # when the outer await is cancelled; asyncio.wait does NOT + # cancel its input tasks, so without this branch a user + # force-stop unwound the turn while the executing ACTION + # kept running as an orphaned task — its message/file output + # landed seconds after "Run stopped." (PR #410). Cancel the + # inner task and AWAIT its unwind before re-raising: that + # wait is what keeps the stop spinner honest — an action + # whose sync body is mid-flight in a worker thread only + # unwinds when the thread finishes, so settlement (and the + # "Run stopped." bubble) waits for the last real work. + task.cancel() + try: + await task + except BaseException: + pass + raise if task in pending: task.cancel() try: @@ -328,6 +347,7 @@ async def execute_action( logger.debug(f"Starting execution of action {action.name}...") + was_cancelled = False try: # ──────────────────────────────────────────────────────────── # 2. Execute @@ -398,8 +418,19 @@ async def execute_action( status = "success" except asyncio.CancelledError: + # A user force-stop cancelled the turn mid-action. Record the + # outcome (the event stream and idempotency ledger must show the + # action was cancelled), but DO NOT swallow the cancellation: + # catching CancelledError without re-raising un-cancels the task, + # and the react loop then treated "Action cancelled" as an + # ordinary failed action and started the NEXT LLM call — a zombie + # turn the stop's settlement wait could never catch, surfacing as + # "Stop settlement timed out" + forceful finalize (observed live + # 2026-08-07, PR #410). The re-raise happens AFTER persistence, + # at the end of this method. status = "error" outputs = {"error": "Action cancelled", "error_code": "cancelled"} + was_cancelled = True except Exception as e: status = "error" outputs = {"error": str(e)} @@ -503,6 +534,11 @@ async def execute_action( logger.debug(f"Action {action.name} removed from in-flight tracking.") + if was_cancelled: + # Bookkeeping is done (event stream + ledger show the cancelled + # outcome); now let the stop actually stop the turn. + raise asyncio.CancelledError() + return outputs @profile( @@ -574,19 +610,33 @@ async def execute_single( input_data=input_data, ) - # All parallel actions run under the parent session_id. + # All parallel actions run under the parent session_id. Real tasks + # (not bare coroutines) so a cancelled batch can be awaited below. parallel_tasks = [ - execute_single(action, input_data, session_id) + asyncio.ensure_future(execute_single(action, input_data, session_id)) for action, input_data in actions ] # Execute all actions in parallel - results = await asyncio.gather(*parallel_tasks, return_exceptions=True) - - # Process results, converting exceptions to error dicts + try: + results = await asyncio.gather(*parallel_tasks, return_exceptions=True) + except asyncio.CancelledError: + # User force-stop: gather cancels the children but raises without + # waiting for their unwind — which includes each action's + # cancelled-outcome bookkeeping and the wait for uninterruptible + # thread bodies. Settling here keeps the stop contract (spinner + # until the last in-flight action finalizes) for parallel batches + # exactly as execute_action keeps it for single ones (PR #410). + await asyncio.wait(parallel_tasks) + raise + + # Process results, converting exceptions to error dicts. + # BaseException, not Exception: a child's CancelledError would + # otherwise pass isinstance() and leak an exception OBJECT into the + # results list, crashing the status tally below. processed = [] for i, result in enumerate(results): - if isinstance(result, Exception): + if isinstance(result, BaseException): logger.error(f"[PARALLEL] Action {actions[i][0].name} failed: {result}") processed.append( { diff --git a/agent_core/core/impl/llm/interface.py b/agent_core/core/impl/llm/interface.py index c2b3a931..43a89489 100644 --- a/agent_core/core/impl/llm/interface.py +++ b/agent_core/core/impl/llm/interface.py @@ -643,9 +643,7 @@ def _generate_response_sync( raise LLMConsecutiveFailureError(self._consecutive_failures) if log_response: - #logger.info(f"[LLM SEND] system={system_prompt} | user={user_prompt}") - #REMOVEEEE - pass + logger.info(f"[LLM SEND] system={system_prompt} | user={user_prompt}") try: if self.provider in ( diff --git a/agent_core/core/impl/trigger/session_queue.py b/agent_core/core/impl/trigger/session_queue.py index 2decb343..6db2dd97 100644 --- a/agent_core/core/impl/trigger/session_queue.py +++ b/agent_core/core/impl/trigger/session_queue.py @@ -67,7 +67,6 @@ async def put(self, trig: Trigger) -> None: async with self._cv: if self._closed: raise QueueClosed(self.session_id) - logger.warning(f"Trigger added: {trig.source}") heapq.heappush(self._heap, (trig.fire_at, next(self._seq), trig)) self._cv.notify() @@ -120,31 +119,10 @@ async def pop_due_batch(self) -> List[Trigger]: return [] now = time.time() batch: List[tuple] = [] - - - logger.warning("-------------queue before pop") - for _, _, t in self._heap: - logger.warning(t) - - while self._heap and self._heap[0][0] <= now: - entry = heapq.heappop(self._heap) #Storing the heappop - trigger = entry[2] #find the trigger - - logger.warning(f"popping trigger: {trigger.source}") - logger.warning(f"trigger={trigger}") - batch.append(entry) - - - logger.warning("---------------queue after pop") - for _, _, t in self._heap: - logger.warning(t) - - + batch.append(heapq.heappop(self._heap)) return [entry[2] for entry in batch] - - async def purge(self, predicate) -> int: """Remove queued triggers matching ``predicate`` (a Trigger -> bool). @@ -154,21 +132,14 @@ async def purge(self, predicate) -> int: durable rows settle instead of rehydrating next boot. Returns the number of triggers removed. """ - async with self._cv: if self._closed or not self._heap: return 0 - logger.warning("\n===== QUEUE BEFORE PURGE =====") - for _, _, trigger in self._heap: - logger.warning(trigger) kept = [entry for entry in self._heap if not predicate(entry[2])] removed = [entry[2] for entry in self._heap if predicate(entry[2])] if not removed: return 0 self._heap = kept - logger.warning("\n===== QUEUE AFTERRR PURGE =====") - for _, _, trigger in self._heap: - logger.warning(trigger) heapq.heapify(self._heap) self._notify_evicted(removed) return len(removed) diff --git a/app/agent_base.py b/app/agent_base.py index 3fcbf964..c6098764 100644 --- a/app/agent_base.py +++ b/app/agent_base.py @@ -519,7 +519,6 @@ async def react(self, trigger: Trigger) -> None: trigger: The Trigger that wakes the session and describes when and why it should act. """ - session_id = trigger.session_id or MAIN_SESSION_ID try: @@ -1426,7 +1425,6 @@ async def _on_run_stopped(self, session_id: str) -> None: resurrecting it immediately would make the stop button a no-op. """ self._lui_run_writes.pop(session_id, None) - logger.warning("Run stopped, agent base") # A force-stopped memory run must not leave the unprocessed buffer # frozen forever. @@ -1442,7 +1440,6 @@ async def _on_run_stopped(self, session_id: str) -> None: if self.event_stream_manager: msg = "User force-stopped the run. The work in progress was halted." try: - logger.warning("run stopped.") self.event_stream_manager.log( "system", msg, diff --git a/app/triggers/runtime.py b/app/triggers/runtime.py index 137bf8e8..dd8893de 100644 --- a/app/triggers/runtime.py +++ b/app/triggers/runtime.py @@ -261,9 +261,7 @@ async def request_stop(self, session_id: str) -> bool: (bounded by STOP_SETTLE_TIMEOUT_S) before returning, so the caller can treat the return as "everything is shut". """ - logger.warning("Stop requested ()") purged = await self._purge_continuations(session_id) - logger.warning("Purging continuations ()") turn = self._turns.get(session_id) turn_inflight = turn is not None and not turn.done() @@ -290,11 +288,9 @@ async def request_stop(self, session_id: str) -> bool: except Exception as e: logger.warning(f"[SessionRuntime] Process kill failed: {e}") - logger.warning("Cancelling turn ()") turn.cancel() try: await asyncio.wait_for(signal.settled.wait(), STOP_SETTLE_TIMEOUT_S) - logger.warning("Turn settled ()") except asyncio.TimeoutError: # The turn is stuck in a non-cancellable await. Finalize anyway: # the UI must reach idle, and the abandoned coroutine can no @@ -311,7 +307,6 @@ async def request_stop(self, session_id: str) -> bool: async def _purge_continuations(self, session_id: str) -> int: """Drop a session's queued run-continuation triggers.""" - """I need to make a change here""" queue = self._queues.get(session_id) if queue is None: return 0 diff --git a/app/ui_layer/controller/ui_controller.py b/app/ui_layer/controller/ui_controller.py index 781cd988..56ea7946 100644 --- a/app/ui_layer/controller/ui_controller.py +++ b/app/ui_layer/controller/ui_controller.py @@ -19,6 +19,7 @@ from app.agent_base import AgentBase from app.ui_layer.adapters.base import InterfaceAdapter + @dataclass class UIControllerConfig: """ @@ -295,7 +296,6 @@ async def stop_run(self, session_id: Optional[str] = None) -> bool: Returns True when a run was actually stopped. Run-state broadcasts ("stopping" then "idle") come from the agent, not from here. """ - logger.warning("stop button pressed.") return await self._agent.request_run_stop(session_id or "main") async def notify_session_updated(self, session_id: str) -> None: