feat(comprehension): ingestion-time LLM digests so ask/chat answer structure + combination questions

.env.example

@@ -116,3 +116,71 @@ CRE_LIBRARIAN_LINK_THRESHOLD=0.8
 CRE_LIBRARIAN_BATCH_SIZE=32
 CRE_LIBRARIAN_ECE_TARGET=0.10
 CRE_LIBRARIAN_CONFORMAL_ALPHA=0.10
+
+# --- OWASP agent answers: ingestion-time comprehension (#1158) ---------------------------
+# Values below are the measured winning combination on staging (full 589-question set:
+# whole-project context 57.6% correct vs 54.7% with no flags, ~20 s end to end). Every new
+# feature is off in code unless its flag is set, so production behaves as before without them.
+# Retrieval depth for the OWASP agent lane (default 8) and for comprehension documents.
+OWASP_AGENT_TOP_K=16
+OWASP_COMPREHENSION_TOP_K=24
+# Answer from ingestion-time comprehension documents; say "not found" instead of guessing.
+OWASP_COMPREHENSION_ENABLED=1
+OWASP_COMPREHENSION_ABSTAIN=1
+# How many retrieved documents the answer reads (more did not help: 24 scored the same as 10).
+OWASP_COMPREHENSION_SOURCES=10
+# Search only "raw" source chunks or only "digest" documents (default all). Raw chunks alone
+# score within ~3 points of all, which is the cheap full-import option.
+OWASP_COMPREHENSION_DOCS=all
+# Model that writes the final answer ('' = the default chat model). Gemini Pro did not help.
+OWASP_COMPREHENSION_MODEL=
+# Scope retrieval to the project named in the question as a score boost. Hurt (-10 points): keep 0.
+OWASP_COMPREHENSION_SCOPE=0
+# Half-precision HNSW vector search. Needs migration e5f6a7b8c9d0 (the index); 1 cuts ~10 s
+# of database time per question on a large embeddings table.
+OWASP_PGVECTOR_HALFVEC=0
+
+# Feature flags (1 = on). Measured on the 252-question sample unless noted.
+# L4 whole-project context: when a question names exactly one project that fits, the model
+# reads the project's whole text instead of snippets (+3 points on the full set).
+OWASP_CX_LONGCTX=1
+# Max characters of one project to read whole.
+OWASP_CX_LONGCTX_CHARS=2000000
+# How many named projects to read whole at once (questions naming more fall back to retrieval).
+# 1 scored 59.1% on the sample; 3 scored 56.3% (counts and multi-part got worse), so keep 1.
+OWASP_CX_LONGCTX_PROJECTS=1
+# Off: hybrid keyword search (-2.5 points, +5 s), rerank (-2), query rewrite (-3), verify
+# (+/-3, hurts with LONGCTX), per-project (0), expand (+1, noise), count (+1, slower),
+# merged single call (-13, loses citations), in-process answer cache (real traffic only).
+# Final-answer model with thinking off (latency); multi-part questions answered part by part.
+OWASP_CX_FAST=0
+# Thinking budget for the final answer: '' (model default), low, medium or high. Measured: low and
+# medium are NOT faster than the default (~7 s); only OWASP_CX_FAST=1 (thinking off, 2.5 s but
+# 49% on the sample, 54% with OWASP_COMPREHENSION_MODEL=gemini/gemini-2.5-flash-lite) is.
+OWASP_CX_THINK_EFFORT=
+OWASP_CX_DECOMPOSE=0
+OWASP_CX_HYBRID=0
+OWASP_CX_RERANK=0
+OWASP_CX_QUERY=0
+OWASP_CX_VERIFY=0
+OWASP_CX_PERPROJECT=0
+OWASP_CX_EXPAND=0
+OWASP_CX_COUNT=0
+OWASP_CX_MERGED=0
+OWASP_CX_CACHE=0
+# Cheap model used by the rerank/query/verify helpers when those flags are on.
+OWASP_CX_HELPER_MODEL=gemini/gemini-2.5-flash-lite
+
+# Ingestion (scripts/comprehend_standard.py): bulk digests and chunk enrichment use a cheap
+# model with thinking off; repo overviews use a stronger one. Run with --estimate first.
+CRE_COMPREHENSION_BULK_MODEL=gemini/gemini-2.5-flash
+CRE_COMPREHENSION_OVERVIEW_MODEL=gemini/gemini-2.5-pro
+CRE_COMPREHENSION_ENRICH_MODEL=gemini/gemini-2.5-flash-lite
+CRE_COMPREHENSION_THINKING=off
+
+# Warm the answer path in the background at server start (the first question after a restart
+# otherwise takes ~30 s). 0 disables.
+OWASP_WARMUP=1
+
+# Remote MCP server: how many questions may be answered at the same time (each in its own thread).
+OPENCRE_MCP_ASK_CONCURRENCY=8

application/mcp/ask.py

