diff --git a/open_strix/app.py b/open_strix/app.py index 5425393..bf0f54a 100644 --- a/open_strix/app.py +++ b/open_strix/app.py @@ -844,6 +844,60 @@ def _validate_memory_blocks(self) -> list[str]: return errors async def _process_event(self, event: AgentEvent) -> None: + await self._run_turn(event, batched=False) + + # Part 2 of #91: end-of-turn batching layer. After responding on a + # Discord channel, drain any same-channel discord_message events that + # arrived while the first turn was running and process them in one + # follow-up turn so bursts cost 2 turns instead of N. + if event.event_type != "discord_message" or not event.channel_id: + return + drained = self._drain_same_channel_discord_events(event.channel_id) + if not drained: + return + self.log_event( + "batched_turn_start", + channel_id=event.channel_id, + trigger_event_type=event.event_type, + batch_size=len(drained), + ) + try: + # Newest drained event is the trigger; _render_prompt already pulls + # the full recent-channel history (including drained messages). + await self._run_turn(drained[-1], batched=True) + finally: + for drained_event in drained: + if drained_event.dedupe_key: + self.pending_scheduler_keys.discard(drained_event.dedupe_key) + self.queue.task_done() + + def _drain_same_channel_discord_events( + self, channel_id: str + ) -> list[AgentEvent]: + """Pop all queued discord_message events for ``channel_id`` while + preserving other queued events in their original order. + + Mutates the queue's backing deque directly so ``_unfinished_tasks`` + accounting stays intact — caller is responsible for calling + ``self.queue.task_done()`` once per drained event.""" + internal: deque[AgentEvent] = self.queue._queue # type: ignore[attr-defined] + if not internal: + return [] + drained: list[AgentEvent] = [] + kept: deque[AgentEvent] = deque() + while internal: + candidate = internal.popleft() + if ( + candidate.event_type == "discord_message" + and candidate.channel_id == channel_id + ): + drained.append(candidate) + else: + kept.append(candidate) + internal.extend(kept) + return drained + + async def _run_turn(self, event: AgentEvent, *, batched: bool) -> None: self._current_turn_sent_messages = [] self._reset_send_message_circuit_breaker() # Turn-time instrumentation (#91): baseline measurement that lets the @@ -940,6 +994,7 @@ async def _process_event(self, event: AgentEvent) -> None: scheduler_name=event.scheduler_name, total_seconds=round(time.monotonic() - turn_start, 4), repair_invoke_count=repair_invoke_count, + batched=batched, **rounded, ) diff --git a/tests/test_batched_processing.py b/tests/test_batched_processing.py new file mode 100644 index 0000000..296acf5 --- /dev/null +++ b/tests/test_batched_processing.py @@ -0,0 +1,157 @@ +"""Tests for end-of-turn batching of same-channel discord messages (#91 part 2).""" +from __future__ import annotations + +import asyncio +from pathlib import Path +from typing import Any + +from langchain_core.messages import AIMessage + +import open_strix.app as app_mod + + +class CountingAgent: + def __init__(self) -> None: + self.invoke_count = 0 + + async def ainvoke(self, _: dict[str, Any]) -> dict[str, Any]: + self.invoke_count += 1 + return {"messages": [AIMessage(content="ok")]} + + +def _build_app(tmp_path: Path, monkeypatch, *, agent: CountingAgent) -> app_mod.OpenStrixApp: + monkeypatch.setattr(app_mod, "create_deep_agent", lambda **_: agent) + app = app_mod.OpenStrixApp(tmp_path) + + async def _noop_git(_event: Any) -> str: + return "skip: test" + + app._run_post_turn_git_sync = _noop_git # type: ignore[assignment] + return app + + +def _collect_events(app: app_mod.OpenStrixApp) -> list[dict[str, Any]]: + calls: list[dict[str, Any]] = [] + original = app.log_event + + def _capture(event_type: str, **payload: Any) -> None: + calls.append({"type": event_type, **payload}) + original(event_type, **payload) + + app.log_event = _capture # type: ignore[assignment] + return calls + + +def _discord_event(channel_id: str, author: str, prompt: str) -> app_mod.AgentEvent: + return app_mod.AgentEvent( + event_type="discord_message", + prompt=prompt, + channel_id=channel_id, + author=author, + ) + + +def test_same_channel_messages_drain_into_one_batched_turn( + tmp_path: Path, monkeypatch +) -> None: + agent = CountingAgent() + app = _build_app(tmp_path, monkeypatch, agent=agent) + events = _collect_events(app) + + trigger = _discord_event("chan-1", "alice", "first") + followup_a = _discord_event("chan-1", "bob", "second") + followup_b = _discord_event("chan-1", "carol", "third") + other_channel = _discord_event("chan-2", "dave", "other") + app.queue.put_nowait(followup_a) + app.queue.put_nowait(followup_b) + app.queue.put_nowait(other_channel) + + asyncio.run(app._process_event(trigger)) + + # Two agent invocations: one per turn (trigger turn + batched turn). + assert agent.invoke_count == 2 + + timing_events = [e for e in events if e["type"] == "turn_timing"] + assert len(timing_events) == 2 + assert timing_events[0]["batched"] is False + assert timing_events[1]["batched"] is True + + batch_starts = [e for e in events if e["type"] == "batched_turn_start"] + assert len(batch_starts) == 1 + assert batch_starts[0]["channel_id"] == "chan-1" + assert batch_starts[0]["batch_size"] == 2 + + # Other-channel event stays in queue untouched. + assert app.queue.qsize() == 1 + remaining = app.queue.get_nowait() + assert remaining is other_channel + app.queue.task_done() + + +def test_no_batched_turn_when_queue_has_no_same_channel_messages( + tmp_path: Path, monkeypatch +) -> None: + agent = CountingAgent() + app = _build_app(tmp_path, monkeypatch, agent=agent) + events = _collect_events(app) + + trigger = _discord_event("chan-1", "alice", "only one") + other_channel = _discord_event("chan-2", "bob", "unrelated") + app.queue.put_nowait(other_channel) + + asyncio.run(app._process_event(trigger)) + + assert agent.invoke_count == 1 + timing_events = [e for e in events if e["type"] == "turn_timing"] + assert len(timing_events) == 1 + assert timing_events[0]["batched"] is False + assert not any(e["type"] == "batched_turn_start" for e in events) + + # Untouched. + assert app.queue.qsize() == 1 + + +def test_non_discord_trigger_does_not_batch(tmp_path: Path, monkeypatch) -> None: + agent = CountingAgent() + app = _build_app(tmp_path, monkeypatch, agent=agent) + events = _collect_events(app) + + trigger = app_mod.AgentEvent( + event_type="scheduler_tick", + prompt="tick", + channel_id="chan-1", + scheduler_name="perch", + ) + # Even if a matching discord_message is queued, a scheduler tick should + # not absorb it — only a discord_message trigger should drain. + queued = _discord_event("chan-1", "alice", "hey") + app.queue.put_nowait(queued) + + asyncio.run(app._process_event(trigger)) + + assert agent.invoke_count == 1 + assert not any(e["type"] == "batched_turn_start" for e in events) + assert app.queue.qsize() == 1 + + +def test_drain_helper_preserves_order_of_kept_events( + tmp_path: Path, monkeypatch +) -> None: + agent = CountingAgent() + app = _build_app(tmp_path, monkeypatch, agent=agent) + + a = _discord_event("chan-1", "alice", "a") + b = _discord_event("chan-2", "bob", "b") + c = _discord_event("chan-1", "carol", "c") + d = _discord_event("chan-3", "dave", "d") + for event in (a, b, c, d): + app.queue.put_nowait(event) + + drained = app._drain_same_channel_discord_events("chan-1") + + assert [e.prompt for e in drained] == ["a", "c"] + remaining: list[app_mod.AgentEvent] = [] + while app.queue.qsize(): + remaining.append(app.queue.get_nowait()) + app.queue.task_done() + assert [e.prompt for e in remaining] == ["b", "d"]