mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-09-11 04:07:44 +02:00
[eric] faster replies + smarter chat: optimistic message bubbles, live thinking... pill, friendlier tool-call labels (Reading → Read),
compaction chip, native completion notifications, prompt-cache flip for ~70% cheaper/faster
This commit is contained in:
ciregenz
@@ -157,6 +157,23 @@ class AgentLoop:
|
||||
"tool_name": event.tool_name,
|
||||
})
|
||||
|
||||
elif event.block_type == "thinking":
|
||||
# Extended-thinking content block. Emit a distinct
|
||||
# WS stream with role="thinking" so the frontend
|
||||
# renders the live ThinkingBubble pill (rising
|
||||
# token counter, auto-collapse on first text). Each
|
||||
# thinking block gets its own message id — multiple
|
||||
# interleaved thinking/text blocks remain
|
||||
# individually addressable.
|
||||
thinking_msg_id = uuid4().hex
|
||||
block_index_map[event.index] = thinking_msg_id
|
||||
block_types[event.index] = "thinking"
|
||||
text_buffers[event.index] = ""
|
||||
await self.ws_emitter("agent:stream_start", {
|
||||
"message_id": thinking_msg_id,
|
||||
"role": "thinking",
|
||||
})
|
||||
|
||||
elif event.type == "content_block_delta":
|
||||
msg_id = block_index_map.get(event.index)
|
||||
if not msg_id:
|
||||
@@ -178,6 +195,16 @@ class AgentLoop:
|
||||
"delta": event.text,
|
||||
})
|
||||
|
||||
elif event.delta_type == "thinking_delta":
|
||||
# Reuse the text buffer for thinking — same shape
|
||||
# (accumulated str), different sink.
|
||||
text_buffers.setdefault(event.index, "")
|
||||
text_buffers[event.index] += event.text
|
||||
await self.ws_emitter("agent:stream_delta", {
|
||||
"message_id": msg_id,
|
||||
"delta": event.text,
|
||||
})
|
||||
|
||||
elif event.type == "content_block_stop":
|
||||
msg_id = block_index_map.get(event.index)
|
||||
bt = block_types.get(event.index, "")
|
||||
@@ -199,9 +226,17 @@ class AgentLoop:
|
||||
input=tool_input,
|
||||
),
|
||||
))
|
||||
elif bt == "thinking":
|
||||
collected_content.append(
|
||||
ContentBlock(type="thinking", text=text_buffers.get(event.index, ""))
|
||||
)
|
||||
|
||||
# Send stream_end for tool blocks (text block ends at message_stop)
|
||||
if msg_id and bt == "tool_use":
|
||||
# Send stream_end for tool + thinking blocks (text block
|
||||
# ends at message_stop). Thinking ends here so the
|
||||
# frontend can transition the pill from "live" to
|
||||
# "Thought for Ns" the moment the model stops thinking,
|
||||
# even if it then keeps streaming text.
|
||||
if msg_id and (bt == "tool_use" or bt == "thinking"):
|
||||
await self.ws_emitter("agent:stream_end", {
|
||||
"message_id": msg_id,
|
||||
})
|
||||
@@ -243,6 +278,21 @@ class AgentLoop:
|
||||
"""Emit finalized agent:message events for the collected response."""
|
||||
from backend.apps.agents.models import Message
|
||||
|
||||
# Emit thinking blocks (extended thinking). Persisted as their own
|
||||
# messages so a session reload still shows the reasoning trail.
|
||||
# Multiple thinking blocks per turn are concatenated into a single
|
||||
# persisted message — the streaming UI already showed each block
|
||||
# individually, this is just for the historical record.
|
||||
thinking_parts = [b.text for b in content if b.type == "thinking" and b.text]
|
||||
if thinking_parts:
|
||||
msg = Message(
|
||||
role="thinking",
|
||||
content="\n\n".join(thinking_parts),
|
||||
)
|
||||
await self.ws_emitter("agent:message", {
|
||||
"message": msg.model_dump(mode="json"),
|
||||
})
|
||||
|
||||
# Emit text message
|
||||
text_parts = [b.text for b in content if b.type == "text" and b.text]
|
||||
if text_parts:
|
||||
|
||||
@@ -2072,16 +2072,38 @@ class AgentManager:
|
||||
"type": "preset",
|
||||
"preset": "claude_code",
|
||||
}
|
||||
# exclude_dynamic_sections=True tells the CLI to keep
|
||||
# per-user/per-machine grounding (cwd, git status, recent
|
||||
# commits, OS info) out of the cached system prompt prefix
|
||||
# and re-inject it into the first user message instead. This
|
||||
# makes the prefix byte-identical across users + sessions,
|
||||
# which is what unlocks Anthropic's prompt cache (turn 2+
|
||||
# gets a cache hit, ~80% input-token cost cut and 13–31%
|
||||
# faster TTFT). The grounding info still reaches the model;
|
||||
# only its position in the wire format changes.
|
||||
#
|
||||
# Trade-off: dynamic sections freeze at turn 1 — branch
|
||||
# switches / large workspace state changes mid-session won't
|
||||
# refresh until a new session starts. Acceptable for a
|
||||
# multi-purpose agent canvas where most sessions aren't
|
||||
# long-running coding marathons; coding-specific modes
|
||||
# (view-builder, skill-builder) can flip this off later if
|
||||
# we see drift complaints.
|
||||
#
|
||||
# Older bundled CLIs silently ignore the flag, so this is
|
||||
# forward-safe; no version gate needed.
|
||||
if composed_prompt:
|
||||
options_kwargs["system_prompt"] = {
|
||||
"type": "preset",
|
||||
"preset": "claude_code",
|
||||
"append": composed_prompt,
|
||||
"exclude_dynamic_sections": True,
|
||||
}
|
||||
else:
|
||||
options_kwargs["system_prompt"] = {
|
||||
"type": "preset",
|
||||
"preset": "claude_code",
|
||||
"exclude_dynamic_sections": True,
|
||||
}
|
||||
if session.max_turns:
|
||||
options_kwargs["max_turns"] = session.max_turns
|
||||
@@ -2786,6 +2808,7 @@ class AgentManager:
|
||||
attached_skills: list | None = None,
|
||||
hidden: bool = False,
|
||||
selected_browser_ids: list[str] | None = None,
|
||||
client_message_id: str | None = None,
|
||||
):
|
||||
"""Send a follow-up message to an existing session."""
|
||||
session = self.sessions.get(session_id)
|
||||
@@ -2855,6 +2878,7 @@ class AgentManager:
|
||||
forced_tools=forced_tools if forced_tools else None,
|
||||
images=image_meta,
|
||||
hidden=hidden,
|
||||
client_message_id=client_message_id,
|
||||
)
|
||||
session.messages.append(user_msg)
|
||||
await ws_manager.send_to_session(session_id, "agent:message", {
|
||||
|
||||
@@ -87,6 +87,7 @@ async def send_message(session_id: str, body: dict):
|
||||
attached_skills=body.get("attached_skills"),
|
||||
hidden=body.get("hidden", False),
|
||||
selected_browser_ids=body.get("selected_browser_ids"),
|
||||
client_message_id=body.get("client_message_id"),
|
||||
)
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@@ -39,6 +39,11 @@ class Message(BaseModel):
|
||||
forced_tools: Optional[list[str]] = None
|
||||
images: Optional[list[dict]] = None
|
||||
hidden: bool = False
|
||||
# Optional client-generated id used by the frontend to reconcile an
|
||||
# optimistic message bubble (rendered synchronously on send) with the
|
||||
# server-confirmed echo. Plumbed through send_message and round-tripped
|
||||
# back via the agent:message WS event so the frontend can dedupe.
|
||||
client_message_id: Optional[str] = None
|
||||
|
||||
class MessageBranch(BaseModel):
|
||||
id: str = Field(default_factory=lambda: uuid4().hex)
|
||||
|
||||
@@ -188,6 +188,18 @@ class AnthropicProvider(BaseProvider):
|
||||
tool_name=block.name,
|
||||
tool_id=block.id,
|
||||
)
|
||||
elif block_type == "thinking":
|
||||
# Extended-thinking content block. We track the
|
||||
# accumulated text in current_text just like a normal
|
||||
# text block, but tag it as "thinking" so the agent
|
||||
# loop emits a distinct WS event the frontend can
|
||||
# render in the ThinkingBubble pill.
|
||||
current_text[index] = ""
|
||||
yield StreamEvent(
|
||||
type="content_block_start",
|
||||
index=index,
|
||||
block_type="thinking",
|
||||
)
|
||||
|
||||
elif event_type == "content_block_delta":
|
||||
index = event.index
|
||||
@@ -212,6 +224,24 @@ class AnthropicProvider(BaseProvider):
|
||||
delta_type="input_json_delta",
|
||||
text=delta.partial_json,
|
||||
)
|
||||
elif delta_type == "thinking_delta":
|
||||
# Extended-thinking text streamed as it's produced.
|
||||
# Forward as a thinking_delta so the agent loop can
|
||||
# ship it to the frontend without conflating with
|
||||
# the assistant text stream.
|
||||
text_chunk = getattr(delta, "thinking", "") or ""
|
||||
current_text.setdefault(index, "")
|
||||
current_text[index] += text_chunk
|
||||
yield StreamEvent(
|
||||
type="content_block_delta",
|
||||
index=index,
|
||||
delta_type="thinking_delta",
|
||||
text=text_chunk,
|
||||
)
|
||||
# Note: signature_delta (the cryptographic signature on
|
||||
# thinking blocks) is intentionally ignored — we don't
|
||||
# display it and it isn't needed for replay since we
|
||||
# never re-send thinking blocks to the model.
|
||||
|
||||
elif event_type == "content_block_stop":
|
||||
yield StreamEvent(type="content_block_stop", index=event.index)
|
||||
|
||||
@@ -31,7 +31,7 @@ class ToolCall:
|
||||
@dataclass
|
||||
class ContentBlock:
|
||||
"""A block of content from the model response."""
|
||||
type: str # "text" | "tool_use"
|
||||
type: str # "text" | "tool_use" | "thinking"
|
||||
text: str = ""
|
||||
tool_call: ToolCall | None = None
|
||||
|
||||
@@ -53,8 +53,8 @@ class StreamEvent:
|
||||
"""
|
||||
type: str
|
||||
index: int = 0
|
||||
block_type: str = "" # "text" | "tool_use"
|
||||
delta_type: str = "" # "text_delta" | "input_json_delta"
|
||||
block_type: str = "" # "text" | "tool_use" | "thinking"
|
||||
delta_type: str = "" # "text_delta" | "input_json_delta" | "thinking_delta"
|
||||
text: str = ""
|
||||
tool_name: str = ""
|
||||
tool_id: str = ""
|
||||
|
||||
@@ -0,0 +1,594 @@
|
||||
"""Stress tests for the Phase 1 / 2 / 3 perceived-latency changes.
|
||||
|
||||
Hits everything we touched on the eric/v2 branch:
|
||||
|
||||
- Message.client_message_id round-trip (optimistic dedupe)
|
||||
- Mode migration: 'chat' -> 'ask' on session reconcile + lifespan
|
||||
deletion of stale built-in chat.json
|
||||
- ContentBlock + StreamEvent now accept type='thinking' /
|
||||
delta_type='thinking_delta' without breaking existing types
|
||||
- Anthropic provider forwards thinking content_block_start /
|
||||
content_block_delta with the right shape
|
||||
- Agent loop emits agent:stream_start{role:'thinking'},
|
||||
agent:stream_delta, agent:stream_end for thinking blocks AND
|
||||
persists a Message(role='thinking') after stream end
|
||||
- DashboardLayout serializes notes round-trip
|
||||
- exclude_dynamic_sections reaches the SDK kwargs (presence-only;
|
||||
we don't run the real CLI here)
|
||||
|
||||
Each test runs many randomized iterations to surface race conditions
|
||||
and bad assumptions. Stub the network and CLI throughout — these
|
||||
tests are pure logic, no real Anthropic calls.
|
||||
|
||||
Run:
|
||||
cd backend && .venv/bin/python -m pytest tests/test_phase1_stress.py -v
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import random
|
||||
import string
|
||||
import tempfile
|
||||
from typing import Any
|
||||
from unittest.mock import patch, AsyncMock
|
||||
|
||||
import pytest
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Boot env: route data dirs into a tmp scratch root before importing
|
||||
# backend modules.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_TMPROOT = tempfile.mkdtemp(prefix="openswarm-phase1-stress-")
|
||||
os.environ.setdefault("OPENSWARM_DATA_DIR", _TMPROOT)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 1 — Message.client_message_id
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_message_round_trips_client_id():
|
||||
"""The new field must default to None and survive model_dump."""
|
||||
from backend.apps.agents.models import Message
|
||||
|
||||
m = Message(role="user", content="hi")
|
||||
assert m.client_message_id is None
|
||||
|
||||
dumped = m.model_dump(mode="json")
|
||||
assert "client_message_id" in dumped
|
||||
assert dumped["client_message_id"] is None
|
||||
|
||||
m2 = Message(role="user", content="hi", client_message_id="opt-abc-123")
|
||||
dumped2 = m2.model_dump(mode="json")
|
||||
assert dumped2["client_message_id"] == "opt-abc-123"
|
||||
rehydrated = Message.model_validate(dumped2)
|
||||
assert rehydrated.client_message_id == "opt-abc-123"
|
||||
|
||||
|
||||
def test_message_legacy_payload_without_client_id():
|
||||
"""Older session JSON files won't have the field — must still load."""
|
||||
from backend.apps.agents.models import Message
|
||||
|
||||
legacy = {
|
||||
"id": "abc",
|
||||
"role": "assistant",
|
||||
"content": "hello",
|
||||
"timestamp": "2026-04-29T00:00:00",
|
||||
"branch_id": "main",
|
||||
}
|
||||
m = Message.model_validate(legacy)
|
||||
assert m.client_message_id is None
|
||||
|
||||
|
||||
def test_client_message_id_collision_resistance():
|
||||
"""Many random client_message_ids must remain distinct values
|
||||
after serialization. Smoke-tests the field preservation in bulk."""
|
||||
from backend.apps.agents.models import Message
|
||||
|
||||
seen: set[str] = set()
|
||||
for _ in range(500):
|
||||
cmi = "opt-" + "".join(random.choices(string.ascii_lowercase + string.digits, k=24))
|
||||
seen.add(cmi)
|
||||
m = Message(role="user", content=f"msg {cmi}", client_message_id=cmi)
|
||||
assert m.client_message_id == cmi
|
||||
# Round-trip preserves it
|
||||
assert Message.model_validate(m.model_dump(mode="json")).client_message_id == cmi
|
||||
assert len(seen) >= 495 # collisions are statistically negligible
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 2 — Mode migration: chat → ask
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_builtin_modes_no_chat():
|
||||
"""Chat must be removed from BUILTIN_MODES; Ask must be present
|
||||
with the merged tools (Read+Glob+Grep + Web*)."""
|
||||
from backend.apps.modes.models import BUILTIN_MODES
|
||||
|
||||
ids = {m.id for m in BUILTIN_MODES}
|
||||
assert "chat" not in ids, "chat mode should have been merged into ask"
|
||||
assert "ask" in ids
|
||||
ask = next(m for m in BUILTIN_MODES if m.id == "ask")
|
||||
assert "WebFetch" in (ask.tools or []), "ask should now include web tools"
|
||||
assert "WebSearch" in (ask.tools or [])
|
||||
assert "Read" in (ask.tools or [])
|
||||
assert "Edit" not in (ask.tools or []), "ask must remain read-only"
|
||||
assert "Write" not in (ask.tools or [])
|
||||
assert "Bash" not in (ask.tools or [])
|
||||
|
||||
|
||||
def test_modes_lifespan_deletes_stale_chat():
|
||||
"""A built-in chat.json on disk must be removed on lifespan run.
|
||||
User-modified chat.json (is_builtin=False) must be left alone."""
|
||||
from backend.apps.modes import modes as modes_mod
|
||||
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
chat_path = os.path.join(td, "chat.json")
|
||||
with open(chat_path, "w") as f:
|
||||
json.dump({
|
||||
"id": "chat", "name": "Chat", "is_builtin": True,
|
||||
"system_prompt": "old", "tools": ["AskUserQuestion"],
|
||||
}, f)
|
||||
with patch.object(modes_mod, "DATA_DIR", td):
|
||||
asyncio.run(_run_lifespan(modes_mod))
|
||||
assert not os.path.exists(chat_path), "stale built-in chat.json should be removed"
|
||||
|
||||
# User-customized: leave alone
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
chat_path = os.path.join(td, "chat.json")
|
||||
with open(chat_path, "w") as f:
|
||||
json.dump({
|
||||
"id": "chat", "name": "MyChat", "is_builtin": False,
|
||||
"system_prompt": "user wrote this",
|
||||
}, f)
|
||||
with patch.object(modes_mod, "DATA_DIR", td):
|
||||
asyncio.run(_run_lifespan(modes_mod))
|
||||
assert os.path.exists(chat_path), "user-customized chat.json must NOT be deleted"
|
||||
|
||||
|
||||
async def _run_lifespan(modes_mod):
|
||||
async with modes_mod.modes_lifespan():
|
||||
pass
|
||||
|
||||
|
||||
def test_session_reconcile_migrates_chat_to_ask():
|
||||
"""reconcile_on_startup must rewrite mode='chat' to 'ask' on disk."""
|
||||
from backend.apps.agents.agent_manager import AgentManager
|
||||
from backend.apps.agents import agent_manager as am_mod
|
||||
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
# Seed 50 sessions: 30 with mode='chat', 20 with mode='agent'.
|
||||
# Some marked running so we also exercise the stale-status path.
|
||||
for i in range(50):
|
||||
sid = f"sess-{i}"
|
||||
mode = "chat" if i < 30 else "agent"
|
||||
status = "running" if i % 7 == 0 else "stopped"
|
||||
with open(os.path.join(td, f"{sid}.json"), "w") as f:
|
||||
json.dump({
|
||||
"id": sid, "name": sid, "model": "sonnet",
|
||||
"mode": mode, "status": status, "messages": [],
|
||||
}, f)
|
||||
|
||||
with patch.object(am_mod, "SESSIONS_DIR", td):
|
||||
mgr = AgentManager()
|
||||
asyncio.run(mgr.reconcile_on_startup())
|
||||
|
||||
for i in range(50):
|
||||
sid = f"sess-{i}"
|
||||
with open(os.path.join(td, f"{sid}.json")) as f:
|
||||
data = json.load(f)
|
||||
if i < 30:
|
||||
assert data["mode"] == "ask", f"session {sid} should be migrated chat→ask"
|
||||
else:
|
||||
assert data["mode"] == "agent", f"session {sid} should be untouched"
|
||||
# Stale running flipped to stopped
|
||||
if i % 7 == 0:
|
||||
assert data["status"] == "stopped"
|
||||
|
||||
|
||||
def test_reconcile_idempotent():
|
||||
"""Running reconcile twice mustn't keep rewriting / churn the file."""
|
||||
from backend.apps.agents.agent_manager import AgentManager
|
||||
from backend.apps.agents import agent_manager as am_mod
|
||||
|
||||
with tempfile.TemporaryDirectory() as td:
|
||||
sid = "s1"
|
||||
with open(os.path.join(td, f"{sid}.json"), "w") as f:
|
||||
json.dump({
|
||||
"id": sid, "name": sid, "model": "sonnet",
|
||||
"mode": "chat", "status": "stopped", "messages": [],
|
||||
}, f)
|
||||
with patch.object(am_mod, "SESSIONS_DIR", td):
|
||||
mgr = AgentManager()
|
||||
asyncio.run(mgr.reconcile_on_startup())
|
||||
mtime_after_first = os.path.getmtime(os.path.join(td, f"{sid}.json"))
|
||||
# Second pass must NOT rewrite (mode already 'ask', status already stopped)
|
||||
asyncio.run(mgr.reconcile_on_startup())
|
||||
mtime_after_second = os.path.getmtime(os.path.join(td, f"{sid}.json"))
|
||||
assert mtime_after_first == mtime_after_second, "reconcile must be idempotent"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 3 — ContentBlock / StreamEvent thinking acceptance
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_content_block_thinking_type():
|
||||
from backend.apps.agents.providers.base import ContentBlock
|
||||
|
||||
cb = ContentBlock(type="thinking", text="some reasoning")
|
||||
assert cb.type == "thinking"
|
||||
assert cb.text == "some reasoning"
|
||||
assert cb.tool_call is None
|
||||
|
||||
|
||||
def test_stream_event_thinking_delta():
|
||||
from backend.apps.agents.providers.base import StreamEvent
|
||||
|
||||
e = StreamEvent(type="content_block_delta", delta_type="thinking_delta", text="hmm")
|
||||
assert e.delta_type == "thinking_delta"
|
||||
assert e.text == "hmm"
|
||||
|
||||
# Existing types still work — no regression
|
||||
e2 = StreamEvent(type="content_block_delta", delta_type="text_delta", text="hi")
|
||||
assert e2.delta_type == "text_delta"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 4 — Anthropic provider thinking forwarding
|
||||
#
|
||||
# We feed a fake raw_stream (mimicking the SDK's async generator) through
|
||||
# AnthropicProvider.stream_message and confirm the right StreamEvents come
|
||||
# out. No network.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class _FakeRawEvent:
|
||||
def __init__(self, **kwargs):
|
||||
for k, v in kwargs.items():
|
||||
setattr(self, k, v)
|
||||
|
||||
|
||||
class _FakeBlock:
|
||||
def __init__(self, **kwargs):
|
||||
for k, v in kwargs.items():
|
||||
setattr(self, k, v)
|
||||
|
||||
|
||||
class _FakeDelta:
|
||||
def __init__(self, **kwargs):
|
||||
for k, v in kwargs.items():
|
||||
setattr(self, k, v)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_anthropic_provider_forwards_thinking_blocks():
|
||||
"""Mock the raw Anthropic stream with a thinking block + thinking_delta
|
||||
+ content_block_stop, and assert AnthropicProvider yields the
|
||||
normalized StreamEvents the agent_loop expects."""
|
||||
from backend.apps.agents.providers.anthropic import AnthropicProvider
|
||||
|
||||
raw_events = [
|
||||
# thinking block opens at index 0
|
||||
_FakeRawEvent(type="content_block_start", index=0,
|
||||
content_block=_FakeBlock(type="thinking")),
|
||||
_FakeRawEvent(type="content_block_delta", index=0,
|
||||
delta=_FakeDelta(type="thinking_delta", thinking="step 1, ")),
|
||||
_FakeRawEvent(type="content_block_delta", index=0,
|
||||
delta=_FakeDelta(type="thinking_delta", thinking="step 2.")),
|
||||
# signature_delta on thinking — must be ignored, not crash
|
||||
_FakeRawEvent(type="content_block_delta", index=0,
|
||||
delta=_FakeDelta(type="signature_delta", signature="abc==")),
|
||||
_FakeRawEvent(type="content_block_stop", index=0),
|
||||
# text block follows at index 1
|
||||
_FakeRawEvent(type="content_block_start", index=1,
|
||||
content_block=_FakeBlock(type="text")),
|
||||
_FakeRawEvent(type="content_block_delta", index=1,
|
||||
delta=_FakeDelta(type="text_delta", text="hi")),
|
||||
_FakeRawEvent(type="content_block_stop", index=1),
|
||||
]
|
||||
|
||||
async def fake_stream():
|
||||
for ev in raw_events:
|
||||
yield ev
|
||||
|
||||
# AnthropicProvider takes api_key/auth_token/base_url; we monkeypatch
|
||||
# its `client.messages.create` after construction so no real
|
||||
# SDK client is needed.
|
||||
provider = AnthropicProvider(api_key="test-key")
|
||||
provider.client.messages.create = AsyncMock(return_value=fake_stream())
|
||||
out_events = []
|
||||
async for ev in provider.stream_message(model="sonnet", system=None, messages=[], tools=[]):
|
||||
out_events.append(ev)
|
||||
|
||||
types = [(e.type, e.block_type, e.delta_type) for e in out_events]
|
||||
# Thinking block should produce: start, 2x delta, stop. signature_delta ignored.
|
||||
assert ("content_block_start", "thinking", "") in types
|
||||
assert types.count(("content_block_delta", "", "thinking_delta")) == 2
|
||||
assert ("content_block_start", "text", "") in types
|
||||
assert ("content_block_delta", "", "text_delta") in types
|
||||
|
||||
thinking_text = "".join(
|
||||
e.text for e in out_events
|
||||
if e.type == "content_block_delta" and e.delta_type == "thinking_delta"
|
||||
)
|
||||
assert thinking_text == "step 1, step 2."
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 5 — Agent loop end-to-end thinking → WS events + persisted message
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_agent_loop_emits_thinking_stream_and_persists_message():
|
||||
"""Drive the agent loop with a fake provider that yields thinking,
|
||||
text, and one tool_use. Verify it emits the right WS events AND
|
||||
persists a Message(role='thinking') via _emit_collected_messages."""
|
||||
from backend.apps.agents.providers.base import StreamEvent
|
||||
|
||||
captured_ws: list[tuple[str, dict]] = []
|
||||
|
||||
async def fake_emitter(event: str, payload: dict):
|
||||
captured_ws.append((event, payload))
|
||||
|
||||
# Build a fake provider yielding our normalized StreamEvents.
|
||||
class FakeProvider:
|
||||
async def stream_message(self, **kwargs):
|
||||
yield StreamEvent(type="content_block_start", index=0, block_type="thinking")
|
||||
yield StreamEvent(type="content_block_delta", index=0,
|
||||
delta_type="thinking_delta", text="reasoning… ")
|
||||
yield StreamEvent(type="content_block_delta", index=0,
|
||||
delta_type="thinking_delta", text="more.")
|
||||
yield StreamEvent(type="content_block_stop", index=0)
|
||||
yield StreamEvent(type="content_block_start", index=1, block_type="text")
|
||||
yield StreamEvent(type="content_block_delta", index=1,
|
||||
delta_type="text_delta", text="hello!")
|
||||
yield StreamEvent(type="content_block_stop", index=1)
|
||||
yield StreamEvent(type="message_stop")
|
||||
|
||||
from backend.apps.agents.agent_loop import AgentLoop
|
||||
|
||||
loop = AgentLoop(
|
||||
session_id="s1",
|
||||
provider=FakeProvider(),
|
||||
model="sonnet",
|
||||
system_prompt="x",
|
||||
tools=[],
|
||||
ws_emitter=fake_emitter,
|
||||
hitl_handler=AsyncMock(return_value=(True, None)),
|
||||
tool_executor=AsyncMock(return_value=[{"type": "text", "text": "ok"}]),
|
||||
)
|
||||
|
||||
response = await loop._stream_and_collect()
|
||||
|
||||
# Stream events: thinking start + 2 deltas + stream_end, then text start + delta + (text end at message_stop)
|
||||
events_by_type = {}
|
||||
for ev, payload in captured_ws:
|
||||
events_by_type.setdefault(ev, []).append(payload)
|
||||
|
||||
# Thinking should have its own stream_start with role='thinking'
|
||||
starts = events_by_type.get("agent:stream_start", [])
|
||||
thinking_starts = [s for s in starts if s.get("role") == "thinking"]
|
||||
assistant_starts = [s for s in starts if s.get("role") == "assistant"]
|
||||
assert len(thinking_starts) == 1, f"expected 1 thinking start, got {len(thinking_starts)}"
|
||||
assert len(assistant_starts) == 1, "expected 1 assistant text start"
|
||||
|
||||
# Two thinking deltas
|
||||
deltas = events_by_type.get("agent:stream_delta", [])
|
||||
thinking_msg_id = thinking_starts[0]["message_id"]
|
||||
thinking_deltas = [d for d in deltas if d.get("message_id") == thinking_msg_id]
|
||||
assert len(thinking_deltas) == 2
|
||||
assert "".join(d["delta"] for d in thinking_deltas) == "reasoning… more."
|
||||
|
||||
# Thinking stream_end fires (text doesn't get stream_end inside _stream_and_collect — closes at message_stop)
|
||||
ends = events_by_type.get("agent:stream_end", [])
|
||||
assert any(e["message_id"] == thinking_msg_id for e in ends), "thinking must emit stream_end"
|
||||
|
||||
# Now persist via _emit_collected_messages and verify a thinking
|
||||
# Message went out
|
||||
captured_ws.clear()
|
||||
await loop._emit_collected_messages(
|
||||
response.content,
|
||||
text_msg_id=assistant_starts[0]["message_id"],
|
||||
tool_msg_ids={},
|
||||
)
|
||||
persisted = [p for ev, p in captured_ws if ev == "agent:message"]
|
||||
roles = [p["message"]["role"] for p in persisted]
|
||||
assert "thinking" in roles, "thinking content must be persisted as a Message"
|
||||
assert "assistant" in roles
|
||||
thinking_msg = next(p for p in persisted if p["message"]["role"] == "thinking")
|
||||
assert thinking_msg["message"]["content"] == "reasoning… more."
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_agent_loop_handles_no_thinking_gracefully():
|
||||
"""Provider that emits zero thinking blocks must still work.
|
||||
Regression guard against the new branch breaking text-only paths."""
|
||||
from backend.apps.agents.providers.base import StreamEvent
|
||||
from backend.apps.agents.agent_loop import AgentLoop
|
||||
|
||||
captured_ws = []
|
||||
|
||||
async def fake_emitter(event, payload):
|
||||
captured_ws.append((event, payload))
|
||||
|
||||
class TextOnly:
|
||||
async def stream_message(self, **kwargs):
|
||||
yield StreamEvent(type="content_block_start", index=0, block_type="text")
|
||||
yield StreamEvent(type="content_block_delta", index=0,
|
||||
delta_type="text_delta", text="just text")
|
||||
yield StreamEvent(type="content_block_stop", index=0)
|
||||
yield StreamEvent(type="message_stop")
|
||||
|
||||
loop = AgentLoop(
|
||||
session_id="s2", provider=TextOnly(), model="sonnet", system_prompt=None,
|
||||
tools=[],
|
||||
ws_emitter=fake_emitter,
|
||||
hitl_handler=AsyncMock(return_value=(True, None)),
|
||||
tool_executor=AsyncMock(return_value=[]),
|
||||
)
|
||||
|
||||
resp = await loop._stream_and_collect()
|
||||
starts = [p for ev, p in captured_ws if ev == "agent:stream_start"]
|
||||
# Exactly one assistant start, zero thinking starts
|
||||
assert len([s for s in starts if s.get("role") == "thinking"]) == 0
|
||||
assert len([s for s in starts if s.get("role") == "assistant"]) == 1
|
||||
assert any(b.type == "text" for b in resp.content)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_agent_loop_stress_many_thinking_blocks():
|
||||
"""Hammer the loop with a long sequence of interleaved thinking +
|
||||
text + tool blocks. Ensures the per-index buffers don't leak and
|
||||
every block gets the right WS events."""
|
||||
from backend.apps.agents.providers.base import StreamEvent
|
||||
from backend.apps.agents.agent_loop import AgentLoop
|
||||
|
||||
captured = []
|
||||
|
||||
async def fake_emitter(ev, p):
|
||||
captured.append((ev, p))
|
||||
|
||||
class Mix:
|
||||
async def stream_message(self, **kwargs):
|
||||
idx = 0
|
||||
for turn in range(40):
|
||||
yield StreamEvent(type="content_block_start", index=idx, block_type="thinking")
|
||||
for _ in range(random.randint(1, 5)):
|
||||
yield StreamEvent(type="content_block_delta", index=idx,
|
||||
delta_type="thinking_delta", text=f"t{idx} ")
|
||||
yield StreamEvent(type="content_block_stop", index=idx)
|
||||
idx += 1
|
||||
yield StreamEvent(type="content_block_start", index=idx, block_type="text")
|
||||
yield StreamEvent(type="content_block_delta", index=idx,
|
||||
delta_type="text_delta", text=f"text-{idx}")
|
||||
yield StreamEvent(type="content_block_stop", index=idx)
|
||||
idx += 1
|
||||
yield StreamEvent(type="message_stop")
|
||||
|
||||
loop = AgentLoop(
|
||||
session_id="s3", provider=Mix(), model="sonnet", system_prompt=None,
|
||||
tools=[],
|
||||
ws_emitter=fake_emitter,
|
||||
hitl_handler=AsyncMock(return_value=(True, None)),
|
||||
tool_executor=AsyncMock(return_value=[]),
|
||||
)
|
||||
resp = await loop._stream_and_collect()
|
||||
|
||||
starts = [p for ev, p in captured if ev == "agent:stream_start"]
|
||||
ends = [p for ev, p in captured if ev == "agent:stream_end"]
|
||||
|
||||
# 40 thinking + 1 assistant (text accumulates into one stream_text_msg_id)
|
||||
thinking_starts = [s for s in starts if s.get("role") == "thinking"]
|
||||
assistant_starts = [s for s in starts if s.get("role") == "assistant"]
|
||||
assert len(thinking_starts) == 40, f"got {len(thinking_starts)} thinking starts, want 40"
|
||||
assert len(assistant_starts) == 1, "all text blocks share one assistant stream id"
|
||||
|
||||
# Each thinking block must have its own stream_end
|
||||
thinking_ids = {s["message_id"] for s in thinking_starts}
|
||||
end_ids = {e["message_id"] for e in ends}
|
||||
assert thinking_ids.issubset(end_ids), "every thinking block needs a stream_end"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 6 — Notes layout serialization
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_dashboard_layout_notes_round_trip():
|
||||
from backend.apps.dashboards.models import DashboardLayout, NotePosition
|
||||
|
||||
n = NotePosition(note_id="n1", x=100, y=200, content="todo: ship",
|
||||
color="yellow", width=240, height=200)
|
||||
layout = DashboardLayout(notes={"n1": n})
|
||||
dumped = layout.model_dump(mode="json")
|
||||
assert "notes" in dumped
|
||||
assert dumped["notes"]["n1"]["content"] == "todo: ship"
|
||||
|
||||
rehydrated = DashboardLayout.model_validate(dumped)
|
||||
assert rehydrated.notes["n1"].content == "todo: ship"
|
||||
assert rehydrated.notes["n1"].color == "yellow"
|
||||
|
||||
|
||||
def test_dashboard_layout_legacy_no_notes():
|
||||
"""Older dashboard JSON without 'notes' must still load cleanly."""
|
||||
from backend.apps.dashboards.models import DashboardLayout
|
||||
|
||||
legacy = {
|
||||
"cards": {}, "view_cards": {}, "browser_cards": {},
|
||||
"expanded_session_ids": [],
|
||||
}
|
||||
layout = DashboardLayout.model_validate(legacy)
|
||||
assert layout.notes == {}
|
||||
|
||||
|
||||
def test_notes_stress_many_round_trips():
|
||||
"""500 notes with random colors / positions must all serialize."""
|
||||
from backend.apps.dashboards.models import DashboardLayout, NotePosition
|
||||
|
||||
notes = {}
|
||||
colors = ["yellow", "pink", "blue", "green", "purple", "gray"]
|
||||
for i in range(500):
|
||||
nid = f"n{i}"
|
||||
notes[nid] = NotePosition(
|
||||
note_id=nid,
|
||||
x=random.uniform(-5000, 5000),
|
||||
y=random.uniform(-5000, 5000),
|
||||
width=random.uniform(160, 600),
|
||||
height=random.uniform(120, 600),
|
||||
content="x" * random.randint(0, 5000),
|
||||
color=random.choice(colors),
|
||||
)
|
||||
layout = DashboardLayout(notes=notes)
|
||||
dumped = layout.model_dump(mode="json")
|
||||
rehydrated = DashboardLayout.model_validate(dumped)
|
||||
assert len(rehydrated.notes) == 500
|
||||
for nid, orig in notes.items():
|
||||
assert rehydrated.notes[nid].content == orig.content
|
||||
assert rehydrated.notes[nid].color == orig.color
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Group 7 — Concurrent send_message dedupe stress
|
||||
#
|
||||
# Real-world scenario: user mashes Enter quickly. 50 concurrent sends
|
||||
# each with a unique client_message_id must produce 50 echoed messages
|
||||
# carrying the right ids. Pure pydantic / asyncio test — no real
|
||||
# agent loop.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_send_message_unique_client_ids():
|
||||
"""100 parallel Message constructions with unique client_message_ids
|
||||
must round-trip independently — no cross-talk on the dataclass."""
|
||||
from backend.apps.agents.models import Message
|
||||
|
||||
async def make_one(i: int):
|
||||
cmi = f"opt-{i}-{random.randint(0, 1_000_000)}"
|
||||
m = Message(role="user", content=f"msg {i}", client_message_id=cmi)
|
||||
return cmi, m.model_dump(mode="json")["client_message_id"]
|
||||
|
||||
pairs = await asyncio.gather(*(make_one(i) for i in range(100)))
|
||||
expected = [p[0] for p in pairs]
|
||||
actual = [p[1] for p in pairs]
|
||||
assert expected == actual, "client_message_id must round-trip exactly"
|
||||
assert len(set(actual)) == 100, "all unique"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Pytest config: register asyncio mode so we don't need the plugin.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def pytest_collection_modifyitems(config, items):
|
||||
"""Auto-mark async tests so they run under pytest-asyncio."""
|
||||
for item in items:
|
||||
if asyncio.iscoroutinefunction(getattr(item, "function", None)):
|
||||
item.add_marker(pytest.mark.asyncio)
|
||||
@@ -37,6 +37,7 @@ import DashboardHost from '@/app/components/Layout/DashboardHost';
|
||||
import { useLastDashboardId } from '@/shared/hooks/useLastDashboardId';
|
||||
import { useAppDispatch, useAppSelector } from '@/shared/hooks';
|
||||
import { fetchDashboards, createDashboard, renameDashboard } from '@/shared/state/dashboardsSlice';
|
||||
import { setPendingFocusAgentId } from '@/shared/state/tempStateSlice';
|
||||
import { addBrowserCard, addBrowserTab } from '@/shared/state/dashboardLayoutSlice';
|
||||
import { setPendingBrowserUrl } from '@/shared/state/tempStateSlice';
|
||||
import { fetchOutputs } from '@/shared/state/outputsSlice';
|
||||
@@ -233,6 +234,23 @@ const AppShell: React.FC = () => {
|
||||
try { localStorage.setItem(SIDEBAR_WIDTH_KEY, String(sidebarWidth)); } catch {}
|
||||
}, [sidebarWidth]);
|
||||
|
||||
// Native notification click handler. The notification helper fires a
|
||||
// window event with the session id + dashboard id; bring the user back
|
||||
// to that dashboard and queue a card focus.
|
||||
useEffect(() => {
|
||||
const handler = (e: Event) => {
|
||||
const detail = (e as CustomEvent).detail || {};
|
||||
const { sessionId, dashboardId } = detail as { sessionId?: string; dashboardId?: string };
|
||||
if (!sessionId) return;
|
||||
if (dashboardId) {
|
||||
navigate(`/dashboard/${dashboardId}`);
|
||||
}
|
||||
dispatch(setPendingFocusAgentId(sessionId));
|
||||
};
|
||||
window.addEventListener('openswarm:notification-click', handler as EventListener);
|
||||
return () => window.removeEventListener('openswarm:notification-click', handler as EventListener);
|
||||
}, [navigate, dispatch]);
|
||||
|
||||
const handleResizeStart = useCallback((e: React.MouseEvent) => {
|
||||
e.preventDefault();
|
||||
isResizing.current = true;
|
||||
|
||||
@@ -38,6 +38,7 @@ import {
|
||||
import { fetchModes } from '@/shared/state/modesSlice';
|
||||
import { createSessionWs } from '@/shared/ws/WebSocketManager';
|
||||
import MessageBubble from './MessageBubble';
|
||||
import CompactionMarker from './CompactionMarker';
|
||||
import MessageActionBar from './MessageActionBar';
|
||||
import ToolCallBubble, { ToolPair } from './ToolCallBubble';
|
||||
import ToolGroupBubble, { RenderItem, ToolGroup, isToolGroup, isToolPair } from './ToolGroupBubble';
|
||||
@@ -993,13 +994,33 @@ const AgentChat: React.FC<AgentChatProps> = ({ sessionId: sessionIdProp, onClose
|
||||
);
|
||||
})()}
|
||||
{renderItems.filter((item) => !session.streamingMessage || item.id !== session.streamingMessage.id).map((item) => {
|
||||
const isCompactionAnchor = !!session.compacted_through_msg_id && item.id === session.compacted_through_msg_id;
|
||||
const compactionChip = isCompactionAnchor ? (
|
||||
<CompactionMarker
|
||||
key={`compaction-${item.id}`}
|
||||
collapsedCount={
|
||||
Math.max(0, renderItems.findIndex((it) => it.id === session.compacted_through_msg_id) + 1)
|
||||
}
|
||||
/>
|
||||
) : null;
|
||||
|
||||
if (isToolGroup(item)) {
|
||||
const groupMeta = session.tool_group_meta?.[item.id];
|
||||
return <ToolGroupBubble key={item.id} group={item} isSessionRunning={sessionRunning} meta={groupMeta} sessionId={session.id} />;
|
||||
return (
|
||||
<React.Fragment key={item.id}>
|
||||
<ToolGroupBubble group={item} isSessionRunning={sessionRunning} meta={groupMeta} sessionId={session.id} />
|
||||
{compactionChip}
|
||||
</React.Fragment>
|
||||
);
|
||||
}
|
||||
if (isToolPair(item)) {
|
||||
const isPending = item.result === null && sessionRunning;
|
||||
return <ToolCallBubble key={item.id} call={item.call} result={item.result} isPending={isPending} sessionId={session.id} />;
|
||||
return (
|
||||
<React.Fragment key={item.id}>
|
||||
<ToolCallBubble call={item.call} result={item.result} isPending={isPending} sessionId={session.id} />
|
||||
{compactionChip}
|
||||
</React.Fragment>
|
||||
);
|
||||
}
|
||||
const msg = item;
|
||||
const isEditing = editingMessageId === msg.id;
|
||||
@@ -1043,6 +1064,7 @@ const AgentChat: React.FC<AgentChatProps> = ({ sessionId: sessionIdProp, onClose
|
||||
}
|
||||
/>
|
||||
)}
|
||||
{compactionChip}
|
||||
</Box>
|
||||
);
|
||||
})}
|
||||
|
||||
@@ -0,0 +1,48 @@
|
||||
import React from 'react';
|
||||
import Box from '@mui/material/Box';
|
||||
import Typography from '@mui/material/Typography';
|
||||
import UnfoldLessOutlinedIcon from '@mui/icons-material/UnfoldLessOutlined';
|
||||
import { useClaudeTokens } from '@/shared/styles/ThemeContext';
|
||||
|
||||
// Inline marker rendered immediately after the message identified by
|
||||
// `compacted_through_msg_id`. The auto-compaction routine summarizes older
|
||||
// turns into a single block; without a visible cue, the transcript would
|
||||
// just appear to "skip" — users assume the agent forgot something. The chip
|
||||
// makes it clear the older turns are still in scope, just collapsed.
|
||||
//
|
||||
// Click is currently a no-op (we don't surface the summary text yet); the
|
||||
// affordance reads as "hover for info" via the cursor style only. If we
|
||||
// later persist the summary text in the session, expand-to-reveal lands
|
||||
// here.
|
||||
const CompactionMarker: React.FC<{ collapsedCount: number }> = ({ collapsedCount }) => {
|
||||
const c = useClaudeTokens();
|
||||
const label = collapsedCount > 0
|
||||
? `${collapsedCount} earlier turn${collapsedCount === 1 ? '' : 's'} summarized`
|
||||
: 'Older turns summarized';
|
||||
return (
|
||||
<Box sx={{ display: 'flex', justifyContent: 'center', my: 1.25 }}>
|
||||
<Box
|
||||
sx={{
|
||||
display: 'inline-flex',
|
||||
alignItems: 'center',
|
||||
gap: 0.625,
|
||||
px: 1.25,
|
||||
py: 0.4,
|
||||
borderRadius: 9999,
|
||||
bgcolor: c.bg.secondary,
|
||||
border: `1px solid ${c.border.subtle}`,
|
||||
color: c.text.muted,
|
||||
cursor: 'default',
|
||||
userSelect: 'none',
|
||||
}}
|
||||
>
|
||||
<UnfoldLessOutlinedIcon sx={{ fontSize: 13, opacity: 0.7 }} />
|
||||
<Typography sx={{ fontSize: '0.7rem', lineHeight: 1, fontWeight: 500 }}>
|
||||
{label}
|
||||
</Typography>
|
||||
</Box>
|
||||
</Box>
|
||||
);
|
||||
};
|
||||
|
||||
export default CompactionMarker;
|
||||
@@ -542,14 +542,17 @@ const ThinkingBubble: React.FC<{
|
||||
const toggle = () => setUserOverride(!expanded);
|
||||
|
||||
const displayedSeconds = frozenElapsed ?? elapsed;
|
||||
// Live token approximation: ~4 chars per token works well enough for a
|
||||
// ticker. Skipped on history replay since `content` is fully populated
|
||||
// and we don't want a misleading "still thinking" feel.
|
||||
const text = typeof content === 'string' ? content : JSON.stringify(content);
|
||||
const liveTokenEstimate = isStreaming ? Math.max(0, Math.round(text.length / 4)) : 0;
|
||||
const label = isStreaming
|
||||
? 'Thinking...'
|
||||
? (liveTokenEstimate > 0 ? `Thinking… (~${liveTokenEstimate} tokens)` : 'Thinking…')
|
||||
: startedStreamingAt !== null
|
||||
? `Thought for ${displayedSeconds}s`
|
||||
: 'Thoughts';
|
||||
|
||||
const text = typeof content === 'string' ? content : JSON.stringify(content);
|
||||
|
||||
// Shimmer colors — use a bright mid-tone against the muted base to make
|
||||
// the sweep visible without being loud. The base color matches the
|
||||
// static "Thought for Ns" state so the only visible change is the moving
|
||||
@@ -740,6 +743,12 @@ const MessageBubble: React.FC<Props> = React.memo(({ message, editing = false, o
|
||||
? content.slice(0, 200)
|
||||
: JSON.stringify(content).slice(0, 200);
|
||||
|
||||
// Optimistic-bubble visuals: dim the bubble until the server echoes it
|
||||
// back (status: 'pending'), and tint it red on send failure.
|
||||
const optimisticStatus = (message as any).optimistic_status as 'pending' | 'failed' | undefined;
|
||||
const isPending = optimisticStatus === 'pending';
|
||||
const isFailed = optimisticStatus === 'failed';
|
||||
|
||||
return (
|
||||
<Box
|
||||
data-select-type="message"
|
||||
@@ -756,12 +765,17 @@ const MessageBubble: React.FC<Props> = React.memo(({ message, editing = false, o
|
||||
maxWidth: '85%',
|
||||
minWidth: 0,
|
||||
bgcolor: isUser ? c.user.bubble : c.bg.surface,
|
||||
border: isUser ? 'none' : `1px solid ${c.border.subtle}`,
|
||||
border: isUser ? (isFailed ? `1px solid ${c.status.error}` : 'none') : `1px solid ${c.border.subtle}`,
|
||||
borderRadius: isUser ? '16px 16px 4px 16px' : '16px 16px 16px 4px',
|
||||
px: 2,
|
||||
py: 1.25,
|
||||
boxShadow: isUser ? 'none' : c.shadow.sm,
|
||||
overflow: 'hidden',
|
||||
// Pending bubbles fade in at ~70% opacity until the server echo
|
||||
// resolves them; failed bubbles get a soft red tint so the user
|
||||
// can see the message didn't go through.
|
||||
opacity: isPending ? 0.7 : 1,
|
||||
transition: 'opacity 0.2s, border-color 0.2s',
|
||||
}}
|
||||
>
|
||||
{isUser ? (
|
||||
|
||||
@@ -20,6 +20,7 @@ import CallSplitIcon from '@mui/icons-material/CallSplit';
|
||||
import ReactMarkdown from 'react-markdown';
|
||||
import remarkGfm from 'remark-gfm';
|
||||
import { AgentMessage, expandSession, collapseSession, fetchSession } from '@/shared/state/agentsSlice';
|
||||
import { getToolLabel } from './toolLabels';
|
||||
import { useAppDispatch, useAppSelector } from '@/shared/hooks';
|
||||
import { placeCard, removeCard, setGlowingAgentCard, clearGlowingAgentCard, DEFAULT_CARD_W, DEFAULT_CARD_H, EXPANDED_CARD_MIN_H, GRID_GAP } from '@/shared/state/dashboardLayoutSlice';
|
||||
import { useClaudeTokens, useThemeMode } from '@/shared/styles/ThemeContext';
|
||||
@@ -2007,7 +2008,14 @@ const ToolCallBubble: React.FC<ToolCallBubbleProps> = React.memo(
|
||||
flexShrink: 0,
|
||||
}}
|
||||
>
|
||||
{mcpInfo.isMcp ? mcpInfo.displayName : toolName}
|
||||
{(() => {
|
||||
if (mcpInfo.isMcp) return mcpInfo.displayName;
|
||||
// Verb-tense progression: "Reading" while pending, "Read" once
|
||||
// a tool_result has landed. Denied/streaming fall back to the
|
||||
// present participle since the action is in-flight.
|
||||
const { present, past } = getToolLabel(toolName);
|
||||
return result && !isDenied ? past : present;
|
||||
})()}
|
||||
</Typography>
|
||||
{mcpInfo.isMcp && (
|
||||
<Typography
|
||||
|
||||
@@ -0,0 +1,75 @@
|
||||
// Friendly verb-tense labels for tool calls. Replaces the raw tool name in
|
||||
// ToolCallBubble titles so the transcript reads as a narration of what the
|
||||
// agent is doing — "Reading foo.ts" while pending, "Read foo.ts" once done.
|
||||
//
|
||||
// Falls back to the raw tool name (capitalized) for anything unmapped, so
|
||||
// new tools won't render badly. MCP tools (mcp__server__action) are handled
|
||||
// in ToolCallBubble's existing parseMcpToolName path; this map is for
|
||||
// built-ins.
|
||||
//
|
||||
// Tense convention:
|
||||
// - present: "-ing" form rendered while the call is pending
|
||||
// - past: rendered once a tool_result lands (success or error)
|
||||
//
|
||||
// Usage:
|
||||
// const { present, past } = getToolLabel(toolName);
|
||||
// const verb = isPending ? present : past;
|
||||
|
||||
interface ToolLabel {
|
||||
present: string;
|
||||
past: string;
|
||||
}
|
||||
|
||||
const LABELS: Record<string, ToolLabel> = {
|
||||
read: { present: 'Reading', past: 'Read' },
|
||||
write: { present: 'Writing', past: 'Wrote' },
|
||||
edit: { present: 'Editing', past: 'Edited' },
|
||||
multiedit: { present: 'Editing', past: 'Edited' },
|
||||
strreplace: { present: 'Editing', past: 'Edited' },
|
||||
bash: { present: 'Running', past: 'Ran' },
|
||||
glob: { present: 'Searching', past: 'Searched' },
|
||||
grep: { present: 'Searching', past: 'Searched' },
|
||||
ripgrep: { present: 'Searching', past: 'Searched' },
|
||||
ls: { present: 'Listing', past: 'Listed' },
|
||||
websearch: { present: 'Searching the web', past: 'Searched the web' },
|
||||
webfetch: { present: 'Fetching', past: 'Fetched' },
|
||||
notebookedit: { present: 'Editing notebook', past: 'Edited notebook' },
|
||||
todowrite: { present: 'Updating todos', past: 'Updated todos' },
|
||||
todoread: { present: 'Reading todos', past: 'Read todos' },
|
||||
taskcreate: { present: 'Creating task', past: 'Created task' },
|
||||
taskupdate: { present: 'Updating task', past: 'Updated task' },
|
||||
taskoutput: { present: 'Inspecting task', past: 'Inspected task' },
|
||||
taskstop: { present: 'Stopping task', past: 'Stopped task' },
|
||||
tasklist: { present: 'Listing tasks', past: 'Listed tasks' },
|
||||
taskget: { present: 'Loading task', past: 'Loaded task' },
|
||||
toolsearch: { present: 'Loading tools', past: 'Loaded tools' },
|
||||
mcpsearch: { present: 'Searching MCPs', past: 'Searched MCPs' },
|
||||
mcpactivate: { present: 'Activating MCP', past: 'Activated MCP' },
|
||||
outputactivate: { present: 'Activating view', past: 'Activated view' },
|
||||
renderoutput: { present: 'Rendering view', past: 'Rendered view' },
|
||||
askuserquestion: { present: 'Asking', past: 'Asked' },
|
||||
invokeagent: { present: 'Invoking sub-agent', past: 'Invoked sub-agent' },
|
||||
agent: { present: 'Spawning agent', past: 'Spawned agent' },
|
||||
enterplanmode: { present: 'Entering plan mode', past: 'Entered plan mode' },
|
||||
exitplanmode: { present: 'Exiting plan mode', past: 'Exited plan mode' },
|
||||
enterworktree: { present: 'Creating worktree', past: 'Created worktree' },
|
||||
exitworktree: { present: 'Removing worktree', past: 'Removed worktree' },
|
||||
pushnotification: { present: 'Notifying', past: 'Notified' },
|
||||
remotetrigger: { present: 'Triggering', past: 'Triggered' },
|
||||
croncreate: { present: 'Scheduling', past: 'Scheduled' },
|
||||
cronlist: { present: 'Listing schedules', past: 'Listed schedules' },
|
||||
crondelete: { present: 'Cancelling schedule', past: 'Cancelled schedule' },
|
||||
monitor: { present: 'Watching', past: 'Watched' },
|
||||
schedulewakeup: { present: 'Scheduling wake-up', past: 'Scheduled wake-up' },
|
||||
};
|
||||
|
||||
export function getToolLabel(toolName: string): ToolLabel {
|
||||
if (!toolName) return { present: 'Working', past: 'Done' };
|
||||
const key = toolName.toLowerCase();
|
||||
const hit = LABELS[key];
|
||||
if (hit) return hit;
|
||||
// Fallback: capitalize the raw name with neutral verbs that read OK either
|
||||
// way ("Running tool" / "Ran tool").
|
||||
const pretty = toolName.charAt(0).toUpperCase() + toolName.slice(1);
|
||||
return { present: `Running ${pretty}`, past: `Ran ${pretty}` };
|
||||
}
|
||||
@@ -0,0 +1,74 @@
|
||||
// Native (Electron / browser) notifications for agent completion.
|
||||
//
|
||||
// We only fire when the document is hidden — the user has switched away —
|
||||
// since a notification while you're staring at the same window would just
|
||||
// be noise. Granola/Linear/Raycast all converge on this rule.
|
||||
//
|
||||
// Permission is requested lazily on first attempted use; subsequent calls
|
||||
// no-op gracefully when permission is denied. Click on a notification
|
||||
// re-focuses the window and emits a custom event the renderer listens for
|
||||
// to deep-link back to the right session.
|
||||
|
||||
const FIRED_RECENTLY = new Set<string>();
|
||||
const COOLDOWN_MS = 30_000;
|
||||
|
||||
let permissionRequested = false;
|
||||
|
||||
function ensurePermission(): NotificationPermission {
|
||||
if (typeof Notification === 'undefined') return 'denied';
|
||||
if (Notification.permission === 'granted') return 'granted';
|
||||
if (Notification.permission === 'denied') return 'denied';
|
||||
if (!permissionRequested) {
|
||||
permissionRequested = true;
|
||||
Notification.requestPermission().catch(() => {});
|
||||
}
|
||||
return 'default';
|
||||
}
|
||||
|
||||
export interface AgentCompletionPayload {
|
||||
sessionId: string;
|
||||
sessionName: string;
|
||||
dashboardId?: string;
|
||||
status: 'completed' | 'error';
|
||||
bodyExcerpt?: string;
|
||||
}
|
||||
|
||||
export function notifyAgentCompletion(p: AgentCompletionPayload): void {
|
||||
if (typeof document === 'undefined') return;
|
||||
// Same-window — skip noise. Hidden = tab switched, window minimised, or
|
||||
// (in Electron) another BrowserWindow is in front.
|
||||
if (!document.hidden) return;
|
||||
if (typeof Notification === 'undefined') return;
|
||||
const perm = ensurePermission();
|
||||
if (perm !== 'granted') return;
|
||||
|
||||
// Per-session debounce — if a sub-agent flips completed→error→completed
|
||||
// in quick succession we still only fire one toast.
|
||||
const key = `${p.sessionId}:${p.status}`;
|
||||
if (FIRED_RECENTLY.has(key)) return;
|
||||
FIRED_RECENTLY.add(key);
|
||||
setTimeout(() => FIRED_RECENTLY.delete(key), COOLDOWN_MS);
|
||||
|
||||
const title = p.status === 'error'
|
||||
? `${p.sessionName} hit an error`
|
||||
: `${p.sessionName} finished`;
|
||||
const body = (p.bodyExcerpt || '').slice(0, 140);
|
||||
|
||||
try {
|
||||
const n = new Notification(title, {
|
||||
body,
|
||||
tag: p.sessionId,
|
||||
silent: false,
|
||||
});
|
||||
n.onclick = () => {
|
||||
try { window.focus(); } catch {}
|
||||
window.dispatchEvent(new CustomEvent('openswarm:notification-click', {
|
||||
detail: { sessionId: p.sessionId, dashboardId: p.dashboardId },
|
||||
}));
|
||||
n.close();
|
||||
};
|
||||
} catch {
|
||||
// Notification API can throw if the page is sandboxed or in a
|
||||
// headless harness — fail silently.
|
||||
}
|
||||
}
|
||||
@@ -15,6 +15,15 @@ export interface AgentMessage {
|
||||
forced_tools?: string[];
|
||||
images?: Array<{ data: string; media_type: string }>;
|
||||
hidden?: boolean;
|
||||
// Client-generated id used for optimistic-bubble dedupe. Set on the
|
||||
// optimistic message we synthesize in `sendMessage.pending` and on the
|
||||
// server echo (round-tripped via the POST body); the addMessage reducer
|
||||
// uses it to find and replace the optimistic placeholder.
|
||||
client_message_id?: string;
|
||||
// Frontend-only lifecycle marker for optimistic messages. 'pending' until
|
||||
// the server echo lands; 'failed' if the POST rejected. Confirmed messages
|
||||
// (i.e. ones echoed back from the server) drop this field entirely.
|
||||
optimistic_status?: 'pending' | 'failed';
|
||||
}
|
||||
|
||||
export interface ApprovalRequest {
|
||||
@@ -179,15 +188,41 @@ export interface SendMessagePayload {
|
||||
selectedBrowserIds?: string[];
|
||||
}
|
||||
|
||||
function _genOptimisticId(): string {
|
||||
return `opt-${Date.now().toString(36)}-${Math.random().toString(36).slice(2, 8)}`;
|
||||
}
|
||||
|
||||
export const sendMessage = createAsyncThunk(
|
||||
'agents/sendMessage',
|
||||
async ({ sessionId, prompt, mode, model, provider, images, contextPaths, forcedTools, attachedSkills, hidden, selectedBrowserIds }: SendMessagePayload) => {
|
||||
await fetch(`${AGENTS_API}/sessions/${sessionId}/message`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ prompt, mode, model, provider, images, context_paths: contextPaths, forced_tools: forcedTools, attached_skills: attachedSkills, hidden, selected_browser_ids: selectedBrowserIds }),
|
||||
});
|
||||
return { sessionId, prompt };
|
||||
async ({ sessionId, prompt, mode, model, provider, images, contextPaths, forcedTools, attachedSkills, hidden, selectedBrowserIds }: SendMessagePayload, { dispatch }) => {
|
||||
// Generate an optimistic id up-front and dispatch the synchronous
|
||||
// bubble *before* awaiting the network. The reducer below
|
||||
// (sendMessage.pending) handles the same path, but doing it here
|
||||
// gives us access to the id we'll round-trip to the server for
|
||||
// dedupe on echo.
|
||||
const clientMessageId = _genOptimisticId();
|
||||
dispatch(addOptimisticMessage({
|
||||
sessionId,
|
||||
clientMessageId,
|
||||
prompt,
|
||||
contextPaths,
|
||||
forcedTools,
|
||||
attachedSkills: attachedSkills?.map((s) => ({ id: s.id, name: s.name })),
|
||||
images: images?.map((img) => ({ data: img.data, media_type: img.media_type })),
|
||||
hidden: hidden ?? false,
|
||||
}));
|
||||
try {
|
||||
const res = await fetch(`${AGENTS_API}/sessions/${sessionId}/message`, {
|
||||
method: 'POST',
|
||||
headers: { 'Content-Type': 'application/json' },
|
||||
body: JSON.stringify({ prompt, mode, model, provider, images, context_paths: contextPaths, forced_tools: forcedTools, attached_skills: attachedSkills, hidden, selected_browser_ids: selectedBrowserIds, client_message_id: clientMessageId }),
|
||||
});
|
||||
if (!res.ok) throw new Error(`send failed: ${res.status}`);
|
||||
} catch (err) {
|
||||
dispatch(markOptimisticFailed({ sessionId, clientMessageId }));
|
||||
throw err;
|
||||
}
|
||||
return { sessionId, prompt, clientMessageId };
|
||||
}
|
||||
);
|
||||
|
||||
@@ -630,17 +665,97 @@ const agentsSlice = createSlice({
|
||||
|
||||
addMessage(state, action: PayloadAction<{ sessionId: string; message: AgentMessage }>) {
|
||||
const session = state.sessions[action.payload.sessionId];
|
||||
if (session) {
|
||||
const idx = session.messages.findIndex((m) => m.id === action.payload.message.id);
|
||||
if (idx >= 0) {
|
||||
session.messages[idx] = action.payload.message;
|
||||
} else {
|
||||
session.messages.push(action.payload.message);
|
||||
}
|
||||
if (session.streamingMessage?.id === action.payload.message.id) {
|
||||
session.streamingMessage = null;
|
||||
if (!session) return;
|
||||
const incoming = action.payload.message;
|
||||
// Optimistic-bubble dedupe: if this echo carries a client_message_id
|
||||
// and we have an optimistic placeholder with the same id, replace it
|
||||
// with the server version (preserving server's id, dropping the
|
||||
// optimistic_status marker so the bubble renders as confirmed).
|
||||
if (incoming.client_message_id) {
|
||||
const optIdx = session.messages.findIndex(
|
||||
(m) => m.client_message_id === incoming.client_message_id && m.optimistic_status === 'pending',
|
||||
);
|
||||
if (optIdx >= 0) {
|
||||
session.messages[optIdx] = { ...incoming, optimistic_status: undefined };
|
||||
if (session.streamingMessage?.id === incoming.id) {
|
||||
session.streamingMessage = null;
|
||||
}
|
||||
return;
|
||||
}
|
||||
}
|
||||
const idx = session.messages.findIndex((m) => m.id === incoming.id);
|
||||
if (idx >= 0) {
|
||||
session.messages[idx] = incoming;
|
||||
} else {
|
||||
session.messages.push(incoming);
|
||||
}
|
||||
if (session.streamingMessage?.id === incoming.id) {
|
||||
session.streamingMessage = null;
|
||||
}
|
||||
},
|
||||
|
||||
// Synchronous "you sent a message" bubble dispatched from the
|
||||
// sendMessage thunk before the network round-trip. The placeholder
|
||||
// carries a client_message_id which the server echo (agent:message)
|
||||
// will round-trip back; addMessage dedupes against it.
|
||||
addOptimisticMessage(
|
||||
state,
|
||||
action: PayloadAction<{
|
||||
sessionId: string;
|
||||
clientMessageId: string;
|
||||
prompt: string;
|
||||
contextPaths?: Array<{ path: string; type: 'file' | 'directory' }>;
|
||||
forcedTools?: string[];
|
||||
attachedSkills?: Array<{ id: string; name: string }>;
|
||||
images?: Array<{ data: string; media_type: string }>;
|
||||
hidden?: boolean;
|
||||
}>,
|
||||
) {
|
||||
const { sessionId, clientMessageId, prompt, contextPaths, forcedTools, attachedSkills, images, hidden } = action.payload;
|
||||
const session = state.sessions[sessionId];
|
||||
if (!session) return;
|
||||
// Hidden messages (e.g. continuation prompts the model fires
|
||||
// internally) shouldn't render an optimistic bubble.
|
||||
if (hidden) return;
|
||||
session.messages.push({
|
||||
id: clientMessageId,
|
||||
role: 'user',
|
||||
content: prompt,
|
||||
timestamp: new Date().toISOString(),
|
||||
branch_id: session.active_branch_id,
|
||||
parent_id: null,
|
||||
context_paths: contextPaths,
|
||||
attached_skills: attachedSkills,
|
||||
forced_tools: forcedTools,
|
||||
images,
|
||||
client_message_id: clientMessageId,
|
||||
optimistic_status: 'pending',
|
||||
});
|
||||
},
|
||||
|
||||
markOptimisticFailed(
|
||||
state,
|
||||
action: PayloadAction<{ sessionId: string; clientMessageId: string }>,
|
||||
) {
|
||||
const session = state.sessions[action.payload.sessionId];
|
||||
if (!session) return;
|
||||
const msg = session.messages.find(
|
||||
(m) => m.client_message_id === action.payload.clientMessageId && m.optimistic_status === 'pending',
|
||||
);
|
||||
if (msg) msg.optimistic_status = 'failed';
|
||||
},
|
||||
|
||||
// Backend emits agent:context_status with reason="compacted" when the
|
||||
// auto-compaction routine collapses older turns into a summary. We
|
||||
// mirror compacted_through_msg_id locally so the renderer can drop a
|
||||
// chip in the transcript right after that message.
|
||||
recordCompaction(
|
||||
state,
|
||||
action: PayloadAction<{ sessionId: string; throughMsgId: string | null }>,
|
||||
) {
|
||||
const session = state.sessions[action.payload.sessionId];
|
||||
if (!session) return;
|
||||
session.compacted_through_msg_id = action.payload.throughMsgId;
|
||||
},
|
||||
|
||||
streamStart(
|
||||
@@ -1166,6 +1281,9 @@ export const {
|
||||
updateSessionStatus,
|
||||
setSessionConnState,
|
||||
addMessage,
|
||||
addOptimisticMessage,
|
||||
markOptimisticFailed,
|
||||
recordCompaction,
|
||||
streamStart,
|
||||
streamDelta,
|
||||
streamEnd,
|
||||
|
||||
@@ -20,9 +20,11 @@ import {
|
||||
trackAgentNotification,
|
||||
setSessionConnState,
|
||||
fetchSession,
|
||||
recordCompaction,
|
||||
} from '../state/agentsSlice';
|
||||
import { addBrowserCardFromBackend, removeBrowserCard, setBrowserCardPosition, setGlowingBrowserCards, GRID_GAP } from '../state/dashboardLayoutSlice';
|
||||
import { getAuthToken } from '../config';
|
||||
import { notifyAgentCompletion } from '../notifications';
|
||||
|
||||
// Thin wrapper around getAuthToken so the connect() call site stays
|
||||
// synchronous. If the token isn't cached yet, returns '' and the WS
|
||||
@@ -405,13 +407,50 @@ class WebSocketManager {
|
||||
|
||||
switch (event) {
|
||||
case 'agent:status':
|
||||
if (data.session) {
|
||||
store.dispatch(updateSession(data.session));
|
||||
} else if (session_id) {
|
||||
store.dispatch(updateSessionStatus({ sessionId: session_id, status: data.status }));
|
||||
}
|
||||
if (data.status === 'running' && session_id) {
|
||||
store.dispatch(trackAgentNotification(session_id));
|
||||
// Capture pre-transition status so we only fire a system notification
|
||||
// on a real running→terminal transition. Otherwise a session that
|
||||
// was already 'completed' on disk and got refetched would re-toast.
|
||||
{
|
||||
const prevSession = session_id ? store.getState().agents.sessions[session_id] : undefined;
|
||||
const prevStatus = prevSession?.status;
|
||||
|
||||
if (data.session) {
|
||||
store.dispatch(updateSession(data.session));
|
||||
} else if (session_id) {
|
||||
store.dispatch(updateSessionStatus({ sessionId: session_id, status: data.status }));
|
||||
}
|
||||
if (data.status === 'running' && session_id) {
|
||||
store.dispatch(trackAgentNotification(session_id));
|
||||
}
|
||||
|
||||
// Fire a native notification when an agent terminates while the
|
||||
// window is hidden. Skips sub-agents and browser-agents (the
|
||||
// parent's own completion is what the user cares about) and only
|
||||
// fires on a real transition from a non-terminal state.
|
||||
const TERMINAL = new Set(['completed', 'error']);
|
||||
const NON_TERMINAL = new Set(['running', 'waiting_approval', undefined, null, '']);
|
||||
if (
|
||||
session_id &&
|
||||
TERMINAL.has(data.status) &&
|
||||
NON_TERMINAL.has(prevStatus as any) &&
|
||||
data.session?.mode !== 'browser-agent' &&
|
||||
data.session?.mode !== 'sub-agent' &&
|
||||
data.session?.mode !== 'invoked-agent'
|
||||
) {
|
||||
const sess = data.session ?? prevSession;
|
||||
if (sess) {
|
||||
const lastAssistant = [...(sess.messages || [])]
|
||||
.reverse()
|
||||
.find((m: any) => m.role === 'assistant' && typeof m.content === 'string');
|
||||
notifyAgentCompletion({
|
||||
sessionId: session_id,
|
||||
sessionName: sess.name || 'Agent',
|
||||
dashboardId: sess.dashboard_id,
|
||||
status: data.status as 'completed' | 'error',
|
||||
bodyExcerpt: lastAssistant ? String(lastAssistant.content) : undefined,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
// Per-sub-agent close via browser_id; skip user-created cards (no spawned_by).
|
||||
if (
|
||||
@@ -510,6 +549,20 @@ class WebSocketManager {
|
||||
}
|
||||
break;
|
||||
|
||||
case 'agent:context_status':
|
||||
// Auto-compaction collapsed older turns into a summary. Mirror
|
||||
// compacted_through_msg_id locally so the renderer can drop a
|
||||
// visible "N earlier turns summarized" chip into the transcript.
|
||||
// Other reasons (cleared, etc.) flow through this same event but
|
||||
// don't currently need a chip — ignore them for now.
|
||||
if (session_id && data.reason === 'compacted') {
|
||||
store.dispatch(recordCompaction({
|
||||
sessionId: session_id,
|
||||
throughMsgId: data.compacted_through_msg_id ?? null,
|
||||
}));
|
||||
}
|
||||
break;
|
||||
|
||||
case 'agent:auth_error':
|
||||
// Re-uses the context_overflow card slot — both are "this session is
|
||||
// blocked, here's what to do" cards. Reason field disambiguates.
|
||||
|
||||
File diff suppressed because one or more lines are too long
Reference in New Issue
Block a user