fix(triggers): reject deliveries to disabled channels
This commit is contained in:
@@ -2076,6 +2076,7 @@ def _run_gateway(
|
|||||||
run_local_trigger_queue(
|
run_local_trigger_queue(
|
||||||
store=trigger_store,
|
store=trigger_store,
|
||||||
submit_turn=getattr(agent, "submit_local_trigger_turn", None),
|
submit_turn=getattr(agent, "submit_local_trigger_turn", None),
|
||||||
|
is_channel_available=lambda name: name in channels.channels,
|
||||||
),
|
),
|
||||||
name="nanobot-local-triggers",
|
name="nanobot-local-triggers",
|
||||||
),
|
),
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ async def run_local_trigger_queue(
|
|||||||
*,
|
*,
|
||||||
store: LocalTriggerStore,
|
store: LocalTriggerStore,
|
||||||
submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]] | None = None,
|
submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]] | None = None,
|
||||||
|
is_channel_available: Callable[[str], bool] | None = None,
|
||||||
poll_interval_s: float = 0.5,
|
poll_interval_s: float = 0.5,
|
||||||
batch_size: int = 20,
|
batch_size: int = 20,
|
||||||
) -> None:
|
) -> None:
|
||||||
@@ -46,6 +47,7 @@ async def run_local_trigger_queue(
|
|||||||
store,
|
store,
|
||||||
delivery,
|
delivery,
|
||||||
submit_turn=submit_turn,
|
submit_turn=submit_turn,
|
||||||
|
is_channel_available=is_channel_available,
|
||||||
)
|
)
|
||||||
store.complete_delivery(delivery)
|
store.complete_delivery(delivery)
|
||||||
except asyncio.CancelledError as exc:
|
except asyncio.CancelledError as exc:
|
||||||
@@ -130,12 +132,15 @@ async def _deliver_delivery(
|
|||||||
delivery: TriggerDelivery,
|
delivery: TriggerDelivery,
|
||||||
*,
|
*,
|
||||||
submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]],
|
submit_turn: Callable[[InboundMessage], Awaitable[OutboundMessage | None]],
|
||||||
|
is_channel_available: Callable[[str], bool] | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
trigger = store.get(delivery.trigger_id)
|
trigger = store.get(delivery.trigger_id)
|
||||||
if trigger is None:
|
if trigger is None:
|
||||||
raise _TerminalDeliveryError("trigger not found")
|
raise _TerminalDeliveryError("trigger not found")
|
||||||
if not trigger.enabled:
|
if not trigger.enabled:
|
||||||
raise _TerminalDeliveryError("trigger is disabled")
|
raise _TerminalDeliveryError("trigger is disabled")
|
||||||
|
if is_channel_available is not None and not is_channel_available(trigger.channel):
|
||||||
|
raise _TerminalDeliveryError(f"target channel is not enabled: {trigger.channel}")
|
||||||
|
|
||||||
store.write_delivery_run_record(delivery, trigger=trigger, status="processing")
|
store.write_delivery_run_record(delivery, trigger=trigger, status="processing")
|
||||||
msg = InboundMessage(
|
msg = InboundMessage(
|
||||||
|
|||||||
@@ -2748,7 +2748,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
|
|||||||
enabled_channels: list[str] = []
|
enabled_channels: list[str] = []
|
||||||
|
|
||||||
def __init__(self, *_args, **_kwargs) -> None:
|
def __init__(self, *_args, **_kwargs) -> None:
|
||||||
return None
|
self.channels: dict[str, object] = {}
|
||||||
|
|
||||||
async def start_all(self) -> None:
|
async def start_all(self) -> None:
|
||||||
await asyncio.Event().wait()
|
await asyncio.Event().wait()
|
||||||
@@ -2776,6 +2776,7 @@ def test_gateway_local_trigger_queue_submits_agent_turns(
|
|||||||
assert kwargs["store"] is agent_kwargs["local_trigger_store"]
|
assert kwargs["store"] is agent_kwargs["local_trigger_store"]
|
||||||
assert "bus" not in kwargs
|
assert "bus" not in kwargs
|
||||||
assert kwargs["submit_turn"] is agent.submit_local_trigger_turn
|
assert kwargs["submit_turn"] is agent.submit_local_trigger_turn
|
||||||
|
assert kwargs["is_channel_available"]("websocket") is False
|
||||||
|
|
||||||
|
|
||||||
def test_gateway_workspace_override_does_not_migrate_legacy_cron(
|
def test_gateway_workspace_override_does_not_migrate_legacy_cron(
|
||||||
|
|||||||
@@ -299,6 +299,52 @@ async def test_local_trigger_queue_submits_bound_inbound_message(tmp_path: Path)
|
|||||||
assert record["trigger_id"] == trigger.id
|
assert record["trigger_id"] == trigger.id
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_local_trigger_queue_rejects_unavailable_target_channel(tmp_path: Path) -> None:
|
||||||
|
store = LocalTriggerStore(tmp_path)
|
||||||
|
trigger = store.create(
|
||||||
|
name="PR review",
|
||||||
|
channel="telegram",
|
||||||
|
chat_id="chat-1",
|
||||||
|
session_key="telegram:chat-1",
|
||||||
|
)
|
||||||
|
delivery = store.enqueue(trigger.id, "Review PR #4502")
|
||||||
|
submitted: list[InboundMessage] = []
|
||||||
|
|
||||||
|
async def _submit_turn(msg: InboundMessage):
|
||||||
|
submitted.append(msg)
|
||||||
|
return OutboundMessage(channel=msg.channel, chat_id=msg.chat_id, content="done")
|
||||||
|
|
||||||
|
task = asyncio.create_task(
|
||||||
|
run_local_trigger_queue(
|
||||||
|
store=store,
|
||||||
|
submit_turn=_submit_turn,
|
||||||
|
is_channel_available=lambda name: name == "websocket",
|
||||||
|
poll_interval_s=0.01,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
try:
|
||||||
|
for _ in range(100):
|
||||||
|
stored = store.get(trigger.id)
|
||||||
|
if stored and stored.last_status == "error":
|
||||||
|
break
|
||||||
|
await asyncio.sleep(0.01)
|
||||||
|
finally:
|
||||||
|
task.cancel()
|
||||||
|
with suppress(asyncio.CancelledError):
|
||||||
|
await task
|
||||||
|
|
||||||
|
stored = store.get(trigger.id)
|
||||||
|
assert submitted == []
|
||||||
|
assert stored is not None
|
||||||
|
assert stored.last_status == "error"
|
||||||
|
assert stored.last_error == "target channel is not enabled: telegram"
|
||||||
|
assert store.claim_deliveries() == []
|
||||||
|
record = _read_run_record(store, delivery.id)
|
||||||
|
assert record["status"] == "error"
|
||||||
|
assert record["error"] == "target channel is not enabled: telegram"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
@pytest.mark.asyncio
|
||||||
async def test_local_trigger_queue_waits_for_submitted_turn_before_ack(
|
async def test_local_trigger_queue_waits_for_submitted_turn_before_ack(
|
||||||
tmp_path: Path,
|
tmp_path: Path,
|
||||||
|
|||||||
Reference in New Issue
Block a user