636 lines
27 KiB
Python
636 lines
27 KiB
Python
"""RC-BE-004 gateway tests for /api/report-collaboration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.gateway.routers import intent, report_collaboration
|
|
from app.gateway.routers._roundtable_seed import REPORT_COLLABORATION_INTENT_AGENT_ID
|
|
from app.report_collaboration.execution.live_hub import PublishingReportCollaborationStore, ReportCollaborationLiveHub, SseStreamSettings
|
|
from deerflow.config.model_config import ModelConfig
|
|
from deerflow.config.report_collaboration_config import ReportCollaborationConfig
|
|
from deerflow.persistence.report_collaboration import MemoryReportCollaborationStore
|
|
|
|
|
|
class _Cfg:
|
|
report_collaboration = ReportCollaborationConfig(enabled=True)
|
|
models = [ModelConfig(name="demo-chat", use="langchain_openai:ChatOpenAI", model="demo")]
|
|
|
|
def get_model_config(self, name: str):
|
|
return next((item for item in self.models if item.name == name), None)
|
|
|
|
|
|
class _DisabledCfg:
|
|
report_collaboration = ReportCollaborationConfig(enabled=False)
|
|
|
|
|
|
class _PlannerFakeLLM:
|
|
"""规划师假模型：需求确认回「读者」追问；工作流设计回单份定制方案。"""
|
|
|
|
_converse = (
|
|
'{"reply": "好的。先确认：这份报告的主要读者是谁？", "action": "clarify",'
|
|
'"questions": ["这份报告的主要读者是谁？"], "impact": ["读者影响研究线划分"],'
|
|
'"requirement": {"topic": "新能源汽车趋势报告"}}'
|
|
)
|
|
_ready = (
|
|
'{"reply": "明白了，读者为企业战略决策人员。我来设计工作流。", "action": "ready",'
|
|
'"assumptions": [], "requirement": {"topic": "新能源汽车趋势报告", "audience": "企业战略决策人员"}}'
|
|
)
|
|
_design = (
|
|
'{"reply": "已按需求设计定制工作流。", "strategy": "custom", "title": "定制方案",'
|
|
'"summary": "研究后写作", "rationale": "贴合需求", "estimated_duration_seconds": 900,'
|
|
'"estimated_cost_level": "low", "quality_gate_ids": ["cite_required"],'
|
|
'"nodes": ['
|
|
'{"slot_id": "research-1", "role_key": "researcher", "display_name": "研究员", "mission": "检索证据", "depends_on": [], "output_artifact_type": "EvidenceBundle"},'
|
|
'{"slot_id": "write-1", "role_key": "writer", "display_name": "撰写员", "mission": "成稿", "depends_on": ["research-1"], "output_artifact_type": "ReportSectionDraft"}]}'
|
|
)
|
|
|
|
async def ainvoke(self, messages):
|
|
from types import SimpleNamespace
|
|
|
|
prompt = str(messages[-1][1]) if messages else ""
|
|
if "请设计工作流" in prompt:
|
|
payload = self._design
|
|
elif "企业战略决策人员" in prompt or "读者" in prompt and "用户最新消息" in prompt:
|
|
payload = self._ready
|
|
else:
|
|
payload = self._converse
|
|
return SimpleNamespace(content=payload)
|
|
|
|
async def astream(self, messages):
|
|
from types import SimpleNamespace
|
|
|
|
yield SimpleNamespace(
|
|
content="",
|
|
additional_kwargs={"reasoning_content": "正在比较可用智能体的能力与依赖关系。"},
|
|
)
|
|
yield SimpleNamespace(content=self._design, additional_kwargs={})
|
|
|
|
|
|
class _BrokenPlannerStream:
|
|
async def astream(self, messages):
|
|
if False:
|
|
yield None
|
|
raise RuntimeError("provider stream closed early")
|
|
|
|
|
|
class _ReasoningOnlyPlannerStream:
|
|
async def astream(self, messages):
|
|
from types import SimpleNamespace
|
|
|
|
yield SimpleNamespace(
|
|
content="",
|
|
additional_kwargs={"reasoning_content": _PlannerFakeLLM._design},
|
|
)
|
|
|
|
|
|
def test_report_intent_agent_contract_supports_nested_requirement() -> None:
|
|
assert (
|
|
intent._requested_intent_agent_id(
|
|
{"intent_agent_id": REPORT_COLLABORATION_INTENT_AGENT_ID}
|
|
)
|
|
== REPORT_COLLABORATION_INTENT_AGENT_ID
|
|
)
|
|
content = (
|
|
"意图已经明确。\n[INTENT_READY]\n```json\n"
|
|
'{"objective":"写报告","constraints":[],"assumptions":[],'
|
|
'"report_requirement":{"topic":"新能源汽车","required_angles":["市场"]}}\n```'
|
|
)
|
|
parsed = intent._extract_intent_ready(content)
|
|
assert parsed is not None
|
|
assert parsed[0]["report_requirement"]["topic"] == "新能源汽车"
|
|
assert content[: parsed[1]].endswith("```")
|
|
|
|
|
|
def _client(monkeypatch: pytest.MonkeyPatch, *, enabled: bool = True, user: str | None = "user_a") -> tuple[TestClient, MemoryReportCollaborationStore]:
|
|
monkeypatch.setattr(report_collaboration, "get_app_config", lambda: _Cfg() if enabled else _DisabledCfg())
|
|
app = FastAPI()
|
|
hub = ReportCollaborationLiveHub()
|
|
inner = MemoryReportCollaborationStore()
|
|
store = PublishingReportCollaborationStore(inner, hub)
|
|
app.state.report_collaboration_store = store
|
|
app.state.report_collaboration_planner_llm = _PlannerFakeLLM
|
|
app.state.report_collaboration_live_hub = hub
|
|
app.state.report_collaboration_sse_settings = SseStreamSettings(
|
|
deadline_seconds=0.6,
|
|
live_wait_seconds=0.05,
|
|
db_poll_seconds=0.2,
|
|
heartbeat_seconds=30,
|
|
flush_padding="",
|
|
)
|
|
app.include_router(report_collaboration.router)
|
|
|
|
async def _fake_user(_request=None):
|
|
return user
|
|
|
|
monkeypatch.setattr(report_collaboration, "get_current_user", _fake_user)
|
|
return TestClient(app), store
|
|
|
|
|
|
def _stream_plans(
|
|
client: TestClient,
|
|
session_id: str,
|
|
key: str,
|
|
adjustment_text: str | None = None,
|
|
) -> str:
|
|
with client.stream(
|
|
"POST",
|
|
f"/api/report-collaboration/sessions/{session_id}/plan-stream",
|
|
json={"adjustment_text": adjustment_text},
|
|
headers={"X-Idempotency-Key": key},
|
|
) as response:
|
|
assert response.status_code == 200, response.text
|
|
assert response.headers["content-type"].startswith("text/event-stream")
|
|
return "".join(response.iter_text())
|
|
|
|
|
|
def _ready_session(client: TestClient, *, key_prefix: str) -> str:
|
|
session = client.post(
|
|
"/api/report-collaboration/sessions",
|
|
json={"title": "规划恢复测试"},
|
|
headers={"X-Idempotency-Key": f"{key_prefix}-session"},
|
|
).json()
|
|
session_id = session["id"]
|
|
response = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/intent-turns",
|
|
json={
|
|
"text": "规划一份新能源汽车市场报告",
|
|
"assistant_content": "意图已经确认。",
|
|
"intent_thread_id": f"{key_prefix}-thread",
|
|
"status": "done",
|
|
"summary": {
|
|
"objective": "形成新能源汽车市场判断",
|
|
"constraints": [],
|
|
"assumptions": [],
|
|
"report_requirement": {
|
|
"topic": "新能源汽车市场报告",
|
|
"required_angles": ["市场规模", "竞争格局"],
|
|
"excluded_angles": [],
|
|
},
|
|
},
|
|
"clarifications": [],
|
|
},
|
|
headers={"X-Idempotency-Key": f"{key_prefix}-intent"},
|
|
)
|
|
assert response.status_code == 200
|
|
return session_id
|
|
|
|
|
|
def test_plan_stream_recovers_to_candidates_when_provider_stream_fails(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, _store = _client(monkeypatch)
|
|
client.app.state.report_collaboration_planner_llm = _BrokenPlannerStream
|
|
session_id = _ready_session(client, key_prefix="recover")
|
|
|
|
stream = _stream_plans(client, session_id, "recover-plan")
|
|
|
|
assert "event: phase" in stream
|
|
assert "正在收敛为可执行候选工作流" in stream
|
|
assert "event: complete" in stream
|
|
snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
|
|
assert snapshot["session"]["status"] == "proposal_ready"
|
|
assert len(snapshot["plans"]) == 3
|
|
|
|
|
|
def test_plan_stream_accepts_final_json_from_reasoning_channel(
|
|
monkeypatch: pytest.MonkeyPatch,
|
|
) -> None:
|
|
client, _store = _client(monkeypatch)
|
|
client.app.state.report_collaboration_planner_llm = _ReasoningOnlyPlannerStream
|
|
session_id = _ready_session(client, key_prefix="reasoning-json")
|
|
|
|
stream = _stream_plans(client, session_id, "reasoning-json-plan")
|
|
|
|
assert "event: reasoning" in stream
|
|
assert "event: complete" in stream
|
|
assert "正在收敛为可执行候选工作流" not in stream
|
|
snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
|
|
assert snapshot["session"]["status"] == "proposal_ready"
|
|
assert any(plan["strategy"] == "custom" for plan in snapshot["plans"])
|
|
|
|
|
|
def test_disabled_flag_returns_503(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client, _ = _client(monkeypatch, enabled=False)
|
|
res = client.get("/api/report-collaboration/sessions")
|
|
assert res.status_code == 503
|
|
assert res.json()["code"] == "REPORT_COLLABORATION_DISABLED"
|
|
|
|
|
|
def test_unauthenticated_returns_401(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client, _ = _client(monkeypatch, user=None)
|
|
res = client.get("/api/report-collaboration/sessions")
|
|
assert res.status_code == 401
|
|
|
|
|
|
def test_session_message_plan_run_http(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client, store = _client(monkeypatch)
|
|
|
|
missing_key = client.post("/api/report-collaboration/sessions", json={"title": "新能源"})
|
|
assert missing_key.status_code == 422
|
|
|
|
created = client.post(
|
|
"/api/report-collaboration/sessions",
|
|
json={"title": "新能源"},
|
|
headers={"X-Idempotency-Key": "k-create"},
|
|
)
|
|
assert created.status_code == 200, created.text
|
|
session = created.json()
|
|
session_id = session["id"]
|
|
assert session["status"] == "empty"
|
|
|
|
replay = client.post(
|
|
"/api/report-collaboration/sessions",
|
|
json={"title": "other"},
|
|
headers={"X-Idempotency-Key": "k-create"},
|
|
)
|
|
assert replay.json()["id"] == session_id
|
|
|
|
listed = client.get("/api/report-collaboration/sessions")
|
|
assert listed.status_code == 200
|
|
assert listed.json()["sessions"][0]["id"] == session_id
|
|
|
|
sent = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/messages",
|
|
json={"text": "写一份新能源汽车趋势报告"},
|
|
headers={"X-Idempotency-Key": "k-msg"},
|
|
)
|
|
assert sent.status_code == 409
|
|
assert sent.json()["code"] == "USE_STREAMING_INTENT"
|
|
|
|
streamed_turn = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/intent-turns",
|
|
json={
|
|
"text": "写一份新能源汽车趋势报告",
|
|
"assistant_content": "<think>需要确认目标读者</think>请问这份报告主要给谁看？",
|
|
"intent_thread_id": "intent-thread-1",
|
|
"status": "clarification",
|
|
"summary": None,
|
|
"clarifications": [
|
|
{
|
|
"id": "clar-1",
|
|
"question": "这份报告主要给谁看？",
|
|
"clarification_type": "missing_info",
|
|
"options": ["企业战略决策人员", "投资机构"],
|
|
}
|
|
],
|
|
},
|
|
headers={"X-Idempotency-Key": "k-intent-turn-1"},
|
|
)
|
|
assert streamed_turn.status_code == 200, streamed_turn.text
|
|
|
|
snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}")
|
|
assert snapshot.status_code == 200
|
|
body = snapshot.json()
|
|
assert body["session"]["status"] == "clarifying"
|
|
assert body["session"]["requirement_revision"] == 0
|
|
assert body["messages"][0]["content"].startswith("写一份")
|
|
assert any(item["role"] == "ai" and "给谁看" in item["content"] for item in body["messages"])
|
|
intent_ai = next(item for item in body["messages"] if item["role"] == "ai")
|
|
assert intent_ai["metadata"]["reasoning_content"] == "需要确认目标读者"
|
|
assert intent_ai["tool_calls"][0]["name"] == "ask_clarification"
|
|
clarification_tool = next(item for item in body["messages"] if item["role"] == "tool")
|
|
assert clarification_tool["name"] == "ask_clarification"
|
|
assert clarification_tool["tool_call_id"] == "clar-1"
|
|
assert clarification_tool["metadata"]["clarification"]["question"]
|
|
assert body["last_seq"] == 0
|
|
assert body["plans"] == []
|
|
|
|
# 即使绕过前端直接调用兼容接口，也不能在规划师确认意图前抢跑工作流规划。
|
|
premature = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/plan-requests",
|
|
headers={"X-Idempotency-Key": "k-plan-premature", "X-Expected-Revision": "0"},
|
|
)
|
|
assert premature.status_code == 422
|
|
assert premature.json()["code"] == "REQUIREMENT_NOT_READY"
|
|
|
|
# 用户回答完意图规划师的澄清问题后，接口只落意图快照并切到 planning。
|
|
# 工作流必须由独立规划智能体通过 SSE 运行，让客户端能实时显示过程。
|
|
clarified = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/intent-turns",
|
|
json={
|
|
"text": "主要给企业战略决策人员看，重点关注市场格局和价格策略",
|
|
"assistant_content": "明白，我会按这个意图规划报告。\n[INTENT_READY]\n```json\n{}\n```",
|
|
"intent_thread_id": "intent-thread-1",
|
|
"status": "done",
|
|
"summary": {
|
|
"objective": "撰写新能源汽车趋势报告",
|
|
"constraints": [],
|
|
"assumptions": ["中文输出"],
|
|
"report_requirement": {
|
|
"topic": "新能源汽车趋势报告",
|
|
"objective": "支持企业战略决策",
|
|
"audience": "企业战略决策人员",
|
|
"language": "中文",
|
|
"required_angles": ["市场格局", "价格策略"],
|
|
"excluded_angles": [],
|
|
},
|
|
},
|
|
"clarifications": [],
|
|
},
|
|
headers={"X-Idempotency-Key": "k-intent-turn-2"},
|
|
)
|
|
assert clarified.status_code == 200, clarified.text
|
|
|
|
planning_body = clarified.json()
|
|
assert planning_body["session"]["status"] == "planning"
|
|
assert planning_body["plans"] == []
|
|
|
|
planning_stream = _stream_plans(client, session_id, "k-plan-stream")
|
|
assert "event: phase" in planning_stream
|
|
assert "正在读取可用智能体及能力描述" in planning_stream
|
|
assert "event: reasoning" in planning_stream
|
|
assert "正在比较可用智能体的能力与依赖关系" in planning_stream
|
|
assert "event: complete" in planning_stream
|
|
|
|
proposed = client.get(f"/api/report-collaboration/sessions/{session_id}")
|
|
assert proposed.status_code == 200
|
|
proposed_body = proposed.json()
|
|
assert proposed_body["session"]["status"] == "proposal_ready"
|
|
revision = proposed_body["session"]["requirement_revision"]
|
|
assert revision == 1
|
|
assert proposed_body["messages"][-2]["content"] == "明白，我会按这个意图规划报告。"
|
|
planner_message = next(
|
|
item for item in proposed_body["messages"] if item.get("name") == "工作流规划师"
|
|
)
|
|
assert "正在比较可用智能体" in planner_message["metadata"]["reasoning_content"]
|
|
assert [item["name"] for item in planner_message["tool_calls"]] == [
|
|
"workflow_planning_progress",
|
|
"workflow_planning_progress",
|
|
"workflow_planning_progress",
|
|
]
|
|
plans = proposed_body["plans"]
|
|
assert {item["strategy"] for item in plans} == {"custom", "balanced_review", "parallel_depth"}
|
|
plan = next(item for item in plans if item.get("recommended"))
|
|
assert plan["validation"]["ok"] is True
|
|
assert all(node.get("output_artifact_type") and node.get("acceptance_criteria") for node in plan["nodes"])
|
|
|
|
adjusted = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/messages",
|
|
json={"text": "以第二个方案为基础，去掉审核智能体，再增加行业分析智能体"},
|
|
headers={"X-Idempotency-Key": "k-adjust-plan"},
|
|
)
|
|
assert adjusted.status_code == 200, adjusted.text
|
|
adjustment_stream = _stream_plans(
|
|
client,
|
|
session_id,
|
|
"k-adjust-plan-stream",
|
|
"以第二个方案为基础，去掉审核智能体，再增加行业分析智能体",
|
|
)
|
|
assert "event: complete" in adjustment_stream
|
|
adjusted_body = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
|
|
active_plans = [item for item in adjusted_body["plans"] if item["status"] == "proposed"]
|
|
assert len(active_plans) == 3
|
|
assert active_plans[0]["proposal_group_id"] != plan["proposal_group_id"]
|
|
assert any("第二个方案" in item["content"] for item in adjusted_body["messages"])
|
|
plan = next(item for item in active_plans if item.get("recommended"))
|
|
|
|
stale = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/plan-requests",
|
|
headers={"X-Idempotency-Key": "k-plan-stale", "X-Expected-Revision": "9"},
|
|
)
|
|
assert stale.status_code == 409
|
|
assert stale.json()["code"] == "STALE_REVISION"
|
|
|
|
listed = client.get(f"/api/report-collaboration/sessions/{session_id}/plans")
|
|
assert listed.status_code == 200
|
|
assert len(listed.json()["plans"]) == 6
|
|
|
|
selected = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/plans/{plan['id']}/select",
|
|
headers={"X-Idempotency-Key": "k-select", "X-Expected-Revision": str(revision)},
|
|
)
|
|
assert selected.status_code == 200, selected.text
|
|
assert selected.json()["selected_plan_id"] == plan["id"]
|
|
assert client.get(f"/api/report-collaboration/sessions/{session_id}").json()["run"] is None
|
|
|
|
started = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/runs",
|
|
json={"plan_id": plan["id"]},
|
|
headers={"X-Idempotency-Key": "k-run", "X-Expected-Revision": str(revision)},
|
|
)
|
|
assert started.status_code == 200, started.text
|
|
run = started.json()
|
|
assert run["status"] == "queued"
|
|
run_id = run["id"]
|
|
|
|
conflict = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/runs",
|
|
json={"plan_id": plan["id"]},
|
|
headers={"X-Idempotency-Key": "k-run-2", "X-Expected-Revision": str(revision)},
|
|
)
|
|
assert conflict.status_code == 409
|
|
|
|
got = client.get(f"/api/report-collaboration/runs/{run_id}")
|
|
assert got.status_code == 200
|
|
assert got.json()["id"] == run_id
|
|
|
|
events = client.get(f"/api/report-collaboration/runs/{run_id}/events?after_seq=0")
|
|
assert events.status_code == 200
|
|
recorded = events.json()["events"]
|
|
assert len(recorded) == 1
|
|
assert recorded[0]["type"] == "heartbeat"
|
|
assert recorded[0]["data"]["audit"] == "model_snapshot"
|
|
assert "api_key" not in str(recorded[0]["data"])
|
|
assert "sk-" not in str(recorded[0]["data"])
|
|
|
|
stream = client.stream("GET", f"/api/report-collaboration/runs/{run_id}/stream")
|
|
with stream as response:
|
|
assert response.status_code == 200, response.text
|
|
assert response.headers["content-type"].startswith("text/event-stream")
|
|
first = next(response.iter_text())
|
|
assert "stream-ready" in first or first.startswith(":") or "data:" in first
|
|
|
|
still_queued = client.get(f"/api/report-collaboration/runs/{run_id}")
|
|
assert still_queued.status_code == 200
|
|
assert still_queued.json()["status"] == "queued"
|
|
assert still_queued.json()["cancel_requested"] is False
|
|
|
|
contract = client.get(f"/api/report-collaboration/runs/{run_id}/nodes/{plan['nodes'][0]['id']}/contract")
|
|
assert contract.status_code == 200
|
|
assert contract.json()["mission"]
|
|
assert "content" not in contract.json()
|
|
|
|
cancelled = client.post(
|
|
f"/api/report-collaboration/runs/{run_id}/cancel",
|
|
headers={"X-Idempotency-Key": "k-cancel"},
|
|
)
|
|
assert cancelled.status_code == 200
|
|
assert cancelled.json()["cancel_requested"] is True
|
|
assert cancelled.json()["status"] == "cancelled"
|
|
|
|
async def _other_user(_request=None):
|
|
return "user_b"
|
|
|
|
monkeypatch.setattr(report_collaboration, "get_current_user", _other_user)
|
|
hidden = client.get(f"/api/report-collaboration/sessions/{session_id}")
|
|
assert hidden.status_code == 404
|
|
|
|
async def _owner(_request=None):
|
|
return "user_a"
|
|
|
|
monkeypatch.setattr(report_collaboration, "get_current_user", _owner)
|
|
deleted = client.delete(f"/api/report-collaboration/sessions/{session_id}")
|
|
assert deleted.status_code == 204
|
|
|
|
|
|
def test_runtime_command_http_classify_confirm_cancel_and_idempotency(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client, _store = _client(monkeypatch)
|
|
session = client.post(
|
|
"/api/report-collaboration/sessions",
|
|
json={"title": "运行期命令"},
|
|
headers={"X-Idempotency-Key": "cmd-session"},
|
|
).json()
|
|
session_id = session["id"]
|
|
sent = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/intent-turns",
|
|
json={
|
|
"text": "写一份面向企业战略决策人员的新能源汽车市场份额趋势报告",
|
|
"assistant_content": "意图已明确。",
|
|
"intent_thread_id": "intent-thread-cmd",
|
|
"status": "done",
|
|
"summary": {
|
|
"objective": "撰写新能源汽车市场份额趋势报告",
|
|
"constraints": [],
|
|
"assumptions": [],
|
|
"report_requirement": {
|
|
"topic": "新能源汽车市场份额趋势报告",
|
|
"objective": "支持企业战略决策",
|
|
"audience": "企业战略决策人员",
|
|
"required_angles": ["市场份额"],
|
|
"excluded_angles": [],
|
|
},
|
|
},
|
|
"clarifications": [],
|
|
},
|
|
headers={"X-Idempotency-Key": "cmd-message"},
|
|
)
|
|
assert sent.status_code == 200
|
|
assert sent.json()["session"]["status"] == "planning"
|
|
command_planning_stream = _stream_plans(client, session_id, "cmd-plan-stream")
|
|
assert "event: complete" in command_planning_stream
|
|
ready_snapshot = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
|
|
revision = ready_snapshot["session"]["requirement_revision"]
|
|
assert ready_snapshot["session"]["status"] == "proposal_ready"
|
|
plan = next(item for item in ready_snapshot["plans"] if item["recommended"])
|
|
selected = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/plans/{plan['id']}/select",
|
|
headers={"X-Idempotency-Key": "cmd-select", "X-Expected-Revision": str(revision)},
|
|
)
|
|
assert selected.status_code == 200
|
|
run = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/runs",
|
|
json={"plan_id": plan["id"]},
|
|
headers={"X-Idempotency-Key": "cmd-run", "X-Expected-Revision": str(revision)},
|
|
).json()
|
|
run_id = run["id"]
|
|
target = plan["nodes"][0]["id"]
|
|
|
|
rerun = client.post(
|
|
f"/api/report-collaboration/runs/{run_id}/commands",
|
|
json={"text": "重新执行这个节点", "target_node_id": target},
|
|
headers={"X-Idempotency-Key": "cmd-rerun"},
|
|
)
|
|
assert rerun.status_code == 200, rerun.text
|
|
assert rerun.json()["intent"] == "rerun_node"
|
|
assert rerun.json()["status"] == "completed"
|
|
snapshot_after_command = client.get(f"/api/report-collaboration/sessions/{session_id}").json()
|
|
runtime_message = next(item for item in snapshot_after_command["messages"] if item.get("command_id") == rerun.json()["command_id"])
|
|
assert runtime_message["metadata"]["idempotency_key"] == "cmd-rerun"
|
|
replay = client.post(
|
|
f"/api/report-collaboration/runs/{run_id}/commands",
|
|
json={"text": "这次文本应该被幂等忽略", "target_node_id": target},
|
|
headers={"X-Idempotency-Key": "cmd-rerun"},
|
|
)
|
|
assert replay.json()["command_id"] == rerun.json()["command_id"]
|
|
|
|
replan = client.post(
|
|
f"/api/report-collaboration/runs/{run_id}/commands",
|
|
json={"text": "重新规划，换一套方案"},
|
|
headers={"X-Idempotency-Key": "cmd-replan"},
|
|
)
|
|
assert replan.status_code == 200
|
|
assert replan.json()["status"] == "confirmation_required"
|
|
confirmed = client.post(
|
|
f"/api/report-collaboration/runs/{run_id}/commands/{replan.json()['command_id']}/confirm",
|
|
headers={"X-Idempotency-Key": "cmd-confirm"},
|
|
)
|
|
assert confirmed.status_code == 200, confirmed.text
|
|
assert confirmed.json()["status"] == "completed"
|
|
|
|
ambiguous = client.post(
|
|
f"/api/report-collaboration/runs/{run_id}/commands",
|
|
json={"text": "这里不太对，改一下"},
|
|
headers={"X-Idempotency-Key": "cmd-ambiguous"},
|
|
)
|
|
assert ambiguous.status_code == 200
|
|
assert ambiguous.json()["status"] == "confirmation_required"
|
|
cancelled = client.post(
|
|
f"/api/report-collaboration/runs/{run_id}/commands/{ambiguous.json()['command_id']}/cancel",
|
|
headers={"X-Idempotency-Key": "cmd-cancel"},
|
|
)
|
|
assert cancelled.status_code == 200
|
|
assert cancelled.json()["status"] == "cancelled"
|
|
|
|
blocked = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/messages",
|
|
json={"text": "运行中不要走规划前接口"},
|
|
headers={"X-Idempotency-Key": "cmd-blocked-message"},
|
|
)
|
|
assert blocked.status_code == 409
|
|
assert blocked.json()["code"] == "RUN_ACTIVE"
|
|
|
|
events = client.get(f"/api/report-collaboration/runs/{run_id}/events?after_seq=0&limit=500").json()["events"]
|
|
rerun_classified = [
|
|
item
|
|
for item in events
|
|
if item["type"] == "command.classified" and item["data"]["command"]["command_id"] == rerun.json()["command_id"]
|
|
]
|
|
assert len(rerun_classified) == 1
|
|
assert any(
|
|
item["type"] == "command.cancelled" and item["data"]["command_id"] == cancelled.json()["command_id"]
|
|
for item in events
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rewrite_apply_restore_http(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
client, store = _client(monkeypatch)
|
|
created = client.post(
|
|
"/api/report-collaboration/sessions",
|
|
json={"title": "版本联调"},
|
|
headers={"X-Idempotency-Key": "http-rewrite-session"},
|
|
)
|
|
session_id = created.json()["id"]
|
|
await store.create_report_version(
|
|
session_id,
|
|
idempotency_key="http-initial",
|
|
markdown="# 标题\n\n# 市场分析\n\n市场变化，引用 c1。\n\n# 结论与局限\n\n证据有限。",
|
|
source_index=[{"id": "c1", "title": "月报", "added_at": "2026-09-06T00:00:00Z"}],
|
|
change_type="initial",
|
|
change_summary="首稿",
|
|
)
|
|
rewrite = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/report-rewrites",
|
|
json={"instruction": "语气更审慎", "section": "市场分析"},
|
|
headers={"X-Idempotency-Key": "http-rewrite"},
|
|
)
|
|
assert rewrite.status_code == 200, rewrite.text
|
|
applied = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/report-rewrites/{rewrite.json()['id']}/apply",
|
|
headers={"X-Idempotency-Key": "http-apply"},
|
|
)
|
|
assert applied.status_code == 200, applied.text
|
|
assert applied.json()["version"] == 2
|
|
versions = client.get(f"/api/report-collaboration/sessions/{session_id}/reports").json()["versions"]
|
|
initial_id = next(item["id"] for item in versions if item["version"] == 1)
|
|
restored = client.post(
|
|
f"/api/report-collaboration/sessions/{session_id}/report-versions/{initial_id}/restore",
|
|
headers={"X-Idempotency-Key": "http-restore"},
|
|
)
|
|
assert restored.status_code == 200, restored.text
|
|
assert restored.json()["version"] == 3
|
|
assert restored.json()["change_type"] == "restore"
|