fix(trigger): keep local worker alive on turn cancel
This commit is contained in:
@@ -79,6 +79,8 @@ class AutomationTurnCoordinator:
|
|||||||
return await future
|
return await future
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
raise
|
raise
|
||||||
|
except AutomationTurnError:
|
||||||
|
raise
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
raise AutomationTurnError(str(exc) or exc.__class__.__name__) from exc
|
raise AutomationTurnError(str(exc) or exc.__class__.__name__) from exc
|
||||||
finally:
|
finally:
|
||||||
@@ -118,6 +120,8 @@ class AutomationTurnCoordinator:
|
|||||||
if future is None or future.done():
|
if future is None or future.done():
|
||||||
return
|
return
|
||||||
if error is not None:
|
if error is not None:
|
||||||
|
if isinstance(error, asyncio.CancelledError):
|
||||||
|
error = AutomationTurnError(str(error) or error.__class__.__name__)
|
||||||
future.set_exception(error)
|
future.set_exception(error)
|
||||||
else:
|
else:
|
||||||
future.set_result(response)
|
future.set_result(response)
|
||||||
|
|||||||
@@ -920,6 +920,43 @@ async def test_submitted_local_trigger_turn_reports_pending_until_completed(tmp_
|
|||||||
assert loop.pending_local_trigger_ids_for_session(session_key) == set()
|
assert loop.pending_local_trigger_ids_for_session(session_key) == set()
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_local_trigger_turn_cancellation_reports_agent_failure(tmp_path):
|
||||||
|
"""A cancelled agent turn should not cancel the local-trigger worker."""
|
||||||
|
from nanobot.agent.automation_turns import AutomationTurnError
|
||||||
|
from nanobot.bus.events import InboundMessage
|
||||||
|
from nanobot.triggers.local_session_turns import LOCAL_TRIGGER_META
|
||||||
|
|
||||||
|
loop = _make_loop(tmp_path)
|
||||||
|
loop._running = True
|
||||||
|
|
||||||
|
session_key = "websocket:chat-1"
|
||||||
|
msg = InboundMessage(
|
||||||
|
channel="websocket",
|
||||||
|
sender_id="trigger",
|
||||||
|
chat_id="chat-1",
|
||||||
|
content="review failed CI",
|
||||||
|
metadata={
|
||||||
|
LOCAL_TRIGGER_META: {
|
||||||
|
"trigger_id": "trg_123",
|
||||||
|
"trigger_name": "CI review",
|
||||||
|
"delivery_id": "tdl_123",
|
||||||
|
},
|
||||||
|
},
|
||||||
|
session_key_override=session_key,
|
||||||
|
)
|
||||||
|
|
||||||
|
submit_task = asyncio.create_task(loop.submit_local_trigger_turn(msg))
|
||||||
|
assert await asyncio.wait_for(loop.bus.consume_inbound(), timeout=0.5) is msg
|
||||||
|
|
||||||
|
loop._local_trigger_turns.complete(msg, error=asyncio.CancelledError())
|
||||||
|
|
||||||
|
with pytest.raises(AutomationTurnError, match="CancelledError"):
|
||||||
|
await asyncio.wait_for(submit_task, timeout=0.5)
|
||||||
|
assert not submit_task.cancelled()
|
||||||
|
assert loop.pending_local_trigger_ids_for_session(session_key) == set()
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_pending_queue_preserves_overflow_for_next_injection_cycle(tmp_path):
|
async def test_pending_queue_preserves_overflow_for_next_injection_cycle(tmp_path):
|
||||||
"""Pending queue should leave overflow messages queued for later drains."""
|
"""Pending queue should leave overflow messages queued for later drains."""
|
||||||
|
|||||||
Reference in New Issue
Block a user