@@ -211,6 +211,8 @@ def _owasp_meta_facts(
         "intent": intent,
         "message": resp.get("response"),
         "items": owasp_meta.get("data"),
+        # The source passages the answer used, so a calling agent can check it.
+        "evidence": (owasp_meta.get("data") or {}).get("evidence") or [],
         "owasp_agent": owasp_meta,
         "citations": list(owasp_meta.get("citations") or []),
         "standards_filter": None,

application/mcp/cli.py

@@ -90,6 +90,15 @@ def _serve_http() -> int:
     ctx.push()
     try:
         asgi = create_http_app()
+        import threading
+
+        from application.utils.comprehension.shared import warm_up
+
+        def _warm() -> None:
+            with flask_app.app_context():
+                warm_up()
+
+        threading.Thread(target=_warm, name="opencre-warmup", daemon=True).start()
         _say(f"OpenCRE remote MCP (Streamable HTTP) on http://{host}:{port}/mcp")
         # One worker: sessionful Streamable HTTP clients (sticky if scaled later).
         uvicorn.run(asgi, host=host, port=port, workers=1, log_level="info")

application/mcp/owasp_agent_tools.py

@@ -76,6 +76,60 @@
             "additionalProperties": False,
         },
     },
+    {
+        "name": "owasp_meta_sources_search",
+        "summary": (
+            "Search the indexed OWASP source text (not CRE). Returns snippets with paths and "
+            "citation URLs so a caller can verify an answer."
+        ),
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "query": {"type": "string"},
+                "project": {
+                    "type": "string",
+                    "description": "Optional project, e.g. ASVS",
+                },
+                "k": {"type": "integer", "default": 8},
+            },
+            "required": ["query"],
+            "additionalProperties": False,
+        },
+    },
+    {
+        "name": "owasp_meta_sources_list",
+        "summary": "List the indexed source documents of one OWASP project.",
+        "input_schema": {
+            "type": "object",
+            "properties": {"project": {"type": "string"}},
+            "required": ["project"],
+            "additionalProperties": False,
+        },
+    },
+    {
+        "name": "owasp_meta_sources_get",
+        "summary": "Read one indexed source document of a project in full (truncated at max_chars).",
+        "input_schema": {
+            "type": "object",
+            "properties": {
+                "project": {"type": "string"},
+                "path": {"type": "string"},
+                "max_chars": {"type": "integer", "default": 20000},
+            },
+            "required": ["project", "path"],
+            "additionalProperties": False,
+        },
+    },
+    {
+        "name": "owasp_meta_sources_overview",
+        "summary": "Short generated overview of an OWASP project (structure, editions, scope).",
+        "input_schema": {
+            "type": "object",
+            "properties": {"project": {"type": "string"}},
+            "required": ["project"],
+            "additionalProperties": False,
+        },
+    },
 )
 
 _OWASP_META_BY_NAME = {t["name"]: t for t in OWASP_META_TOOLS}
@@ -102,6 +156,8 @@ def call_owasp_meta_tool(
     if name not in _OWASP_META_BY_NAME:
         raise KeyError(f"Unknown OWASP meta tool: {name}")
     args = dict(arguments or {})
+    if name.startswith("owasp_meta_sources_"):
+        return _call_sources_tool(name, args)
     store = IndexStore()
     queries = MetaQueries(store)
 
@@ -177,3 +233,32 @@ def _result_payload(result: Any) -> Dict[str, Any]:
         "data": result.data,
         "citations": list(result.citations or []),
     }
+
+
+def _call_sources_tool(name: str, args: Dict[str, Any]) -> Dict[str, Any]:
+    from application import sqla
+    from application.utils.comprehension import toolkit
+
+    with sqla.session.get_bind().connect() as conn:
+        if name == "owasp_meta_sources_search":
+            query = str(args.get("query") or "").strip()
+            if not query:
+                raise ValueError("query is required")
+            return toolkit.search_sources(
+                conn, query, args.get("project"), int(args.get("k") or 8)
+            )
+        project = str(args.get("project") or "").strip()
+        if not project:
+            raise ValueError("project is required")
+        if name == "owasp_meta_sources_list":
+            return toolkit.list_project_docs(conn, project)
+        if name == "owasp_meta_sources_overview":
+            return toolkit.project_overview(conn, project)
+        if name == "owasp_meta_sources_get":
+            path = str(args.get("path") or "").strip()
+            if not path:
+                raise ValueError("path is required")
+            return toolkit.get_source(
+                conn, project, path, int(args.get("max_chars") or 20000)
+            )
+    raise KeyError(f"Unknown OWASP meta tool: {name}")

application/mcp/server.py

@@ -7,6 +7,7 @@
 logger = get_logger(__name__)
 
 import json
+import os
 from typing import Any, Dict, Optional
 
 import mcp.types as types
@@ -70,6 +71,41 @@ def server_instructions() -> str:
     return base
 
 
