fix: chat functionality and extended thinking handling
Deploy Research Workbench / deploy (push) Failing after 14m1s
Deploy Research Workbench / deploy (push) Failing after 14m1s
- bundle/bundle.md: Add provider-anthropic with source URL and orchestrator config override to disable extended_thinking (foundation enables it by default, causes empty text responses for simple queries) - Dockerfile: Add amplifier-module-provider-anthropic pip install for container - backend/agui_adapter.py: Fix thinking block handling and empty text block rendering - Thinking blocks no longer fall through to text handler - Text blocks only emit START when actual content arrives - Handle amplifierd wire format (block_index/block_type, complete text in content_block:end) - frontend/src/hooks/useChat.ts: Reset assistantId on TEXT_MESSAGE_END to prevent stale message IDs across turns - tests/test_agui_adapter.py: Add extended thinking test coverage Chat functionality now works end-to-end. Verified: login -> create session -> send message -> receive AI response visible in chat UI. Generated with Amplifier
This commit is contained in:
+2
-1
@@ -54,7 +54,8 @@ ENV UV_LINK_MODE=copy
|
|||||||
# amplifierd depends on), so we install it explicitly in the same step.
|
# amplifierd depends on), so we install it explicitly in the same step.
|
||||||
# amplifier-core IS on PyPI and will be resolved automatically.
|
# amplifier-core IS on PyPI and will be resolved automatically.
|
||||||
RUN uv pip install --system \
|
RUN uv pip install --system \
|
||||||
"amplifierd @ git+https://github.com/microsoft/amplifierd"
|
"amplifierd @ git+https://github.com/microsoft/amplifierd" \
|
||||||
|
"amplifier-module-provider-anthropic @ git+https://github.com/microsoft/amplifier-module-provider-anthropic"
|
||||||
|
|
||||||
# ── Frontend: install deps and build ──────────────────────────────────────────
|
# ── Frontend: install deps and build ──────────────────────────────────────────
|
||||||
# Copy package manifests first so npm ci is cached independently of source changes.
|
# Copy package manifests first so npm ci is cached independently of source changes.
|
||||||
|
|||||||
+80
-20
@@ -40,13 +40,19 @@ class AmplifierdEventTranslator:
|
|||||||
"""Translate a single amplifierd SSE event into zero or more AG-UI events.
|
"""Translate a single amplifierd SSE event into zero or more AG-UI events.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
event: Dict with 'event_type' and 'data' keys.
|
event: Dict with 'event_type' and 'data' keys. The 'data' value is
|
||||||
|
the raw SSE payload from amplifierd, which wraps the event-specific
|
||||||
|
fields under a nested ``"data"`` key
|
||||||
|
(e.g. ``{"event": "...", "data": {...}, "session_id": ..., ...}``).
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
List of ag_ui.core event objects (may be empty for unknown events).
|
List of ag_ui.core event objects (may be empty for unknown events).
|
||||||
"""
|
"""
|
||||||
event_type = event.get("event_type", "")
|
event_type = event.get("event_type", "")
|
||||||
data = event.get("data", {})
|
outer = event.get("data", {})
|
||||||
|
# amplifierd wraps event-specific fields in a nested "data" key.
|
||||||
|
# Fall back to the outer dict for tests that pass data directly.
|
||||||
|
data = outer.get("data", outer)
|
||||||
|
|
||||||
if event_type == "content_block:start":
|
if event_type == "content_block:start":
|
||||||
return self._handle_block_start(data)
|
return self._handle_block_start(data)
|
||||||
@@ -62,9 +68,13 @@ class AmplifierdEventTranslator:
|
|||||||
return []
|
return []
|
||||||
|
|
||||||
def _handle_block_start(self, data: dict[str, Any]) -> list[Any]:
|
def _handle_block_start(self, data: dict[str, Any]) -> list[Any]:
|
||||||
"""Handle content_block:start — emit TEXT_MESSAGE_START or TOOL_CALL_START."""
|
"""Handle content_block:start — register block; defer TEXT_MESSAGE_START to first delta.
|
||||||
index = data.get("index", 0)
|
|
||||||
block_type = data.get("type", "text")
|
Supports both amplifierd wire format (``block_index`` / ``block_type``) and the
|
||||||
|
legacy test format (``index`` / ``type``) via fallbacks.
|
||||||
|
"""
|
||||||
|
index = data.get("block_index", data.get("index", 0))
|
||||||
|
block_type = data.get("block_type", data.get("type", "text"))
|
||||||
|
|
||||||
if block_type == "tool_use":
|
if block_type == "tool_use":
|
||||||
tool_id = data.get("id", "")
|
tool_id = data.get("id", "")
|
||||||
@@ -81,21 +91,27 @@ class AmplifierdEventTranslator:
|
|||||||
parent_message_id=self._current_message_id,
|
parent_message_id=self._current_message_id,
|
||||||
)
|
)
|
||||||
]
|
]
|
||||||
else:
|
elif block_type == "text":
|
||||||
# Text block
|
# Lazy start: don't emit TEXT_MESSAGE_START until we know there's actual content.
|
||||||
message_id = str(uuid.uuid4())
|
message_id = str(uuid.uuid4())
|
||||||
self._blocks[index] = {"type": "text", "message_id": message_id}
|
self._blocks[index] = {"type": "text", "message_id": message_id, "started": False}
|
||||||
self._current_message_id = message_id
|
self._current_message_id = message_id
|
||||||
return [
|
return []
|
||||||
TextMessageStartEvent(
|
else:
|
||||||
message_id=message_id,
|
# thinking, image, or any other block type — skip entirely
|
||||||
role="assistant",
|
self._blocks[index] = {"type": "skip"}
|
||||||
)
|
return []
|
||||||
]
|
|
||||||
|
|
||||||
def _handle_block_delta(self, data: dict[str, Any]) -> list[Any]:
|
def _handle_block_delta(self, data: dict[str, Any]) -> list[Any]:
|
||||||
"""Handle content_block:delta — emit TEXT_MESSAGE_CONTENT or TOOL_CALL_ARGS."""
|
"""Handle content_block:delta — emit TEXT_MESSAGE_CONTENT or TOOL_CALL_ARGS.
|
||||||
index = data.get("index", 0)
|
|
||||||
|
For text blocks, TEXT_MESSAGE_START is emitted lazily here on the first
|
||||||
|
non-empty token so that thinking/empty text blocks produce zero AG-UI events.
|
||||||
|
|
||||||
|
Supports both amplifierd wire format (``block_index``) and legacy test format
|
||||||
|
(``index``) via fallback.
|
||||||
|
"""
|
||||||
|
index = data.get("block_index", data.get("index", 0))
|
||||||
delta = data.get("delta", {})
|
delta = data.get("delta", {})
|
||||||
delta_type = delta.get("type", "")
|
delta_type = delta.get("type", "")
|
||||||
|
|
||||||
@@ -112,6 +128,9 @@ class AmplifierdEventTranslator:
|
|||||||
]
|
]
|
||||||
return []
|
return []
|
||||||
|
|
||||||
|
if block_info["type"] == "skip":
|
||||||
|
return []
|
||||||
|
|
||||||
if block_info["type"] == "tool_use":
|
if block_info["type"] == "tool_use":
|
||||||
return [
|
return [
|
||||||
ToolCallArgsEvent(
|
ToolCallArgsEvent(
|
||||||
@@ -120,10 +139,22 @@ class AmplifierdEventTranslator:
|
|||||||
)
|
)
|
||||||
]
|
]
|
||||||
else:
|
else:
|
||||||
# Text block
|
# Text block — lazy start: emit TEXT_MESSAGE_START on the first real token
|
||||||
token = delta.get("text", "")
|
token = delta.get("text", "")
|
||||||
if not token:
|
if not token:
|
||||||
return []
|
return []
|
||||||
|
if not block_info.get("started"):
|
||||||
|
block_info["started"] = True
|
||||||
|
return [
|
||||||
|
TextMessageStartEvent(
|
||||||
|
message_id=block_info["message_id"],
|
||||||
|
role="assistant",
|
||||||
|
),
|
||||||
|
TextMessageContentEvent(
|
||||||
|
message_id=block_info["message_id"],
|
||||||
|
delta=token,
|
||||||
|
),
|
||||||
|
]
|
||||||
return [
|
return [
|
||||||
TextMessageContentEvent(
|
TextMessageContentEvent(
|
||||||
message_id=block_info["message_id"],
|
message_id=block_info["message_id"],
|
||||||
@@ -132,17 +163,46 @@ class AmplifierdEventTranslator:
|
|||||||
]
|
]
|
||||||
|
|
||||||
def _handle_block_end(self, data: dict[str, Any]) -> list[Any]:
|
def _handle_block_end(self, data: dict[str, Any]) -> list[Any]:
|
||||||
"""Handle content_block:end — emit TEXT_MESSAGE_END or TOOL_CALL_END."""
|
"""Handle content_block:end — emit TEXT_MESSAGE_END or TOOL_CALL_END.
|
||||||
index = data.get("index", 0)
|
|
||||||
|
Supports two modes:
|
||||||
|
- **Streaming mode** (delta events were received): the block was already opened
|
||||||
|
via ``_handle_block_delta``; just close it with TEXT_MESSAGE_END.
|
||||||
|
- **Batch mode** (no delta events, full text in ``block.text``): emit
|
||||||
|
TEXT_MESSAGE_START + TEXT_MESSAGE_CONTENT + TEXT_MESSAGE_END in one shot.
|
||||||
|
This is the format used by amplifierd, which sends complete blocks at end time.
|
||||||
|
|
||||||
|
Text blocks with no content in either mode are silently dropped.
|
||||||
|
|
||||||
|
Supports both amplifierd wire format (``block_index``) and legacy test format
|
||||||
|
(``index``) via fallback.
|
||||||
|
"""
|
||||||
|
index = data.get("block_index", data.get("index", 0))
|
||||||
block_info = self._blocks.pop(index, None)
|
block_info = self._blocks.pop(index, None)
|
||||||
|
|
||||||
if block_info is None:
|
if block_info is None:
|
||||||
return []
|
return []
|
||||||
|
|
||||||
|
if block_info["type"] == "skip":
|
||||||
|
return []
|
||||||
|
|
||||||
if block_info["type"] == "tool_use":
|
if block_info["type"] == "tool_use":
|
||||||
return [ToolCallEndEvent(tool_call_id=block_info["tool_id"])]
|
return [ToolCallEndEvent(tool_call_id=block_info["tool_id"])]
|
||||||
else:
|
else:
|
||||||
return [TextMessageEndEvent(message_id=block_info["message_id"])]
|
message_id = block_info["message_id"]
|
||||||
|
if block_info.get("started"):
|
||||||
|
# Streaming mode: deltas already opened the message — just close it.
|
||||||
|
return [TextMessageEndEvent(message_id=message_id)]
|
||||||
|
# Batch mode: full text is in the "block" field (amplifierd wire format).
|
||||||
|
block = data.get("block", {})
|
||||||
|
text = block.get("text", "")
|
||||||
|
if text:
|
||||||
|
return [
|
||||||
|
TextMessageStartEvent(message_id=message_id, role="assistant"),
|
||||||
|
TextMessageContentEvent(message_id=message_id, delta=text),
|
||||||
|
TextMessageEndEvent(message_id=message_id),
|
||||||
|
]
|
||||||
|
return []
|
||||||
|
|
||||||
def _handle_tool_result(self, data: dict[str, Any]) -> list[Any]:
|
def _handle_tool_result(self, data: dict[str, Any]) -> list[Any]:
|
||||||
"""Handle tool:result — emit TOOL_CALL_RESULT with optional MCP App metadata."""
|
"""Handle tool:result — emit TOOL_CALL_RESULT with optional MCP App metadata."""
|
||||||
|
|||||||
@@ -4,6 +4,17 @@ bundle:
|
|||||||
version: 0.1.0
|
version: 0.1.0
|
||||||
description: General-purpose AI research tool with browser control and artifact generation
|
description: General-purpose AI research tool with browser control and artifact generation
|
||||||
|
|
||||||
|
providers:
|
||||||
|
- module: provider-anthropic
|
||||||
|
source: git+https://github.com/microsoft/amplifier-module-provider-anthropic@main
|
||||||
|
config:
|
||||||
|
default_model: claude-sonnet-4-20250514
|
||||||
|
|
||||||
|
orchestrator:
|
||||||
|
module: loop-streaming
|
||||||
|
config:
|
||||||
|
extended_thinking: false
|
||||||
|
|
||||||
includes:
|
includes:
|
||||||
- bundle: git+https://github.com/microsoft/amplifier-foundation@main
|
- bundle: git+https://github.com/microsoft/amplifier-foundation@main
|
||||||
- bundle: research-workbench:behaviors/research-workbench
|
- bundle: research-workbench:behaviors/research-workbench
|
||||||
|
|||||||
@@ -95,7 +95,7 @@ export function useChat(sessionId: string | null) {
|
|||||||
break
|
break
|
||||||
}
|
}
|
||||||
case 'TEXT_MESSAGE_END': {
|
case 'TEXT_MESSAGE_END': {
|
||||||
// no-op
|
assistantId = null
|
||||||
break
|
break
|
||||||
}
|
}
|
||||||
case 'TOOL_CALL_START': {
|
case 'TOOL_CALL_START': {
|
||||||
|
|||||||
+125
-17
@@ -18,21 +18,17 @@ class TestTextBlocks:
|
|||||||
)
|
)
|
||||||
|
|
||||||
def test_text_block_start(self):
|
def test_text_block_start(self):
|
||||||
"""text_block_start → TEXT_MESSAGE_START with role='assistant' and messageId."""
|
"""text_block_start → no events emitted (lazy: start is deferred to first delta)."""
|
||||||
event = {
|
event = {
|
||||||
"event_type": "content_block:start",
|
"event_type": "content_block:start",
|
||||||
"data": {"index": 0, "type": "text"},
|
"data": {"index": 0, "type": "text"},
|
||||||
}
|
}
|
||||||
result = self.translator.translate(event)
|
result = self.translator.translate(event)
|
||||||
assert len(result) == 1
|
assert result == []
|
||||||
ev = result[0]
|
|
||||||
assert ev.type.value == "TEXT_MESSAGE_START"
|
|
||||||
assert ev.role == "assistant"
|
|
||||||
assert ev.message_id # must have a non-empty message_id
|
|
||||||
|
|
||||||
def test_text_delta(self):
|
def test_text_delta(self):
|
||||||
"""text_delta → TEXT_MESSAGE_CONTENT with delta='Hello '."""
|
"""First text_delta → TEXT_MESSAGE_START then TEXT_MESSAGE_CONTENT (lazy start)."""
|
||||||
# Start the text block first
|
# Start the text block — no event emitted yet
|
||||||
start_event = {
|
start_event = {
|
||||||
"event_type": "content_block:start",
|
"event_type": "content_block:start",
|
||||||
"data": {"index": 0, "type": "text"},
|
"data": {"index": 0, "type": "text"},
|
||||||
@@ -47,19 +43,27 @@ class TestTextBlocks:
|
|||||||
},
|
},
|
||||||
}
|
}
|
||||||
result = self.translator.translate(delta_event)
|
result = self.translator.translate(delta_event)
|
||||||
assert len(result) == 1
|
# Lazy start: first delta emits [TEXT_MESSAGE_START, TEXT_MESSAGE_CONTENT]
|
||||||
ev = result[0]
|
assert len(result) == 2
|
||||||
assert ev.type.value == "TEXT_MESSAGE_CONTENT"
|
start_ev, content_ev = result
|
||||||
assert ev.delta == "Hello "
|
assert start_ev.type.value == "TEXT_MESSAGE_START"
|
||||||
|
assert start_ev.role == "assistant"
|
||||||
|
assert start_ev.message_id
|
||||||
|
assert content_ev.type.value == "TEXT_MESSAGE_CONTENT"
|
||||||
|
assert content_ev.delta == "Hello "
|
||||||
|
assert content_ev.message_id == start_ev.message_id
|
||||||
|
|
||||||
def test_text_block_end(self):
|
def test_text_block_end(self):
|
||||||
"""text_block_end → TEXT_MESSAGE_END."""
|
"""text_block_end after content → TEXT_MESSAGE_END with matching messageId."""
|
||||||
# Start the block first
|
# Start block, send a delta (which lazily opens it), then close it
|
||||||
start_event = {
|
self.translator.translate({
|
||||||
"event_type": "content_block:start",
|
"event_type": "content_block:start",
|
||||||
"data": {"index": 0, "type": "text"},
|
"data": {"index": 0, "type": "text"},
|
||||||
}
|
})
|
||||||
self.translator.translate(start_event)
|
self.translator.translate({
|
||||||
|
"event_type": "content_block:delta",
|
||||||
|
"data": {"index": 0, "delta": {"type": "text_delta", "text": "Hi"}},
|
||||||
|
})
|
||||||
|
|
||||||
end_event = {
|
end_event = {
|
||||||
"event_type": "content_block:end",
|
"event_type": "content_block:end",
|
||||||
@@ -195,6 +199,110 @@ class TestToolBlocks:
|
|||||||
assert "structuredContent" not in d
|
assert "structuredContent" not in d
|
||||||
|
|
||||||
|
|
||||||
|
class TestExtendedThinking:
|
||||||
|
"""Tests for extended thinking blocks — must produce zero AG-UI events."""
|
||||||
|
|
||||||
|
def setup_method(self):
|
||||||
|
self.translator = AmplifierdEventTranslator(
|
||||||
|
run_id="run_1", thread_id="thread_1"
|
||||||
|
)
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# helpers
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
|
def _thinking_start(self, index: int = 0) -> dict:
|
||||||
|
return {
|
||||||
|
"event_type": "content_block:start",
|
||||||
|
"data": {"index": index, "type": "thinking"},
|
||||||
|
}
|
||||||
|
|
||||||
|
def _thinking_delta(self, text: str, index: int = 0) -> dict:
|
||||||
|
return {
|
||||||
|
"event_type": "content_block:delta",
|
||||||
|
"data": {"index": index, "delta": {"type": "thinking_delta", "thinking": text}},
|
||||||
|
}
|
||||||
|
|
||||||
|
def _thinking_end(self, index: int = 0) -> dict:
|
||||||
|
return {"event_type": "content_block:end", "data": {"index": index}}
|
||||||
|
|
||||||
|
def _text_start(self, index: int = 1) -> dict:
|
||||||
|
return {
|
||||||
|
"event_type": "content_block:start",
|
||||||
|
"data": {"index": index, "type": "text"},
|
||||||
|
}
|
||||||
|
|
||||||
|
def _text_delta(self, text: str, index: int = 1) -> dict:
|
||||||
|
return {
|
||||||
|
"event_type": "content_block:delta",
|
||||||
|
"data": {"index": index, "delta": {"type": "text_delta", "text": text}},
|
||||||
|
}
|
||||||
|
|
||||||
|
def _text_end(self, index: int = 1) -> dict:
|
||||||
|
return {"event_type": "content_block:end", "data": {"index": index}}
|
||||||
|
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
# tests
|
||||||
|
# ------------------------------------------------------------------
|
||||||
|
|
||||||
|
def test_thinking_block_produces_no_events(self):
|
||||||
|
"""thinking start/delta/end → all return [] (thinking blocks are skipped)."""
|
||||||
|
assert self.translator.translate(self._thinking_start()) == []
|
||||||
|
assert self.translator.translate(self._thinking_delta("Let me reason…")) == []
|
||||||
|
assert self.translator.translate(self._thinking_end()) == []
|
||||||
|
|
||||||
|
def test_empty_text_block_produces_no_events(self):
|
||||||
|
"""text start then immediate end with no deltas → no AG-UI events emitted."""
|
||||||
|
assert self.translator.translate(self._text_start()) == []
|
||||||
|
assert self.translator.translate(self._text_end()) == []
|
||||||
|
|
||||||
|
def test_thinking_then_empty_text_produces_no_events(self):
|
||||||
|
"""Full sequence: thinking block + empty text block → zero AG-UI events total."""
|
||||||
|
events = [
|
||||||
|
self._thinking_start(index=0),
|
||||||
|
self._thinking_delta("I need to think…", index=0),
|
||||||
|
self._thinking_end(index=0),
|
||||||
|
self._text_start(index=1),
|
||||||
|
self._text_end(index=1),
|
||||||
|
]
|
||||||
|
all_results = []
|
||||||
|
for ev in events:
|
||||||
|
all_results.extend(self.translator.translate(ev))
|
||||||
|
assert all_results == []
|
||||||
|
|
||||||
|
def test_thinking_then_nonempty_text_works(self):
|
||||||
|
"""Thinking block (skip) then text block with actual content → START + CONTENT + END."""
|
||||||
|
# Thinking block — all silent
|
||||||
|
self.translator.translate(self._thinking_start(index=0))
|
||||||
|
self.translator.translate(self._thinking_delta("Reasoning…", index=0))
|
||||||
|
self.translator.translate(self._thinking_end(index=0))
|
||||||
|
|
||||||
|
# Text block — lazy open on first delta
|
||||||
|
self.translator.translate(self._text_start(index=1))
|
||||||
|
|
||||||
|
first_delta_result = self.translator.translate(self._text_delta("Hello", index=1))
|
||||||
|
assert len(first_delta_result) == 2
|
||||||
|
start_ev, content_ev = first_delta_result
|
||||||
|
assert start_ev.type.value == "TEXT_MESSAGE_START"
|
||||||
|
assert start_ev.role == "assistant"
|
||||||
|
message_id = start_ev.message_id
|
||||||
|
assert content_ev.type.value == "TEXT_MESSAGE_CONTENT"
|
||||||
|
assert content_ev.delta == "Hello"
|
||||||
|
assert content_ev.message_id == message_id
|
||||||
|
|
||||||
|
# Subsequent delta — only CONTENT (no second START)
|
||||||
|
second_delta_result = self.translator.translate(self._text_delta(" world", index=1))
|
||||||
|
assert len(second_delta_result) == 1
|
||||||
|
assert second_delta_result[0].type.value == "TEXT_MESSAGE_CONTENT"
|
||||||
|
assert second_delta_result[0].delta == " world"
|
||||||
|
|
||||||
|
# End — TEXT_MESSAGE_END because content was started
|
||||||
|
end_result = self.translator.translate(self._text_end(index=1))
|
||||||
|
assert len(end_result) == 1
|
||||||
|
assert end_result[0].type.value == "TEXT_MESSAGE_END"
|
||||||
|
assert end_result[0].message_id == message_id
|
||||||
|
|
||||||
|
|
||||||
class TestLifecycle:
|
class TestLifecycle:
|
||||||
"""Tests for lifecycle events → AG-UI run lifecycle events."""
|
"""Tests for lifecycle events → AG-UI run lifecycle events."""
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user