"""RC-BE-014: runtime intent, impact analysis and safe-point commands.""" from __future__ import annotations import pytest from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.report_collaboration.contracts.commands import IntentDecision from app.report_collaboration.execution.ledger import TaskLedger, TaskResult from app.report_collaboration.interventions.impact_analyzer import ImpactAnalyzer from app.report_collaboration.interventions.intent_router import RuntimeIntentContext, RuntimeIntentRouterAgent from app.report_collaboration.interventions.service import RuntimeCommandService from deerflow.persistence.base import Base from deerflow.persistence.report_collaboration import ( MemoryReportCollaborationStore, ReportCollaborationConflictError, make_report_collaboration_store, ) def _plan() -> dict: nodes = [ _node("research", "researcher", "EvidenceBundle", []), _node("verify", "verifier", "VerificationResult", ["research"]), _node("analyze", "analyst", "AngleAnalysis", ["verify"]), _node("write", "writer", "ReportSectionDraft", ["analyze"]), _node("review", "reviewer", "ReviewDecision", ["write"]), ] return { "id": "plan_intervention", "proposal_group_id": "grp_intervention", "title": "运行期干预测试", "strategy": "quality", "summary": "完整质量链", "rationale": "测试", "recommended": True, "estimated_duration_seconds": 600, "estimated_cost_level": "medium", "requirement_revision": 1, "roles": [], "nodes": nodes, "edges": [ {"source": "research", "target": "verify"}, {"source": "verify", "target": "analyze"}, {"source": "analyze", "target": "write"}, {"source": "write", "target": "review"}, ], "quality_gates": [], "validation": {"ok": True, "errors": []}, "revision": 1, "status": "proposed", } def _node(node_id: str, role: str, artifact: str, deps: list[str]) -> dict: return { "id": node_id, "label": node_id, "role_key": role, "role_display_name": role, "mission": role, "output_artifact_type": artifact, "allowed_tools": ["web_search"] if role == "researcher" else [], "acceptance_criteria": [], "depends_on": deps, "max_attempts": 3, } async def _boot(store=None) -> tuple[object, str, str]: store = store or MemoryReportCollaborationStore() session = await store.create_session(owner_id="u1", title="新能源", idempotency_key="session") requirement = { "topic": "新能源汽车市场趋势", "audience": "企业战略决策人员", "required_angles": ["市场份额"], "excluded_angles": [], "revision": 1, } await store.save_requirement_state( session["id"], requirement_json={"snapshot": requirement, "clarifications": [], "assumptions": [], "ready_to_plan": True}, requirement_revision=1, status="proposal_ready", ) plan = await store.insert_plan(session["id"], _plan()) await store.select_plan(session["id"], plan["id"], idempotency_key="select", expected_revision=1) run = await store.create_run(session["id"], plan_id=plan["id"], idempotency_key="run", expected_revision=1) return store, session["id"], run["id"] @pytest.mark.asyncio async def test_intent_router_prioritizes_answer_to_waiting_node() -> None: router = RuntimeIntentRouterAgent() decision = await router.route( RuntimeIntentContext( text="数据口径按中国乘用车零售量", requirement={"topic": "新能源汽车", "revision": 1}, plan_nodes=_plan()["nodes"], target_node_ids=["analyze"], awaiting_node_ids=["analyze"], ) ) assert decision.intent == "answer_question" assert decision.confidence == 0.99 assert decision.requested_target_ids == ["analyze"] @pytest.mark.asyncio async def test_model_intent_cannot_expand_authorized_target() -> None: class _Model: async def classify(self, _context): return IntentDecision( intent="reanalyze", confidence=0.91, normalized_instruction="重新分析", scope="plan", requested_target_ids=["malicious-node"], requires_confirmation=False, cost_level="high", reason="model", ) decision = await RuntimeIntentRouterAgent(_Model()).route( RuntimeIntentContext( text="请重新审视这里", requirement={"topic": "新能源汽车", "revision": 1}, plan_nodes=_plan()["nodes"], target_node_ids=["analyze"], ) ) assert decision.requested_target_ids == ["analyze"] assert decision.scope == "artifact_lineage" assert decision.cost_level == "medium" assert "malicious-node" not in decision.requested_target_ids @pytest.mark.asyncio async def test_explicit_angle_text_resolves_one_parallel_branch() -> None: nodes = [ {**_node("research-market", "researcher", "EvidenceBundle", []), "angle": "市场份额", "label": "市场研究"}, {**_node("research-overseas", "researcher", "EvidenceBundle", []), "angle": "海外进展", "label": "海外研究"}, _node("write", "writer", "ReportSectionDraft", ["research-market", "research-overseas"]), ] plan = {**_plan(), "nodes": nodes, "edges": [ {"source": "research-market", "target": "write"}, {"source": "research-overseas", "target": "write"}, ]} decision = await RuntimeIntentRouterAgent().route( RuntimeIntentContext( text="海外市场资料太旧，重新检索 2026 年最新信息", requirement={"topic": "新能源汽车", "revision": 1}, plan_nodes=nodes, ) ) assert decision.requested_target_ids == ["research-overseas"] impact = ImpactAnalyzer().analyze( decision, plan=plan, node_runs=[], agent_runs=[], explicit_target_node_ids=decision.requested_target_ids, ) assert impact.impact_nodes == ["research-overseas", "write"] assert impact.requires_confirmation is False add = await RuntimeIntentRouterAgent().route( RuntimeIntentContext( text="增加供应链与原材料角度分析", requirement={"topic": "新能源汽车", "revision": 1}, plan_nodes=nodes, ) ) add_impact = ImpactAnalyzer().analyze( add, plan=plan, node_runs=[], agent_runs=[], explicit_target_node_ids=[], ) assert add_impact.impact_nodes == ["research-market", "write"] assert "research-overseas" not in add_impact.impact_nodes @pytest.mark.asyncio async def test_impact_analyzer_keeps_evidence_when_reanalyzing() -> None: decision = await RuntimeIntentRouterAgent().route( RuntimeIntentContext( text="这部分分析不到位，请重新分析", requirement={"topic": "新能源汽车", "revision": 1}, plan_nodes=_plan()["nodes"], target_node_ids=["analyze"], ) ) impact = ImpactAnalyzer().analyze( decision, plan=_plan(), node_runs=[], agent_runs=[], explicit_target_node_ids=["analyze"], ) assert impact.target_node_ids == ["analyze"] assert impact.impact_nodes == ["analyze", "write", "review"] assert "research" not in impact.impact_nodes assert impact.requires_confirmation is False @pytest.mark.asyncio async def test_safe_point_supersedes_inflight_attempt_and_late_output() -> None: store, _session_id, run_id = await _boot() ledger = TaskLedger(store) assignment = (await ledger.ready_assignments(run_id))[0] node_rows = await store.list_node_runs(run_id) current = next(item for item in node_rows if item["node_run_id"] == assignment.node_run_id) current["status"] = "running" await store.save_node_run(current) await store.save_agent_run( { "agent_run_id": "ar-research", "session_id": assignment.session_id, "run_id": run_id, "node_run_id": assignment.node_run_id, "phase_id": assignment.phase_id, "role": "researcher", "display_name": "研究员", "status": "running", } ) service = RuntimeCommandService(store) command = await service.submit( run_id, text="资料太旧了，请重新检索", idempotency_key="command-research", target_node_id="research", agent_run_id="ar-research", ) assert command["status"] == "accepted_waiting_boundary" assert command["impact_nodes"] == ["research", "verify", "analyze", "write", "review"] await store.save_agent_run({"agent_run_id": "ar-research", "status": "completed"}) applied = await service.process_safe_point(run_id, completed_node_id="research") assert applied == [command["command_id"]] latest = await store.list_node_runs(run_id) research = [item for item in latest if item["node_id"] == "research"] assert [item["status"] for item in research] == ["superseded", "planned"] assert research[-1]["command_overlay_revision"] == 1 assert research[-1]["agent_snapshot"]["command_overlays"][0]["command_id"] == command["command_id"] evidence = { "schema_version": 1, "claims": [{"claim_id": "late", "text": "旧结果", "kind": "fact", "source_ids": ["s1"], "excerpt": "旧"}], "sources": [{"source_id": "s1", "title": "旧来源", "excerpt": "旧"}], } await ledger.apply_result( assignment, TaskResult(kind="artifact_draft", artifact_type="EvidenceBundle", artifact_draft=evidence), ) artifacts = await store.list_artifacts_full(run_id) assert len(artifacts) == 1 assert artifacts[0]["validation_status"] == "superseded" @pytest.mark.asyncio async def test_add_angle_updates_requirement_and_requeues_only_downstream() -> None: store, session_id, run_id = await _boot() service = RuntimeCommandService(store) command = await service.submit( run_id, text="增加供应链角度分析", idempotency_key="command-angle", target_node_id="research", ) assert command["status"] == "completed" session = await store.get_session_record(session_id) assert session["requirement_revision"] == 2 assert "供应链与原材料" in session["requirement_json"]["snapshot"]["required_angles"] nodes = await store.list_node_runs(run_id) assert all(int(item.get("command_overlay_revision") or 0) == 1 for item in nodes) @pytest.mark.asyncio async def test_high_cost_replan_requires_confirmation_and_is_idempotent() -> None: store, _session_id, run_id = await _boot() service = RuntimeCommandService(store) command = await service.submit(run_id, text="重新规划，换一套方案", idempotency_key="command-replan") assert command["status"] == "confirmation_required" confirmed = await service.confirm(run_id, command["command_id"], idempotency_key="confirm-replan") assert confirmed["status"] == "completed" repeated = await service.confirm(run_id, command["command_id"], idempotency_key="confirm-replan") assert repeated["status"] == "completed" events = await store.list_events(run_id) completed = [item for item in events if item["type"] == "command.completed" and item["data"]["command_id"] == command["command_id"]] assert len(completed) == 1 @pytest.mark.asyncio async def test_command_status_compare_and_swap_rejects_second_worker() -> None: store, session_id, run_id = await _boot() command = await store.create_command(session_id, idempotency_key="cas", text="test", run_id=run_id) await store.patch_command(command["command_id"], patch={"status": "executing"}, expected_statuses={"pending"}) with pytest.raises(ReportCollaborationConflictError) as caught: await store.patch_command(command["command_id"], patch={"status": "executing"}, expected_statuses={"pending"}) assert caught.value.code == "COMMAND_STATE_CONFLICT" @pytest.mark.asyncio async def test_sql_command_patch_persists_typed_classification() -> None: engine = create_async_engine("sqlite+aiosqlite:///:memory:") async with engine.begin() as connection: await connection.run_sync(Base.metadata.create_all) store = make_report_collaboration_store(async_sessionmaker(engine, expire_on_commit=False)) try: _store, session_id, run_id = await _boot(store) command = await store.create_command(session_id, idempotency_key="sql-command", text="重新分析", run_id=run_id) patched = await store.patch_command( command["command_id"], patch={ "intent": "reanalyze", "confidence": 0.95, "target_type": "artifact_lineage", "target_ids": ["analyze"], "impact_nodes": ["analyze", "write", "review"], "status": "accepted", }, expected_statuses={"pending"}, ) assert patched["intent"] == "reanalyze" assert patched["impact_nodes"] == ["analyze", "write", "review"] assert (await store.get_command(command["command_id"])) == patched finally: await engine.dispose() @pytest.mark.asyncio async def test_rerun_without_target_does_not_expand_to_whole_graph() -> None: store, _session_id, run_id = await _boot() service = RuntimeCommandService(store) command = await service.submit(run_id, text="重新执行这个节点", idempotency_key="rerun-all") assert command["intent"] == "rerun_node" assert command["status"] == "confirmation_required" assert command["impact_nodes"] == [] confirmed = await service.confirm(run_id, command["command_id"], idempotency_key="confirm-rerun-all") assert confirmed["status"] == "failed" assert confirmed["error"] assert "可安全返工" in confirmed["error"] latest = {item["node_id"]: item for item in await store.list_node_runs(run_id)} assert all(item["status"] == "planned" and int(item["attempt"]) == 1 for item in latest.values()) @pytest.mark.asyncio async def test_requirement_patch_waits_until_rework_succeeds() -> None: store, session_id, run_id = await _boot() for node in await store.list_node_runs(run_id): node["status"] = "cancelled" await store.save_node_run(node) service = RuntimeCommandService(store) command = await service.submit( run_id, text="增加供应链角度分析", idempotency_key="angle-no-rework", target_node_id="research", ) assert command["status"] == "failed" session = await store.get_session_record(session_id) assert session["requirement_revision"] == 1 assert "供应链与原材料" not in (session["requirement_json"]["snapshot"].get("required_angles") or []) @pytest.mark.asyncio async def test_add_angle_without_research_branch_is_rejected() -> None: store = MemoryReportCollaborationStore() session = await store.create_session(owner_id="u1", title="无研究", idempotency_key="no-research") await store.save_requirement_state( session["id"], requirement_json={"snapshot": {"topic": "新能源汽车市场趋势", "required_angles": ["市场份额"], "revision": 1}}, requirement_revision=1, status="proposal_ready", ) plan = _plan() plan["nodes"] = [_node("write", "writer", "ReportSectionDraft", [])] plan["edges"] = [] inserted = await store.insert_plan(session["id"], plan) await store.select_plan(session["id"], inserted["id"], idempotency_key="select", expected_revision=1) run = await store.create_run(session["id"], plan_id=inserted["id"], idempotency_key="run", expected_revision=1) command = await RuntimeCommandService(store).submit( run["id"], text="增加供应链角度分析", idempotency_key="angle-no-branch", ) assert command["status"] == "rejected" assert "研究支路" in (command.get("error") or "") @pytest.mark.asyncio async def test_cancel_command_emits_cancelled_event() -> None: store, _session_id, run_id = await _boot() service = RuntimeCommandService(store) command = await service.submit(run_id, text="重新规划，换一套方案", idempotency_key="cancel-event") cancelled = await service.cancel(run_id, command["command_id"], idempotency_key="cancel-event-ack") assert cancelled["status"] == "cancelled" events = await store.list_events(run_id) assert any(item["type"] == "command.cancelled" and item["data"]["command_id"] == command["command_id"] for item in events) @pytest.mark.asyncio async def test_stuck_executing_command_is_reclaimed_and_applied() -> None: store, _session_id, run_id = await _boot() service = RuntimeCommandService(store) command = await store.create_command( (await store.get_run(run_id))["session_id"], idempotency_key="stuck-exec", text="资料太旧了，请重新检索", run_id=run_id, ) await store.patch_command( command["command_id"], patch={ "intent": "research_again", "impact_nodes": ["research", "verify", "analyze", "write", "review"], "normalized_instruction": "资料太旧了，请重新检索", "status": "executing", }, expected_statuses={"pending"}, ) assert await service.reclaim_executing(run_id, force=True) == [command["command_id"]] assert (await store.get_command(command["command_id"]))["status"] == "accepted" applied = await service.process_safe_point(run_id) assert applied == [command["command_id"]] assert (await store.get_command(command["command_id"]))["status"] == "completed" @pytest.mark.asyncio async def test_expired_executing_command_is_reclaimed_on_safe_point() -> None: store, _session_id, run_id = await _boot() service = RuntimeCommandService(store) command = await store.create_command( (await store.get_run(run_id))["session_id"], idempotency_key="expired-exec", text="资料太旧了，请重新检索", run_id=run_id, ) await store.patch_command( command["command_id"], patch={ "intent": "research_again", "impact_nodes": ["research", "verify", "analyze", "write", "review"], "status": "executing", }, expected_statuses={"pending"}, ) store._commands[command["command_id"]]["updated_at"] = "2020-01-01T00:00:00Z" applied = await service.process_safe_point(run_id) assert applied == [command["command_id"]] @pytest.mark.asyncio async def test_cancelled_run_wins_over_confirmed_rework() -> None: store, _session_id, run_id = await _boot() service = RuntimeCommandService(store) command = await service.submit(run_id, text="重新规划，换一套方案", idempotency_key="cancel-race-command") await store.request_cancel_run(run_id, idempotency_key="cancel-the-run") confirmed = await service.confirm(run_id, command["command_id"], idempotency_key="confirm-after-cancel") assert confirmed["status"] == "failed" assert "运行已取消" in confirmed["error"]