789 lines
35 KiB
Python
789 lines
35 KiB
Python
from __future__ import annotations
|
|
|
|
from app.core.agent.intent_router import IntentRouterV2
|
|
from app.core.agent.runtime.docs_qa_pipeline import DocsQAPipelineRunner, DocsTaskPlanner
|
|
from tests.docs_qa_eval.fixture_adapter import InMemoryDocsRetrievalAdapter
|
|
from tests.unit_tests.rag.intent_router_testkit import repo_context
|
|
|
|
|
|
def test_docs_pipeline_uses_synthesizer_for_explain() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/components/billing.md",
|
|
"title": "Billing facts",
|
|
"content": "Billing обновляет статус заказа после оплаты.",
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни модуль billing в документации", "docs-session")
|
|
|
|
assert result.router_result.intent == "DOCUMENTATION_EXPLAIN"
|
|
assert "Billing" in result.answer
|
|
assert result.openapi_result is None
|
|
assert result.diagnostics.facts_found == 1
|
|
|
|
|
|
def test_docs_pipeline_uses_openapi_branch() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/orders-create.md",
|
|
"title": "POST /orders",
|
|
"content": "Create order",
|
|
"metadata": {
|
|
"endpoint": "/orders",
|
|
"http_method": "post",
|
|
"request_schema": {"type": "object", "properties": {"customer_id": {}}},
|
|
"response_schema": {"type": "object", "properties": {"order_id": {}}},
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй openapi spec для создания заказа", "docs-session")
|
|
|
|
assert result.router_result.intent == "OPENAPI_GENERATION"
|
|
assert result.openapi_result is not None
|
|
assert result.openapi_result.path == "/orders"
|
|
assert result.openapi_result.method == "post"
|
|
assert result.answer_mode != "degraded"
|
|
assert result.diagnostics.gate_decision != "reject"
|
|
assert "paths:" in result.answer
|
|
assert "/orders" in result.answer
|
|
|
|
|
|
def test_docs_pipeline_keeps_query_candidates_for_not_found_entity() -> None:
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter([]), repo_context=repo_context())
|
|
|
|
result = runner.run("Что такое сущность PaymentTransaction?", "docs-session")
|
|
|
|
assert result.answer_mode == "degraded"
|
|
assert "PaymentTransaction" in result.diagnostics.query_entity_candidates
|
|
assert result.diagnostics.resolved_entity_candidates == []
|
|
assert result.diagnostics.degraded_reason == "not_found_exact_anchor"
|
|
|
|
|
|
def test_docs_pipeline_reports_canonical_doc_ids_and_layers() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/api/health.md",
|
|
"title": "Health relations",
|
|
"content": "Health -> runtime health",
|
|
"metadata": {"doc_id": "api.health_endpoint", "target_doc_id": "domain.runtime_health", "endpoint": "/health"},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Runtime health",
|
|
"content": "Runtime health overview",
|
|
"metadata": {"doc_id": "domain.runtime_health"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какие документы связаны с endpoint /health?", "docs-session")
|
|
|
|
assert result.diagnostics.doc_ids[:2] == ["api.health_endpoint", "domain.runtime_health"]
|
|
assert result.diagnostics.planned_layers == ["D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.executed_layers == ["D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.non_empty_layers == ["D5_RELATION_GRAPH"]
|
|
assert result.diagnostics.relation_hits_count == 1
|
|
assert result.diagnostics.relation_targets == ["domain.runtime_health"]
|
|
assert result.diagnostics.answer_mode == "answered"
|
|
assert result.diagnostics.layer_diagnostics["D5_RELATION_GRAPH"]["hits"] == 1
|
|
assert result.diagnostics.gate_decision == "allow"
|
|
|
|
|
|
def test_related_docs_uses_relation_graph_as_primary_layer() -> None:
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter([]), repo_context=repo_context())
|
|
|
|
result = runner.run("Что еще посмотреть по теме billing?", "docs-session")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "RELATED_DOCS_EXPLAIN"
|
|
assert result.diagnostics.planned_layers[0] == "D5_RELATION_GRAPH"
|
|
assert result.diagnostics.executed_layers[0] == "D5_RELATION_GRAPH"
|
|
|
|
|
|
def test_openapi_partial_contract_returns_partial_mode() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/orders-create.md",
|
|
"title": "POST /orders",
|
|
"content": "Create order",
|
|
"metadata": {
|
|
"endpoint": "/orders",
|
|
"http_method": "post",
|
|
"request_schema": {"type": "object", "properties": {"customer_id": {}}},
|
|
"doc_id": "api.orders_create",
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй openapi для /orders", "docs-session")
|
|
|
|
assert result.answer_mode == "structured_spec_partial"
|
|
assert result.degraded_reason == "answered_with_gaps"
|
|
assert result.openapi_result is not None
|
|
assert result.openapi_result.path == "/orders"
|
|
assert result.diagnostics.gate_decision == "partial"
|
|
assert "paths:" in result.answer
|
|
assert "/orders" in result.answer
|
|
|
|
|
|
def test_docs_pipeline_accepts_precomputed_task_plan_without_rerouting() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "/health returns runtime and component statuses.",
|
|
"metadata": {"document_id": "api.health_endpoint", "type": "api_method", "endpoint": "/health"},
|
|
},
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/health.md",
|
|
"title": "api.health_endpoint:response",
|
|
"content": "Returns health summary and component diagnostics.",
|
|
"metadata": {"subject_id": "api.health_endpoint", "type": "api_method"},
|
|
},
|
|
]
|
|
route_result = IntentRouterV2().route(
|
|
"Объясни API метод /health",
|
|
repo_context=repo_context(),
|
|
)
|
|
task_plan = DocsTaskPlanner().plan(
|
|
"Объясни API метод /health",
|
|
"docs-session",
|
|
route_result=route_result,
|
|
)
|
|
|
|
class FailingRouter:
|
|
def route(self, *_args, **_kwargs):
|
|
raise AssertionError("runner should use the precomputed task plan")
|
|
|
|
runner = DocsQAPipelineRunner(FailingRouter(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run(
|
|
"Объясни API метод /health",
|
|
"docs-session",
|
|
mode="pre_llm_only",
|
|
task_plan=task_plan,
|
|
)
|
|
|
|
assert result.router_result.intent == "DOCUMENTATION_EXPLAIN"
|
|
assert result.router_result.query_plan.sub_intent == "API_METHOD_EXPLAIN"
|
|
assert result.diagnostics.selected_primary_documents == ["api.health_endpoint"]
|
|
assert result.diagnostics.gate_decision == "allow_exact"
|
|
|
|
|
|
def test_pre_llm_mode_returns_diagnostic_only_without_answer_generation() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "RuntimeManager",
|
|
"content": "Coordinates runtime orchestration.",
|
|
"metadata": {"doc_id": "architecture.runtime_manager", "component": "RuntimeManager"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какую роль в системе играет RuntimeManager?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.mode == "pre_llm_only"
|
|
assert result.answer == ""
|
|
assert result.diagnostics.gate_decision == "allow"
|
|
assert "RuntimeManager" in result.diagnostics.query_entity_candidates
|
|
assert "RuntimeManager" in result.diagnostics.resolved_entity_candidates
|
|
|
|
|
|
def test_pre_llm_mode_detects_path_anchor_candidates() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "Health endpoint",
|
|
"metadata": {"doc_id": "api.health_endpoint", "endpoint": "/health", "http_method": "get"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни API метод /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert "/health" in result.diagnostics.query_anchor_candidates
|
|
assert "/health" in result.diagnostics.resolved_anchor_candidates
|
|
assert result.diagnostics.planned_layers == ["D1_DOCUMENT_CATALOG", "D2_FACT_INDEX", "D0_DOC_CHUNKS", "D4_WORKFLOW_INDEX"]
|
|
assert set(result.diagnostics.executed_layers) == {"D1_DOCUMENT_CATALOG", "D2_FACT_INDEX", "D4_WORKFLOW_INDEX", "D0_DOC_CHUNKS"}
|
|
|
|
|
|
def test_pre_llm_openapi_gate_reports_missing_signals() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/orders.md",
|
|
"title": "POST /orders",
|
|
"content": "Create order",
|
|
"metadata": {"doc_id": "api.orders_create", "endpoint": "/orders", "http_method": "post"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй openapi для /orders", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.gate_decision == "partial"
|
|
assert result.diagnostics.gate_decision_reason == "answered_with_gaps"
|
|
assert result.answer_mode == "ready_partial"
|
|
assert result.answer
|
|
assert "paths:" in result.answer
|
|
assert "path_found" in result.diagnostics.gate_satisfied_requirements
|
|
assert "http_method_found" in result.diagnostics.gate_satisfied_requirements
|
|
assert "contract_fields_found" in result.diagnostics.gate_missing_requirements
|
|
assert result.diagnostics.evidence_summary["openapi_signals"]["path_found"] is True
|
|
assert result.diagnostics.openapi_evidence["path_found"] is True
|
|
assert result.diagnostics.layer_diagnostics["D2_FACT_INDEX"]["hits"] == 1
|
|
|
|
|
|
def test_related_docs_keeps_relation_hits_when_exact_anchor_is_missing() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/relations/health-links.md",
|
|
"title": "Health relations",
|
|
"content": "Runtime health and service overview",
|
|
"metadata": {
|
|
"document_id": "api.health_endpoint",
|
|
"target_document_id": "domain.runtime_health",
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какие документы связаны с endpoint /health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "RELATED_DOCS_EXPLAIN"
|
|
assert result.diagnostics.doc_ids[:2] == ["api.health_endpoint", "domain.runtime_health"]
|
|
assert result.diagnostics.relation_hits_count == 1
|
|
assert result.diagnostics.exact_anchor_match is False
|
|
assert result.diagnostics.gate_decision == "allow"
|
|
assert result.diagnostics.gate_decision_reason == "relation_evidence_available"
|
|
|
|
|
|
def test_related_docs_exact_path_outranks_neighboring_api_doc() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/documentation/api/control-actions-endpoint.md",
|
|
"title": "HTTP API /actions/{action}",
|
|
"content": "Nearby API relation",
|
|
"metadata": {
|
|
"document_id": "api.control_actions_endpoint",
|
|
"target_document_id": "architecture.telegram_notify_app",
|
|
},
|
|
},
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/documentation/api/health-endpoint.md",
|
|
"title": "HTTP API /health",
|
|
"content": "Health relation",
|
|
"metadata": {
|
|
"document_id": "api.health_endpoint",
|
|
"target_document_id": "domain.runtime_health",
|
|
"endpoint": "/health",
|
|
},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какие документы связаны с endpoint /health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.anchor_candidates[0] == "api.health_endpoint"
|
|
assert result.diagnostics.selected_anchor == "api.health_endpoint"
|
|
assert result.diagnostics.anchor_match_type == "exact_path"
|
|
assert result.diagnostics.anchor_selection_reason == "metadata.endpoint exact match"
|
|
assert result.diagnostics.doc_ids[0] == "api.health_endpoint"
|
|
|
|
|
|
def test_pre_llm_targeted_chunk_fallback_enriches_api_method_layers() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "Health endpoint",
|
|
"metadata": {"doc_id": "api.health_endpoint"},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/api/health.md",
|
|
"title": "api.health_endpoint:Scenario",
|
|
"content": "request -> check runtime -> return status",
|
|
"metadata": {"doc_id": "api.health_endpoint"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни API метод /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert "D0_DOC_CHUNKS" in result.diagnostics.executed_layers
|
|
assert "D0_DOC_CHUNKS" in result.diagnostics.non_empty_layers
|
|
assert result.diagnostics.layer_diagnostics["D0_DOC_CHUNKS"]["hits"] == 1
|
|
assert result.diagnostics.selected_chunk_ids == ["api.health_endpoint"]
|
|
|
|
|
|
def test_openapi_request_fragment_uses_fragment_aware_gate() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/send.md",
|
|
"title": "POST /send",
|
|
"content": "Request payload contains message and chat_id fields.",
|
|
"metadata": {
|
|
"document_id": "api.send_message_endpoint",
|
|
"endpoint": "/send",
|
|
"http_method": "post",
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй request schema для endpoint /send", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "OPENAPI_FRAGMENT_GENERATE"
|
|
assert result.diagnostics.requested_fragment_type == "request_schema"
|
|
assert "path" in result.diagnostics.fragment_evidence_found
|
|
assert "payload_description" in result.diagnostics.fragment_evidence_found
|
|
assert result.diagnostics.gate_decision in {"allow", "partial"}
|
|
assert result.diagnostics.gate_decision_reason in {"evidence_sufficient", "fragment_payload_only"}
|
|
assert result.diagnostics.fragment_missing_requirements == []
|
|
assert result.answer_mode in {"ready", "ready_partial"}
|
|
assert result.answer
|
|
assert "type: object" in result.answer
|
|
|
|
|
|
def test_api_method_explain_prefers_api_method_primary_doc() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Сущность runtime health",
|
|
"content": "Runtime health describes overall service health.",
|
|
"metadata": {"document_id": "domain.runtime_health", "type": "domain_entity"},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "/health returns runtime and component statuses.",
|
|
"metadata": {"document_id": "api.health_endpoint", "type": "api_method", "endpoint": "/health"},
|
|
},
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Runtime health links",
|
|
"content": "runtime health used by health endpoint",
|
|
"metadata": {"document_id": "domain.runtime_health", "target_doc_id": "api.health_endpoint"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Что делает метод health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "API_METHOD_EXPLAIN"
|
|
assert result.diagnostics.target_anchor in {"health", "/health"}
|
|
assert result.diagnostics.api_method_match_found is True
|
|
assert result.diagnostics.selected_primary_documents == ["api.health_endpoint"]
|
|
assert "api.health_endpoint" in result.diagnostics.primary_doc_candidates
|
|
assert result.diagnostics.evidence_gate_require_target_api_spec is True
|
|
assert result.diagnostics.evidence_gate_target_api_spec_found is True
|
|
assert result.answer_mode == "exact"
|
|
assert result.diagnostics.gate_decision == "allow_exact"
|
|
|
|
|
|
def test_api_method_explain_promotes_api_doc_via_links() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Сущность runtime health",
|
|
"content": "Runtime health is the domain model for observability.",
|
|
"metadata": {
|
|
"document_id": "domain.runtime_health",
|
|
"type": "domain_entity",
|
|
"links": [{"target": "api.health_endpoint", "type": "used_by"}],
|
|
},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/api/health.md",
|
|
"title": "api.health_endpoint:Overview",
|
|
"content": "Endpoint /health returns overall runtime status and component diagnostics.",
|
|
"metadata": {"document_id": "api.health_endpoint", "type": "api_method", "endpoint": "/health"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Как работает health endpoint?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.promoted_via_links == ["api.health_endpoint"]
|
|
assert result.diagnostics.selected_primary_documents == ["api.health_endpoint"]
|
|
assert result.diagnostics.api_method_match_found is True
|
|
|
|
|
|
def test_api_method_explain_rejects_cross_endpoint_primary_candidates() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/documentation/api/control-actions-endpoint.md",
|
|
"title": "HTTP API /actions/{action}",
|
|
"content": "Endpoint for controlling actions.",
|
|
"metadata": {
|
|
"document_id": "api.control_actions_endpoint",
|
|
"type": "api_method",
|
|
"endpoint": "/actions/{action}",
|
|
},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/documentation/api/health-endpoint.md",
|
|
"title": "HTTP API /health",
|
|
"content": "Health endpoint returns runtime health and component diagnostics.",
|
|
"metadata": {
|
|
"document_id": "api.health_endpoint",
|
|
"type": "api_method",
|
|
"endpoint": "/health",
|
|
},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/documentation/api/actions-endpoint.md",
|
|
"title": "api.control_actions_endpoint:Scenario",
|
|
"content": "The /actions/{action} endpoint triggers runtime actions.",
|
|
"metadata": {
|
|
"document_id": "api.control_actions_endpoint",
|
|
"type": "api_method",
|
|
"endpoint": "/actions/{action}",
|
|
},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Как работает метод health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "API_METHOD_EXPLAIN"
|
|
assert result.diagnostics.target_endpoint_identity["normalized_doc_id"] == "api.health_endpoint"
|
|
assert result.diagnostics.selected_primary_documents == ["api.health_endpoint"]
|
|
assert result.diagnostics.primary_api_documents_after_filter == ["api.health_endpoint"]
|
|
assert "api.control_actions_endpoint" in result.diagnostics.rejected_endpoint_candidates
|
|
assert result.diagnostics.cross_endpoint_leakage_detected is True
|
|
assert result.diagnostics.evidence_gate_target_api_spec_found is True
|
|
assert "api.control_actions_endpoint" not in result.diagnostics.selected_doc_ids
|
|
|
|
|
|
def test_api_method_explain_without_exact_target_returns_insufficiency() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/documentation/api/control-actions-endpoint.md",
|
|
"title": "HTTP API /actions/{action}",
|
|
"content": "Endpoint for controlling actions.",
|
|
"metadata": {
|
|
"document_id": "api.control_actions_endpoint",
|
|
"type": "api_method",
|
|
"endpoint": "/actions/{action}",
|
|
},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/documentation/api/send-endpoint.md",
|
|
"title": "HTTP API /send",
|
|
"content": "Endpoint for sending messages.",
|
|
"metadata": {
|
|
"document_id": "api.send_message_endpoint",
|
|
"type": "api_method",
|
|
"endpoint": "/send",
|
|
},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Что делает метод health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "API_METHOD_EXPLAIN"
|
|
assert result.diagnostics.target_endpoint_identity["normalized_doc_id"] == "api.health_endpoint"
|
|
assert result.diagnostics.selected_primary_documents == []
|
|
assert "api.control_actions_endpoint" in result.diagnostics.rejected_endpoint_candidates
|
|
assert "api.send_message_endpoint" in result.diagnostics.rejected_endpoint_candidates
|
|
assert result.diagnostics.target_api_spec_found_exact is False
|
|
assert result.diagnostics.evidence_gate_target_api_spec_found is False
|
|
assert result.diagnostics.gate_decision == "reject"
|
|
assert result.answer_mode == "insufficient"
|
|
assert "api.control_actions_endpoint" not in result.diagnostics.selected_doc_ids
|
|
assert "api.send_message_endpoint" not in result.diagnostics.selected_doc_ids
|
|
|
|
|
|
def test_api_method_explain_uses_indirect_mode_from_target_linked_docs() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Runtime health",
|
|
"content": "Runtime health describes overall service state and component diagnostics.",
|
|
"metadata": {
|
|
"document_id": "domain.runtime_health",
|
|
"type": "domain_entity",
|
|
"links": [{"target": "api.health_endpoint", "type": "used_by"}],
|
|
},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Как работает метод health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.answer_mode == "indirect"
|
|
assert result.diagnostics.gate_decision == "allow_indirect"
|
|
assert result.diagnostics.raw_retrieval_non_empty is True
|
|
assert result.diagnostics.target_primary_context_non_empty is False
|
|
assert result.diagnostics.indirect_target_context_non_empty is True
|
|
assert result.diagnostics.graph_promotion_attempted is True
|
|
assert result.diagnostics.graph_promotion_hits == ["api.health_endpoint"]
|
|
assert result.diagnostics.promoted_target_loaded is False
|
|
assert result.diagnostics.materialization_failure_reason == "materialized_rows_empty"
|
|
assert result.diagnostics.final_primary_document_ids == []
|
|
assert "domain.runtime_health" in result.diagnostics.final_secondary_document_ids
|
|
|
|
|
|
def test_api_method_explain_skips_llm_when_no_exact_or_indirect_context() -> None:
|
|
from tests.unit_tests.rag.test_docs_prompt_layer import FakeLlm
|
|
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/documentation/api/send-endpoint.md",
|
|
"title": "HTTP API /send",
|
|
"content": "Endpoint for sending messages.",
|
|
"metadata": {
|
|
"document_id": "api.send_message_endpoint",
|
|
"type": "api_method",
|
|
"endpoint": "/send",
|
|
},
|
|
}
|
|
]
|
|
llm = FakeLlm("should not be called")
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context(), llm=llm)
|
|
|
|
result = runner.run("Что делает метод health?", "docs-session")
|
|
|
|
assert llm.calls == []
|
|
assert result.answer_mode == "insufficient"
|
|
assert result.diagnostics.llm_called is False
|
|
assert result.diagnostics.llm_call_reason == "no_exact_or_indirect_target_context"
|
|
assert result.diagnostics.gate_decision == "reject"
|
|
|
|
|
|
def test_api_method_explain_materializes_promoted_target_into_primary_context() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Runtime health",
|
|
"content": "Runtime health describes service state and component diagnostics.",
|
|
"metadata": {
|
|
"document_id": "domain.runtime_health",
|
|
"type": "domain_entity",
|
|
"links": [{"target": "api.health_endpoint", "type": "used_by"}],
|
|
},
|
|
}
|
|
]
|
|
materialized_rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "/health returns runtime and component statuses.",
|
|
"metadata": {"document_id": "api.health_endpoint", "type": "api_method", "endpoint": "/health"},
|
|
},
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/health.md",
|
|
"title": "api.health_endpoint:response",
|
|
"content": "Returns health summary and component diagnostics.",
|
|
"metadata": {"subject_id": "api.health_endpoint", "type": "api_method"},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/api/health.md",
|
|
"title": "api.health_endpoint:Overview",
|
|
"content": "Endpoint /health returns overall runtime health.",
|
|
"metadata": {"document_id": "api.health_endpoint", "type": "api_method", "endpoint": "/health"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(
|
|
IntentRouterV2(),
|
|
InMemoryDocsRetrievalAdapter(rows, materialized_rows=materialized_rows),
|
|
repo_context=repo_context(),
|
|
)
|
|
|
|
result = runner.run("Как работает метод health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.answer_mode == "exact"
|
|
assert result.diagnostics.graph_promotion_hits == ["api.health_endpoint"]
|
|
assert result.diagnostics.graph_promotion_materialized == ["api.health_endpoint"]
|
|
assert result.diagnostics.promoted_target_loaded is True
|
|
assert result.diagnostics.promoted_target_chunks_loaded == 1
|
|
assert result.diagnostics.promoted_target_facts_loaded == 1
|
|
assert result.diagnostics.pinned_document_ids == ["api.health_endpoint"]
|
|
assert result.diagnostics.final_primary_document_ids == ["api.health_endpoint"]
|
|
assert "domain.runtime_health" in result.diagnostics.final_secondary_document_ids
|
|
assert result.diagnostics.materialized_target_primary_context_non_empty is True
|
|
assert result.diagnostics.gate_decision == "allow_exact"
|
|
|
|
|
|
def test_entity_question_does_not_prefer_api_method_primary_doc() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/domain/runtime-health.md",
|
|
"title": "Сущность runtime health",
|
|
"content": "Runtime health describes service state.",
|
|
"metadata": {"document_id": "domain.runtime_health", "type": "domain_entity"},
|
|
},
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "/health returns runtime status.",
|
|
"metadata": {"document_id": "api.health_endpoint", "type": "api_method", "endpoint": "/health"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Что такое runtime health?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.router_result.query_plan.sub_intent == "ENTITY_EXPLAIN"
|
|
assert result.diagnostics.selected_primary_documents == []
|
|
assert result.diagnostics.api_method_match_found is False
|
|
assert result.answer == ""
|
|
|
|
|
|
def test_openapi_method_with_only_path_is_rejected() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/api/health.md",
|
|
"title": "/health",
|
|
"content": "",
|
|
"metadata": {"endpoint": "/health", "doc_id": "api.health_endpoint"},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй OpenAPI для endpoint /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.gate_decision == "reject"
|
|
assert result.answer_mode == "degraded"
|
|
|
|
|
|
def test_openapi_response_fragment_does_not_require_request_payload() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/api/health.md",
|
|
"title": "GET /health",
|
|
"content": "Returns 200 with status and uptime fields.",
|
|
"metadata": {
|
|
"endpoint": "/health",
|
|
"http_method": "get",
|
|
"response_schema": {"type": "object", "properties": {"status": {}, "uptime": {}}},
|
|
},
|
|
}
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Сгенерируй response schema для endpoint /health", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.requested_fragment_type == "response_schema"
|
|
assert result.diagnostics.gate_decision != "reject"
|
|
assert result.answer_mode != "degraded"
|
|
assert "properties:" in result.answer
|
|
|
|
|
|
def test_system_flow_evidence_prefers_workflows_and_relations() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "Checkout overview",
|
|
"content": "Overview doc",
|
|
"metadata": {"doc_id": "workflow.checkout"},
|
|
},
|
|
{
|
|
"layer": "D4_WORKFLOW_INDEX",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "Checkout steps",
|
|
"content": "cart -> payment -> confirm",
|
|
"metadata": {"workflow_id": "workflow.checkout"},
|
|
},
|
|
{
|
|
"layer": "D5_RELATION_GRAPH",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "Checkout relations",
|
|
"content": "Related to payment entity",
|
|
"metadata": {"doc_id": "workflow.checkout", "target_doc_id": "domain.payment"},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/workflows/checkout.md",
|
|
"title": "workflow.checkout:chunk",
|
|
"content": "payment step persists order",
|
|
"metadata": {"chunk_id": "chunk.checkout.1", "doc_id": "workflow.checkout"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Объясни как работает checkout workflow", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.planned_layers == ["D4_WORKFLOW_INDEX", "D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.selected_workflow_ids == ["workflow.checkout"]
|
|
assert result.diagnostics.selected_relation_ids == ["domain.payment"]
|
|
assert result.diagnostics.selected_chunk_ids == ["chunk.checkout.1"]
|
|
|
|
|
|
def test_component_evidence_prefers_facts_and_targeted_chunks() -> None:
|
|
rows = [
|
|
{
|
|
"layer": "D1_DOCUMENT_CATALOG",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "Runtime manager",
|
|
"content": "Overview",
|
|
"metadata": {"doc_id": "architecture.runtime_manager"},
|
|
},
|
|
{
|
|
"layer": "D2_FACT_INDEX",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "Runtime facts",
|
|
"content": "Coordinates runtime orchestration",
|
|
"metadata": {"fact_id": "fact.runtime_manager.role", "doc_id": "architecture.runtime_manager"},
|
|
},
|
|
{
|
|
"layer": "D0_DOC_CHUNKS",
|
|
"path": "docs/components/runtime-manager.md",
|
|
"title": "RuntimeManager chunk",
|
|
"content": "Starts worker loops and supervises control channels",
|
|
"metadata": {"chunk_id": "chunk.runtime_manager.1", "doc_id": "architecture.runtime_manager"},
|
|
},
|
|
]
|
|
runner = DocsQAPipelineRunner(IntentRouterV2(), InMemoryDocsRetrievalAdapter(rows), repo_context=repo_context())
|
|
|
|
result = runner.run("Какую роль в системе играет RuntimeManager?", "docs-session", mode="pre_llm_only")
|
|
|
|
assert result.diagnostics.planned_layers == ["D2_FACT_INDEX", "D5_RELATION_GRAPH", "D1_DOCUMENT_CATALOG", "D0_DOC_CHUNKS"]
|
|
assert result.diagnostics.selected_fact_ids == ["fact.runtime_manager.role"]
|
|
assert result.diagnostics.selected_doc_ids == ["architecture.runtime_manager"]
|
|
assert result.diagnostics.selected_chunk_ids == ["chunk.runtime_manager.1"]
|