refactor(agent): move feedback issue flow into skill scripts

This commit is contained in:
jxxghp
2026-05-21 19:22:27 +08:00
parent b6b5529d19
commit 737bcb5c62
16 changed files with 1683 additions and 4242 deletions

View File

@@ -8,11 +8,6 @@ from app.agent.tools.impl.ask_user_choice import (
AskUserChoiceTool,
UserChoiceOptionInput,
)
from app.agent.tools.impl.feedback_issue_state import (
FEEDBACK_CONFIRM_VALUE_PREFIX,
build_feedback_draft_hash,
feedback_issue_state_store,
)
from app.helper.interaction import (
AgentInteractionOption,
agent_interaction_manager,
@@ -24,7 +19,6 @@ from app.schemas.types import MessageChannel
class TestAgentInteraction(unittest.TestCase):
def tearDown(self):
agent_interaction_manager.clear()
feedback_issue_state_store.clear()
def test_prompt_injects_choice_tool_hint_only_for_button_channels(self):
telegram_prompt = prompt_manager.get_agent_prompt(
@@ -133,39 +127,6 @@ class TestAgentInteraction(unittest.TestCase):
self.assertIn("质量门槛拒绝", result)
async_post_message.assert_not_awaited()
def test_choice_tool_blocks_after_feedback_preview_pending(self):
"""#5807 回归prepare_feedback_issue 发完按钮后agent 不应再叠 ask_user_choice。
否则用户会收到两个确认按钮、点两次、agent 跑两轮 → 同一条成功
文案在 TG 里重复 3 次。"""
tool = AskUserChoiceTool(session_id="session-feedback", user_id="10001")
tool.set_message_attr(
channel=MessageChannel.Telegram.value,
source="telegram-test",
username="tester",
)
tool.set_agent_context(
agent_context={"reply_mode": "feedback_issue_confirmation"}
)
with patch(
"app.agent.tools.impl.ask_user_choice.ToolChain.async_post_message",
new=AsyncMock(),
) as async_post_message:
result = asyncio.run(
tool.run(
message="已准备 ISSUE请确认是否提交到上游仓库",
options=[
UserChoiceOptionInput(label="确认提交", value="确认提交"),
UserChoiceOptionInput(label="取消", value="取消"),
],
)
)
# 工具应该自我拒绝,不再发第二个按钮卡片
self.assertIn("prepare_feedback_issue", result)
async_post_message.assert_not_awaited()
def test_agent_interaction_callback_routes_selected_value_back_to_agent(self):
chain = MessageChain()
request = agent_interaction_manager.create_request(
@@ -212,103 +173,6 @@ class TestAgentInteraction(unittest.TestCase):
message_put.assert_called_once()
message_add.assert_called_once()
def test_feedback_confirmation_callback_marks_token_confirmed(self):
draft_hash = build_feedback_draft_hash(
title="[错误报告]: 订阅刷新接口返回 500 错误码",
version="v2.12.2",
environment="Docker",
issue_type="主程序运行问题",
description="## 现象\n错误\n## 复现步骤\n点击刷新\n## 期望行为\n正常刷新",
original_user_request="订阅刷新接口返回 500",
logs="ERROR demo",
diagnostics_id="diag-1",
)
confirmation = feedback_issue_state_store.create_confirmation(
session_id="session-feedback",
user_id="10001",
username="tester",
draft_hash=draft_hash,
diagnostics_id="diag-1",
)
request = agent_interaction_manager.create_request(
session_id="session-feedback",
user_id="10001",
channel=MessageChannel.Telegram.value,
source="telegram-test",
username="tester",
title="确认提交问题反馈",
prompt="请确认",
options=[
AgentInteractionOption(
label="确认提交",
value=f"{FEEDBACK_CONFIRM_VALUE_PREFIX}{confirmation.confirmation_token}",
)
],
)
chain = MessageChain()
with patch.object(chain, "_handle_ai_message") as handle_ai_message, patch.object(
chain.messagehelper, "put"
), patch.object(chain.messageoper, "add"), patch.object(
chain, "edit_message", return_value=True
):
chain._handle_callback(
text=f"CALLBACK:agent_interaction:choice:{request.request_id}:1",
channel=MessageChannel.Telegram,
source="telegram-test",
userid="10001",
username="tester",
)
kwargs = handle_ai_message.call_args.kwargs
self.assertIn("confirmation_token", kwargs["text"])
consumed = feedback_issue_state_store.consume_confirmed(
confirmation.confirmation_token,
session_id="session-feedback",
user_id="10001",
draft_hash=draft_hash,
)
self.assertIsNotNone(consumed)
def test_state_store_active_confirmation_helpers(self):
# find_active_confirmation 应只返回 confirmed_at=None 的记录
rec1 = feedback_issue_state_store.create_confirmation(
session_id="s1", user_id="u1", username=None,
draft_hash="h1", diagnostics_id="d1",
)
rec2 = feedback_issue_state_store.create_confirmation(
session_id="s1", user_id="u2", username=None,
draft_hash="h2", diagnostics_id="d2",
)
# 跨用户隔离
self.assertEqual(
feedback_issue_state_store.find_active_confirmation(
session_id="s1", user_id="u1"
).confirmation_token,
rec1.confirmation_token,
)
# 标记为已确认后不应再被 active 检索返回
feedback_issue_state_store.mark_confirmed(
rec1.confirmation_token, session_id="s1", user_id="u1"
)
self.assertIsNone(
feedback_issue_state_store.find_active_confirmation(
session_id="s1", user_id="u1"
)
)
# invalidate_active_confirmations 只清掉当前会话+用户的 pending 记录
dropped = feedback_issue_state_store.invalidate_active_confirmations(
session_id="s1", user_id="u2"
)
self.assertEqual(dropped, 1)
self.assertIsNone(
feedback_issue_state_store.find_active_confirmation(
session_id="s1", user_id="u2"
)
)
# 已 confirmed 的 rec1 不应该被这次 invalidate 误删
self.assertIn(rec1.confirmation_token, feedback_issue_state_store._confirmations)
def test_legacy_agent_choice_callback_still_supported(self):
chain = MessageChain()
request = agent_interaction_manager.create_request(