fix(webui): clear stale run status on reconnect
This commit is contained in:
@@ -346,11 +346,7 @@ class WebSocketChannel(BaseChannel):
|
||||
async def _hydrate_after_subscribe(self, chat_id: str) -> None:
|
||||
"""Replay goal/run strip state after subscribe (same-process refresh)."""
|
||||
await self._maybe_push_active_goal_state(chat_id)
|
||||
t0 = websocket_turn_wall_started_at(chat_id)
|
||||
if t0 is not None:
|
||||
await self.send_goal_status(chat_id, "running", started_at=t0)
|
||||
else:
|
||||
await self.send_goal_status(chat_id, "idle")
|
||||
await self._maybe_push_turn_run_wall_clock(chat_id)
|
||||
|
||||
async def _send_event(self, connection: Any, event: str, **fields: Any) -> None:
|
||||
"""Send a control event (attached, error, ...) to a single connection."""
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
"""Test websocket reconnect pushes idle status when no turn is active."""
|
||||
"""Test websocket subscribe hydration only replays known active turns."""
|
||||
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
@@ -8,29 +8,28 @@ from nanobot.channels.websocket import WebSocketChannel
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_hydrate_after_subscribe_pushes_idle_when_no_turn_active():
|
||||
"""Reconnecting client should receive idle status when no turn is running."""
|
||||
async def test_hydrate_after_subscribe_is_quiet_when_no_turn_active():
|
||||
"""Subscribe hydration must not inject an idle event into normal message order."""
|
||||
channel = WebSocketChannel.__new__(WebSocketChannel)
|
||||
channel.gateway = MagicMock()
|
||||
channel.gateway.session_manager = MagicMock()
|
||||
channel.gateway.session_manager.read_session_file = MagicMock(return_value={})
|
||||
|
||||
|
||||
sent_events = []
|
||||
|
||||
|
||||
async def mock_send_goal_state(chat_id, blob):
|
||||
sent_events.append(("goal_state", chat_id, blob))
|
||||
|
||||
|
||||
async def mock_send_goal_status(chat_id, status, **kwargs):
|
||||
sent_events.append(("goal_status", chat_id, status, kwargs))
|
||||
|
||||
|
||||
channel.send_goal_state = mock_send_goal_state
|
||||
channel.send_goal_status = mock_send_goal_status
|
||||
|
||||
|
||||
with patch("nanobot.channels.websocket.websocket_turn_wall_started_at", return_value=None):
|
||||
await channel._hydrate_after_subscribe("test-chat")
|
||||
|
||||
# Should have pushed idle status
|
||||
assert any(e[0] == "goal_status" and e[2] == "idle" for e in sent_events)
|
||||
|
||||
assert sent_events == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@@ -40,22 +39,21 @@ async def test_hydrate_after_subscribe_pushes_running_when_turn_active():
|
||||
channel.gateway = MagicMock()
|
||||
channel.gateway.session_manager = MagicMock()
|
||||
channel.gateway.session_manager.read_session_file = MagicMock(return_value={})
|
||||
|
||||
|
||||
sent_events = []
|
||||
|
||||
|
||||
async def mock_send_goal_state(chat_id, blob):
|
||||
sent_events.append(("goal_state", chat_id, blob))
|
||||
|
||||
|
||||
async def mock_send_goal_status(chat_id, status, **kwargs):
|
||||
sent_events.append(("goal_status", chat_id, status, kwargs))
|
||||
|
||||
|
||||
channel.send_goal_state = mock_send_goal_state
|
||||
channel.send_goal_status = mock_send_goal_status
|
||||
|
||||
|
||||
with patch("nanobot.channels.websocket.websocket_turn_wall_started_at", return_value=1234567890.0):
|
||||
await channel._hydrate_after_subscribe("test-chat")
|
||||
|
||||
# Should have pushed running status with started_at
|
||||
|
||||
running_events = [e for e in sent_events if e[0] == "goal_status" and e[2] == "running"]
|
||||
assert len(running_events) == 1
|
||||
assert running_events[0][3]["started_at"] == 1234567890.0
|
||||
|
||||
@@ -16,12 +16,12 @@ async def test_cmd_stop_drains_pending_queue():
|
||||
mock_loop = MagicMock()
|
||||
mock_loop._cancel_active_tasks = AsyncMock(return_value=1)
|
||||
mock_loop._pending_queues = {}
|
||||
|
||||
|
||||
pending = asyncio.Queue()
|
||||
await pending.put("msg1")
|
||||
await pending.put("msg2")
|
||||
mock_loop._pending_queues["test-session"] = pending
|
||||
|
||||
|
||||
ctx = CommandContext(
|
||||
msg=MagicMock(channel="websocket", chat_id="test-chat", metadata={}),
|
||||
session=None,
|
||||
@@ -29,9 +29,9 @@ async def test_cmd_stop_drains_pending_queue():
|
||||
raw="/stop",
|
||||
loop=mock_loop,
|
||||
)
|
||||
|
||||
|
||||
result = await cmd_stop(ctx)
|
||||
|
||||
|
||||
assert isinstance(result, OutboundMessage)
|
||||
assert "Stopped 3 task(s)" in result.content # 1 cancelled + 2 drained
|
||||
assert "test-session" not in mock_loop._pending_queues
|
||||
@@ -43,10 +43,10 @@ async def test_cmd_stop_with_empty_pending_queue():
|
||||
mock_loop = MagicMock()
|
||||
mock_loop._cancel_active_tasks = AsyncMock(return_value=2)
|
||||
mock_loop._pending_queues = {}
|
||||
|
||||
|
||||
pending = asyncio.Queue()
|
||||
mock_loop._pending_queues["test-session"] = pending
|
||||
|
||||
|
||||
ctx = CommandContext(
|
||||
msg=MagicMock(channel="websocket", chat_id="test-chat", metadata={}),
|
||||
session=None,
|
||||
@@ -54,9 +54,9 @@ async def test_cmd_stop_with_empty_pending_queue():
|
||||
raw="/stop",
|
||||
loop=mock_loop,
|
||||
)
|
||||
|
||||
|
||||
result = await cmd_stop(ctx)
|
||||
|
||||
|
||||
assert "Stopped 2 task(s)" in result.content
|
||||
assert "test-session" not in mock_loop._pending_queues
|
||||
|
||||
@@ -67,7 +67,7 @@ async def test_cmd_stop_no_pending_queue():
|
||||
mock_loop = MagicMock()
|
||||
mock_loop._cancel_active_tasks = AsyncMock(return_value=0)
|
||||
mock_loop._pending_queues = {}
|
||||
|
||||
|
||||
ctx = CommandContext(
|
||||
msg=MagicMock(channel="websocket", chat_id="test-chat", metadata={}),
|
||||
session=None,
|
||||
@@ -75,7 +75,7 @@ async def test_cmd_stop_no_pending_queue():
|
||||
raw="/stop",
|
||||
loop=mock_loop,
|
||||
)
|
||||
|
||||
|
||||
result = await cmd_stop(ctx)
|
||||
|
||||
|
||||
assert "No active task to stop" in result.content
|
||||
|
||||
@@ -425,6 +425,13 @@ export class NanobotClient {
|
||||
for (const handler of this.statusHandlers) handler(status);
|
||||
}
|
||||
|
||||
private clearRunStatusesForReconnect(): void {
|
||||
if (this.runStartedAtByChatId.size === 0) return;
|
||||
const chatIds = [...this.runStartedAtByChatId.keys()];
|
||||
this.runStartedAtByChatId.clear();
|
||||
for (const chatId of chatIds) this.emitRunStatus(chatId, null);
|
||||
}
|
||||
|
||||
private handleOpen(): void {
|
||||
this.setStatus("open");
|
||||
this.reconnectAttempts = 0;
|
||||
@@ -629,6 +636,7 @@ export class NanobotClient {
|
||||
}
|
||||
|
||||
private scheduleReconnect(): void {
|
||||
this.clearRunStatusesForReconnect();
|
||||
this.setStatus("reconnecting");
|
||||
const attempt = this.reconnectAttempts++;
|
||||
// Exponential backoff: 0.5s, 1s, 2s, 4s, capped.
|
||||
|
||||
@@ -188,6 +188,32 @@ describe("NanobotClient", () => {
|
||||
expect(client.getRunStartedAt("chat-strip")).toBeNull();
|
||||
});
|
||||
|
||||
it("clears stale run strip when reconnecting after a dropped socket", async () => {
|
||||
const client = new NanobotClient({
|
||||
url: "ws://test",
|
||||
reconnect: true,
|
||||
maxBackoffMs: 10,
|
||||
socketFactory: (url) => new FakeSocket(url) as unknown as WebSocket,
|
||||
});
|
||||
const handler = vi.fn();
|
||||
client.onRunStatus(handler);
|
||||
client.connect();
|
||||
lastSocket().fakeOpen();
|
||||
lastSocket().fakeMessage({
|
||||
event: "goal_status",
|
||||
chat_id: "chat-strip",
|
||||
status: "running",
|
||||
started_at: 12_345,
|
||||
});
|
||||
|
||||
lastSocket().close();
|
||||
|
||||
expect(client.getRunStartedAt("chat-strip")).toBeNull();
|
||||
expect(handler).toHaveBeenLastCalledWith("chat-strip", null);
|
||||
await vi.advanceTimersByTimeAsync(20);
|
||||
expect(FakeSocket.instances.length).toBeGreaterThan(1);
|
||||
});
|
||||
|
||||
it("clears run strip when a turn_end arrives without idle", () => {
|
||||
const client = new NanobotClient({
|
||||
url: "ws://test",
|
||||
|
||||
Reference in New Issue
Block a user