+_BLOCKING_LIMITER: Any = None
+
+
+async def run_blocking_with_app(fn: Any) -> Any:
+    """Run a blocking call in a worker thread so one slow question cannot stall the server.
+
+    ``ask`` takes seconds (LLM calls). Called straight from an async handler it blocks the
+    event loop, so parallel users are answered one after another. Each call gets its own Flask
+    app context (database sessions are scoped per context and must not be shared across
+    threads); without a Flask app (stdio mode) it runs inline.
+    """
+    global _BLOCKING_LIMITER
+    try:
+        from flask import current_app
+
+        app = current_app._get_current_object()  # type: ignore[attr-defined]
+    except RuntimeError:
+        return fn()
+
+    import anyio
+
+    if _BLOCKING_LIMITER is None:
+        try:
+            cap = max(1, int(os.environ.get("OPENCRE_MCP_ASK_CONCURRENCY", "8")))
+        except ValueError:
+            cap = 8
+        _BLOCKING_LIMITER = anyio.CapacityLimiter(cap)
+
+    def _call() -> Any:
+        with app.app_context():
+            return fn()
+
+    return await anyio.to_thread.run_sync(_call, limiter=_BLOCKING_LIMITER)
+
+
 def build_server(rest_client: Optional[RestClient] = None) -> Server[Any]:
     """Create an MCP Server with OpenAPI-derived tool schemas and REST dispatch."""
     client = rest_client or RestClient()
@@ -150,15 +186,17 @@ async def on_call_tool(
                             ],
                             is_error=True,
                         )
-                data = run_ask(
-                    str(arguments.get("prompt") or ""),
-                    corpus=str(arguments.get("corpus") or "auto"),
-                    place=(
-                        str(arguments["place"])
-                        if arguments.get("place") is not None
-                        else None
-                    ),
-                    rest_client=client,
+                data = await run_blocking_with_app(
+                    lambda: run_ask(
+                        str(arguments.get("prompt") or ""),
+                        corpus=str(arguments.get("corpus") or "auto"),
+                        place=(
+                            str(arguments["place"])
+                            if arguments.get("place") is not None
+                            else None
+                        ),
+                        rest_client=client,
+                    )
                 )
                 payload = json.dumps(data, ensure_ascii=False, default=str)
                 return types.CallToolResult(
@@ -169,7 +207,12 @@ async def on_call_tool(
                     is_error=False,
                 )
             if get_owasp_meta_tool(name) is not None:
-                data = call_owasp_meta_tool(name, arguments)
+                if name.startswith("owasp_meta_sources_"):
+                    data = await run_blocking_with_app(
+                        lambda: call_owasp_meta_tool(name, arguments)
+                    )
+                else:
+                    data = call_owasp_meta_tool(name, arguments)
                 payload = json.dumps(data, ensure_ascii=False, default=str)
                 return types.CallToolResult(
                     content=[types.TextContent(type="text", text=payload)],

application/tests/agent_hnsw_migration_test.py

@@ -0,0 +1,71 @@
+import importlib.util
+import pathlib
+import unittest
+from unittest import mock
+
+_PATH = (
+    pathlib.Path(__file__).resolve().parents[2]
+    / "migrations"
+    / "versions"
+    / "e5f6a7b8c9d0_add_agent_halfvec_hnsw_index.py"
+)
+
+
+def _load():
+    spec = importlib.util.spec_from_file_location("agent_hnsw_migration", _PATH)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+class AgentHnswMigrationTest(unittest.TestCase):
+    def test_merges_the_two_previous_heads(self) -> None:
+        m = _load()
+        self.assertEqual(set(m.down_revision), {"c9d0e1f2a3b4", "f5a6b7c8d9e0"})
+
+    def test_noop_off_postgres(self) -> None:
+        m = _load()
+        bind = mock.Mock()
+        bind.dialect.name = "sqlite"
+        with mock.patch.object(m.op, "get_bind", return_value=bind), mock.patch.object(
+            m.op, "execute"
+        ) as execute:
+            m.upgrade()
+            m.downgrade()
+        execute.assert_not_called()
+
+    def test_builds_index_with_the_column_dimension(self) -> None:
+        m = _load()
+        bind = mock.Mock()
+        bind.dialect.name = "postgresql"
+        bind.exec_driver_sql.side_effect = [
+            mock.Mock(fetchone=lambda: (1,)),  # vector extension present
+            mock.Mock(fetchone=lambda: (3072,)),  # column typmod
+        ]
+        ctx = mock.MagicMock()
+        with mock.patch.object(m.op, "get_bind", return_value=bind), mock.patch.object(
+            m.op, "get_context", return_value=ctx
+        ), mock.patch.object(m.op, "execute") as execute:
+            m.upgrade()
+        sql = execute.call_args[0][0]
+        self.assertIn("halfvec(3072)", sql)
+        self.assertIn("CONCURRENTLY", sql)
+        self.assertIn("source_channel = 'owasp_agent_meta'", sql)
+
+    def test_skips_columns_wider_than_halfvec_limit(self) -> None:
+        m = _load()
+        bind = mock.Mock()
+        bind.dialect.name = "postgresql"
+        bind.exec_driver_sql.side_effect = [
+            mock.Mock(fetchone=lambda: (1,)),
+            mock.Mock(fetchone=lambda: (8192,)),
+        ]
+        with mock.patch.object(m.op, "get_bind", return_value=bind), mock.patch.object(
+            m.op, "execute"
+        ) as execute:
+            m.upgrade()
+        execute.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()