Resolve Issue #117: Fix evidence tree orphan nesting, outline leakage, and enforce English agent reasoning

apps/api/app/api/v1/routes/retrieval.py

@@ -13,6 +13,7 @@
 from shared.core.database import get_db
 from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace
 from shared.services.retrieval.app_service import run_retrieval_query
+from shared.services.retrieval.settings import DEFAULT_TOP_K
 
 router = APIRouter(tags=["Retrieval"])
 
@@ -29,7 +30,7 @@ class RetrievalQueryRequest(BaseModel):
         description="Effective namespace; defaults to default",
     )
     query: str
-    top_k: int = 10
+    top_k: int = DEFAULT_TOP_K
     exclude_document_ids: list[str] = Field(default_factory=list)
     exclude_sections: list[ExcludeSection] = Field(default_factory=list)
     data_type: int = Field(

apps/api/app/mcp/retrieval_server.py

@@ -13,6 +13,7 @@
 from shared.core.database import get_db_context
 from shared.models.schemas.retrieval_namespace import normalize_retrieval_namespace
 from shared.services.retrieval.app_service import run_retrieval_query
+from shared.services.retrieval.settings import DEFAULT_TOP_K
 
 DbFactory = Callable[[], AsyncContextManager[AsyncSession]]
 KNOWHERE_NAMESPACE_HEADER = "x-knowhere-namespace"
@@ -51,36 +52,20 @@ def resolve_mcp_namespace(*, ctx: Context | None) -> str:
 
 
 def to_mcp_query_response(response: dict[str, Any]) -> dict[str, Any]:
-    results: list[dict[str, Any]] = []
-    for row in response.get("results", []):
-        if not isinstance(row, dict):
-            continue
-
-        source_value = row.get("source")
-        source = source_value if isinstance(source_value, dict) else row
-        result: dict[str, Any] = {
-            "content": row.get("content"),
-            "source_file_name": source.get("source_file_name"),
-            "section_path": source.get("section_path"),
-            "chunk_type": row.get("chunk_type"),
-        }
-        if row.get("asset_url"):
-            result["asset_url"] = row["asset_url"]
-        results.append(result)
-
-    mcp_response: dict[str, Any] = {
+    """Project the internal retrieval response to the MCP agent contract.
+
+    MCP returns exactly 3 PRIMARY fields:
+    - evidence_text: hierarchical evidence tree for LLM consumption
+    - referenced_chunks: structured chunk references for citation / follow-up
+    - decision_trace: navigation decisions including terminal stop/failure
+    """
+    return {
         "query": response.get("query"),
-        "results": results,
         "evidence_text": response.get("evidence_text") or "",
+        "referenced_chunks": response.get("referenced_chunks") or [],
+        "decision_trace": response.get("decision_trace") or [],
     }
 
-    if response.get("stop_reason") is not None:
-        mcp_response["stop_reason"] = response["stop_reason"]
-    if response.get("decision_trace") is not None:
-        mcp_response["decision_trace"] = response["decision_trace"]
-
-    return mcp_response
-
 
 async def resolve_mcp_user_id(*, ctx: Context | None, db: AsyncSession) -> str:
     request = get_mcp_request(ctx)
@@ -101,8 +86,10 @@ def create_retrieval_mcp_server(
         "knowhere-retrieval",
         instructions=(
             "Use this server to search published documents. "
-            "It returns evidence_text and ranked snippets, but never final answers. "
-            "Downstream agents should synthesize from the returned evidence."
+            "It returns evidence_text (hierarchical evidence tree), "
+            "referenced_chunks (structured chunk citations), and "
+            "decision_trace (navigation decisions). "
+            "Downstream agents should synthesize answers from evidence_text."
         ),
         streamable_http_path=streamable_http_path,
         stateless_http=True,
@@ -112,17 +99,53 @@ def create_retrieval_mcp_server(
     @server.tool(
         name="retrieval.query",
         description=(
-            "Search published documents and return relevant snippets plus unified "
-            "evidence_text for downstream answer synthesis."
+            "Search published documents. Returns evidence_text (hierarchical "
+            "evidence for LLM consumption), referenced_chunks (cited chunk "
+            "metadata for follow-up queries), and decision_trace (navigation "
+            "decisions including stop/failure reasons). "
+            "Include navigation intent directly in your query text — the "
+            "engine will automatically locate the right documents and sections."
         ),
     )
     async def query_documents(
-        query: Annotated[str, Field(description="What you want to search for.")],
+        query: Annotated[
+            str,
+            Field(description=(
+                "What you want to search for. You may include navigation "
+                "hints naturally, e.g. 'find tables in chapter 3 of the "
+                "safety report'. The engine understands document structure."
+            )),
+        ],
         top_k: Annotated[
-            int, Field(description="Maximum number of results to return.")
-        ] = 5,
+            int,
+            Field(description=(
+                "Number of candidate chunks for initial discovery. "
+                "The final output is budget-controlled; this only affects "
+                "the discovery recall pool. Usually no need to adjust."
+            )),
+        ] = DEFAULT_TOP_K,
+        exclude_document_ids: Annotated[
+            list[str],
+            Field(description=(
+                "Document IDs to exclude from this query. "
+                "Use document_id values from prior referenced_chunks."
+            )),
+        ] = [],
+        exclude_sections: Annotated[
+            list[dict[str, str]],
+            Field(description=(
+                "Sections to exclude. Each item: "
+                '{"document_id": "...", "section_path": "..."}.'
+            )),
+        ] = [],
         ctx: Context | None = None,
     ) -> dict:
+        # TODO(intent-step): When the Intent Understanding step is
+        # implemented, it will parse `query` here to extract structured
+        # hints (document_hint, scope_hint, content_type_hint) and set
+        # data_type / signal_paths / filter_mode / exclude_document_ids
+        # automatically before calling run_retrieval_query.
+        # See: shared/services/retrieval/intent/ (to be created)
         namespace = resolve_mcp_namespace(ctx=ctx)
         async with db_factory() as db:
             user_id = await resolve_mcp_user_id(ctx=ctx, db=db)
@@ -132,8 +155,9 @@ async def query_documents(
                 namespace=namespace,
                 query=query,
                 top_k=top_k,
-                exclude_document_ids=[],
-                exclude_sections=[],
+                exclude_document_ids=exclude_document_ids,
+                exclude_sections=[item for item in exclude_sections],
+                use_agentic=True,
             )
             return to_mcp_query_response(response)
 

apps/api/tests/contract/test_agentic_discovery_selection_contract.py

@@ -6,7 +6,7 @@
 
 
 def test_root_discovery_hint_is_projected_for_llm_selection() -> None:
-    hint_lines, hint_by_path = _project_discovery_hints(
+    hint_lines, hint_by_path, excluded_hints = _project_discovery_hints(
         [
             {
                 "section_path": "Root",

packages/shared-python/shared/services/retrieval/agentic/core/trace.py

@@ -16,6 +16,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from shared.services.retrieval.agentic.core.types import AgentRunConfig, ToolResult
+from shared.services.retrieval.settings import DEFAULT_TOP_K
 
 
 def _now_utc() -> datetime:
@@ -47,7 +48,7 @@ def __init__(
         namespace: str,
         query: str,
         config: AgentRunConfig,
-        top_k: int = 10,
+        top_k: int = DEFAULT_TOP_K,
         data_type: int = 1,
         filters: dict[str, Any] | None = None,
         parent_run_id: str | None = None,

packages/shared-python/shared/services/retrieval/agentic/discovery/phase.py

@@ -13,6 +13,7 @@
 from shared.services.retrieval.agentic.core.trace import TraceRecorder
 from shared.services.retrieval.agentic.core.types import AgentState, CandidateDoc, ToolResult
 from shared.services.retrieval.llm_adapter import LLMFn
+from shared.services.retrieval.search.lexical_text import normalize_section_path
 
 
 async def run_initial_discovery(
@@ -76,6 +77,9 @@ async def run_initial_discovery(
         f"status={discovery_result.status} latency={discovery_result.latency_ms}ms"
     )
 
+    # Build per-document discovery signals for KG soft-prompting
+    discovery_signals = build_discovery_signals(discovery_rows)
+
     if bootstrap_llm_fn is not None:
         await _select_documents(
             db,
@@ -87,42 +91,39 @@ async def run_initial_discovery(
             query=query,
             exclude_document_ids=exclude_document_ids,
             bootstrap_llm_fn=bootstrap_llm_fn,
+            discovery_signals=discovery_signals,
         )
 
     return discovery_rows
 
 
-async def register_discovery_documents(
-    db: AsyncSession,
-    *,
-    state: AgentState,
-    discovery_by_doc: dict[str, list[dict[str, Any]]],
-) -> None:
-    selected_doc_ids = {doc.document_id for doc in state.selected_docs}
-    for doc_id in discovery_by_doc:
-        if doc_id in selected_doc_ids or doc_id in state.ever_explored_doc_ids:
-            continue
-        doc_stmt = (
-            select(Document.document_id, Document.source_file_name, Document.current_job_result_id)
-            .where(Document.document_id == doc_id)
+def build_discovery_signals(
+    discovery_rows: list[dict[str, Any]],
+) -> dict[str, list[str]]:
+    """Build per-document discovery signals from bottom discovery results.
+
+    Returns a mapping of ``{doc_id: [path1, path2, ...]}`` for documents
+    where keyword/semantic search found potentially relevant section paths.
+    These signals are injected as soft hints into the KG document selection
+    prompt, allowing the LLM to make an informed decision rather than
+    force-injecting documents.
+    """
+    signals: dict[str, list[str]] = {}
+    seen: dict[str, set[str]] = {}
+    for row in discovery_rows:
+        doc_id = row.get("document_id", "")
+        section_path = normalize_section_path(
+            str(row.get("section_path", "") or "").strip()
         )
-        doc_result = await db.execute(doc_stmt)
-        row_data = doc_result.first()
-        if row_data is None:
+        if not doc_id or not section_path or section_path == "Root":
             continue
-        did, fname, job_result_id = row_data
-        state.selected_docs.append(
-            CandidateDoc(
-                document_id=did,
-                source_file_name=fname or did,
-                confidence=0.4,
-                reason="discovery_auto (not in KG selection)",
-                source="discovery_auto",
-            )
-        )
-        state.doc_id_to_name[did] = fname or did
-        if job_result_id:
-            state.doc_job_map[did] = job_result_id
+        if doc_id not in seen:
+            seen[doc_id] = set()
+            signals[doc_id] = []
+        if section_path not in seen[doc_id]:
+            seen[doc_id].add(section_path)
+            signals[doc_id].append(section_path)
+    return signals
 
 
 async def _select_documents(
@@ -136,6 +137,7 @@ async def _select_documents(
     query: str,
     exclude_document_ids: list[str],
     bootstrap_llm_fn: LLMFn,
+    discovery_signals: dict[str, list[str]] | None = None,
 ) -> None:
     try:
         kg_result = await tools.kg_document_select(
@@ -146,6 +148,7 @@ async def _select_documents(
             llm_fn=bootstrap_llm_fn,
             exclude_document_ids=list(state.ever_explored_doc_ids | set(exclude_document_ids)),
             budget_snapshot=state.ledger.snapshot() if state.ledger else None,
+            discovery_signals=discovery_signals,
         )
     except BudgetExceeded:
         logger.info("  agentic: bootstrap budget exhausted during document selection")