feat: resolve #78 — [backend-perf] Sprint 2 — cold start <= ~15s (parse-and-release, CSafeLoader, single-pass)

src/lexflow/core/parser.py

@@ -12,6 +12,7 @@
 import logging
 import re
 import unicodedata
+from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 
@@ -51,6 +52,23 @@ def split_frontmatter(content: str) -> tuple[str, str]:
     return yaml_text, body
 
 
+def _load_yaml_dict(yaml_text: str) -> dict[str, Any]:
+    """Parse YAML frontmatter using CSafeLoader when available (#78 S2.2)."""
+    stripped = yaml_text.strip()
+    if not stripped:
+        return {}
+    loader = getattr(yaml, "CSafeLoader", None)
+    if loader is not None:
+        try:
+            data = yaml.load(stripped, Loader=loader)
+            if isinstance(data, dict):
+                return data
+        except yaml.YAMLError:
+            pass
+    data = yaml.safe_load(stripped)
+    return data if isinstance(data, dict) else {}
+
+
 def parse_frontmatter(yaml_text: str) -> dict[str, Any]:
     """Parse a YAML frontmatter string into a raw dictionary.
 
@@ -60,10 +78,9 @@ def parse_frontmatter(yaml_text: str) -> dict[str, Any]:
     if not yaml_text.strip():
         return {}
     try:
-        data = yaml.safe_load(yaml_text)
+        return _load_yaml_dict(yaml_text)
     except yaml.YAMLError as exc:
         raise ParserError("<unknown>", f"Invalid YAML frontmatter: {exc}") from exc
-    return data if isinstance(data, dict) else {}
 
 
 def _safe_enum(enum_cls: type, value: Any, default: Any) -> Any:
@@ -177,22 +194,33 @@ def frontmatter_to_metadata(raw: dict[str, Any]) -> LawMetadata:
 _HEADING_RE = re.compile(r"^(#{1,5})\s+(.+)$", re.MULTILINE)
 
 
-def extract_heading_tree(body: str) -> list[Section]:
+@dataclass
+class _BodyScan:
+    """Precomputed heading and article match positions for a law body (#78 S2.3)."""
+
+    heading_matches: list[tuple[int, str, int]]
+    article_matches: list[re.Match[str]]
+
+
+def _scan_body(body: str) -> _BodyScan:
+    """Scan *body* once for heading and article heading positions."""
+    return _BodyScan(
+        heading_matches=[(len(m.group(1)), m.group(2).strip(), m.start()) for m in _HEADING_RE.finditer(body)],
+        article_matches=list(_ARTICLE_RE.finditer(body)),
+    )
+
+
+def extract_heading_tree(body: str, *, scan: _BodyScan | None = None) -> list[Section]:
     """Parse Markdown headings into a nested :class:`Section` tree.
 
     Walks the body line-by-line, using heading depth to establish
     parent/child relationships.
     """
-    matches: list[tuple[int, str, int]] = []  # (level, heading, start_pos)
-    for m in _HEADING_RE.finditer(body):
-        level = len(m.group(1))
-        heading = m.group(2).strip()
-        matches.append((level, heading, m.start()))
-
-    if not matches:
+    body_scan = scan or _scan_body(body)
+    if not body_scan.heading_matches:
         return []
 
-    return _build_section_list(body, matches, target_level=0)
+    return _build_section_list(body, body_scan.heading_matches, target_level=0, scan=body_scan)
 
 
 def _build_section_list(
@@ -202,6 +230,8 @@ def _build_section_list(
     start_idx: int = 0,
     end_idx: int | None = None,
     body_end: int | None = None,
+    *,
+    scan: _BodyScan | None = None,
 ) -> list[Section]:
     """Recursively build sections for headings at *target_level* depth.
 
@@ -256,6 +286,7 @@ def _build_section_list(
             start_idx=i + 1,
             end_idx=i + (j - i),
             body_end=content_end,
+            scan=scan,
         )
 
         # Extract articles from this section's DIRECT content only — the
@@ -267,7 +298,10 @@ def _build_section_list(
         # their own articles via the recursion above.
         first_sub_start = subset[i + 1][2] if (i + 1) < j else content_end
         direct_body = body[content_start:first_sub_start]
-        articles = extract_articles(direct_body)
+        if scan is not None:
+            articles = _articles_in_span(body, scan, content_start, first_sub_start)
+        else:
+            articles = extract_articles(direct_body)
         section_text = _extract_section_text(direct_body)
 
         sections.append(
@@ -353,16 +387,40 @@ def _extract_section_text(direct_body: str) -> str:
 _ANY_HEADING_LINE_RE = re.compile(r"^#{1,6}[ \t]+\S", re.MULTILINE)
 
 
-def extract_articles(body: str) -> list[Article]:
+def _articles_in_span(body: str, scan: _BodyScan, start: int, end: int) -> list[Article]:
+    """Build articles whose headings fall within ``[start, end)`` using *scan*."""
+    span_matches = [match for match in scan.article_matches if start <= match.start() < end]
+    entries: list[tuple[int, Article]] = []
+    for idx, match in enumerate(span_matches):
+        number = (match.group(1) or match.group(3)).strip()
+        raw_title = match.group(2)
+        title = raw_title.strip() if raw_title else None
+        text_start = match.end()
+        text_end = span_matches[idx + 1].start() if idx + 1 < len(span_matches) else end
+        raw_text = _extract_article_text(body[text_start:text_end])
+        references = extract_references(raw_text, source_article=number)
+        entries.append((match.start(), _build_article(number, title, raw_text, references)))
+
+    existing_numbers = {(match.group(1) or match.group(3)).strip() for match in scan.article_matches}
+    for pos, article in _extract_range_placeholder_articles(body, existing_numbers):
+        if start <= pos < end:
+            entries.append((pos, article))
+
+    entries.sort(key=lambda entry: entry[0])
+    return [article for _, article in entries]
+
+
+def extract_articles(body: str, *, scan: _BodyScan | None = None) -> list[Article]:
     """Extract all articles from a Markdown body.
 
     Finds ``Articulo N.`` patterns, splitting each heading into the article
     number and its optional title (``Artículo 1. Objeto de la Ley.`` →
     number ``"1"``, title ``"Objeto de la Ley"`` — #112), and captures text
     until the next article heading or section heading.
     """
-    matches = list(_ARTICLE_RE.finditer(body))
+    body_scan = scan or _scan_body(body)
     entries: list[tuple[int, Article]] = []
+    matches = body_scan.article_matches
     for idx, match in enumerate(matches):
         number = (match.group(1) or match.group(3)).strip()
         raw_title = match.group(2)
@@ -379,6 +437,16 @@ def extract_articles(body: str) -> list[Article]:
     return [article for _, article in entries]
 
 
+def extract_law_body_structure(body: str) -> tuple[list[Section], list[Article]]:
+    """Extract the section tree and flat article list with one body scan (#78 S2.3)."""
+    scan = _scan_body(body)
+    sections = extract_heading_tree(body, scan=scan)
+    articles = extract_articles(body, scan=scan)
+    if not articles:
+        articles = extract_ordinal_articles(body)
+    return sections, articles
+
+
 def _extract_range_placeholder_articles(body: str, existing_numbers: set[str]) -> list[tuple[int, Article]]:
     """Materialise placeholder articles for plural range headings (#54).
 
@@ -941,6 +1009,25 @@ def _resolve_reference_id(ref_text: str) -> str | None:
 # ---------------------------------------------------------------------------
 
 
+def extract_law_references_from_path(file_path: Path) -> list[Reference]:
+    """Parse only cross-references from a law file without caching a full :class:`Law` (#78 S2.1)."""
+    try:
+        content = file_path.read_text(encoding="utf-8")
+    except FileNotFoundError as exc:
+        raise ParserError(str(file_path), "File not found") from exc
+    except OSError as exc:
+        raise ParserError(str(file_path), f"Cannot read file: {exc}") from exc
+    return extract_law_references_from_content(content)
+
+
+def extract_law_references_from_content(content: str) -> list[Reference]:
+    """Collect cross-references from law body text without building a full :class:`Law`."""
+    _, body = split_frontmatter(content)
+    sections, articles = extract_law_body_structure(body)
+    disposiciones = extract_disposiciones(body)
+    return _collect_all_references(articles, disposiciones, sections)
+
+
 def parse_law_file(file_path: Path) -> Law:
     """Parse a complete ``.md`` law file into a :class:`Law` model.
 
@@ -978,12 +1065,7 @@ def parse_law_content(content: str, file_path: str) -> Law:
     except ParserError as exc:
         raise ParserError(file_path, exc.reason) from exc
     metadata = frontmatter_to_metadata(raw_fm)
-    sections = extract_heading_tree(body)
-    articles = extract_articles(body)
-    if not articles:
-        # #54: ~38% of the 1,857 zero-article laws use numbered ordinals
-        # (``Primero.``, ``Único.``, ...) instead of ``Artículo`` headings.
-        articles = extract_ordinal_articles(body)
+    sections, articles = extract_law_body_structure(body)
     disposiciones = extract_disposiciones(body)
     all_references = _collect_all_references(articles, disposiciones, sections)
 

src/lexflow/core/registry.py

@@ -7,8 +7,10 @@
 from __future__ import annotations
 
 import logging
+import os
 from collections import Counter
 from collections.abc import Callable
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from functools import lru_cache
 from pathlib import Path
 from threading import Lock
@@ -391,12 +393,44 @@ def preload_all_metadata(self) -> None:
         """Parse frontmatter for all laws in the index.
 
         Called as a background task at startup to make subsequent list/filter
-        operations fast.
+        operations fast. Uses a worker pool and CSafeLoader-backed YAML
+        parsing for cold-start throughput (#78 S2.2).
         """
+        pending = [law_id for law_id in self._snapshot_law_ids() if law_id not in self._metadata_cache]
+        if not pending:
+            logger.info("Metadata preload complete: 0 laws loaded")
+            return
+
+        workers = min(32, (os.cpu_count() or 1) + 4)
         loaded = 0
-        for law_id in self._snapshot_law_ids():
-            if law_id not in self._metadata_cache and self._safe_metadata(law_id) is not None:
-                loaded += 1
+        skipped: set[str] = set()
+
+        def _parse_one(law_id: str) -> tuple[str, LawMetadata | None]:
+            path = self._index.get(law_id)
+            if path is None:
+                return law_id, None
+            try:
+                return law_id, parse_metadata_only(path)
+            except (OSError, ValueError, LexFlowError):
+                logger.warning("Skipping malformed law %s during metadata preload", law_id, exc_info=True)
+                return law_id, None
+
+        with ThreadPoolExecutor(max_workers=workers) as executor:
+            futures = {executor.submit(_parse_one, law_id): law_id for law_id in pending}
+            for future in as_completed(futures):
+                law_id, metadata = future.result()
+                if metadata is None:
+                    skipped.add(law_id)
+                    continue
+                with self._lock:
+                    if law_id not in self._metadata_cache:
+                        self._metadata_cache[law_id] = metadata
+                        loaded += 1
+
+        if skipped:
+            with self._lock:
+                self._skipped_law_ids.update(skipped)
+
         logger.info("Metadata preload complete: %d laws loaded", loaded)
 
     # ------------------------------------------------------------------

src/lexflow/graph/builder.py

@@ -9,16 +9,19 @@
 from __future__ import annotations
 
 import logging
+import os
 import unicodedata
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from pathlib import Path
 
 from lexflow.core.delta_sync import CorpusDiff
 from lexflow.core.enums import EdgeResolution, ReferenceKind, parse_edge_resolution
 from lexflow.core.exceptions import LexFlowError
-from lexflow.core.models import Law
+from lexflow.core.models import Reference
 
 # Reuse the parser's citation pattern so a reference's text and a law title's
 # leading citation are matched by the exact same rule (no drift, #569).
-from lexflow.core.parser import _LAW_REF_RE
+from lexflow.core.parser import _LAW_REF_RE, extract_law_references_from_path
 from lexflow.core.registry import LawRegistry
 from lexflow.graph.algorithms import enrich_graph_analytics
 from lexflow.graph.model import LegalGraph
@@ -60,6 +63,50 @@ def _build_citation_index(registry: LawRegistry) -> dict[str, str]:
     return index
 
 
+def _default_graph_workers() -> int:
+    """Worker count for parallel reference extraction during cold graph builds."""
+    return min(32, (os.cpu_count() or 1) + 4)
+
+
+def _extract_references_for_path(law_id: str, path: Path) -> tuple[str, list[Reference]] | None:
+    """Parse references from one law file without touching the registry cache."""
+    try:
+        return law_id, extract_law_references_from_path(path)
+    except (OSError, ValueError, LexFlowError):
+        logger.warning("Could not extract references for %s", law_id, exc_info=True)
+        return None
+
+
+def _collect_law_references(registry: LawRegistry) -> dict[str, list[Reference]]:
+    """Extract cross-references for every indexed law in a worker pool (#78 S2.1)."""
+    paths: list[tuple[str, Path]] = []
+    fallback_ids: list[str] = []
+    path_lookup = getattr(registry, "law_file_path", None)
+    for law_id in registry.law_ids:
+        path = path_lookup(law_id) if path_lookup is not None else None
+        if path is not None:
+            paths.append((law_id, path))
+        else:
+            fallback_ids.append(law_id)
+
+    references_by_law: dict[str, list[Reference]] = {}
+    workers = _default_graph_workers()
+    with ThreadPoolExecutor(max_workers=workers) as executor:
+        futures = {executor.submit(_extract_references_for_path, law_id, path): law_id for law_id, path in paths}
+        for future in as_completed(futures):
+            result = future.result()
+            if result is not None:
+                references_by_law[result[0]] = result[1]
+
+    for law_id in fallback_ids:
+        try:
+            references_by_law[law_id] = registry.get_law(law_id).references
+        except (OSError, ValueError, LexFlowError):
+            logger.warning("Could not process references for %s", law_id, exc_info=True)
+
+    return references_by_law
+
+
 def build_graph(registry: LawRegistry) -> LegalGraph:
     graph = LegalGraph()
     # Pass 1: add all law nodes
@@ -77,12 +124,12 @@ def build_graph(registry: LawRegistry) -> LegalGraph:
     # them; only resolved edges are counted for the log.
     added_edges = 0
     citation_index = _build_citation_index(registry)
+    references_by_law = _collect_law_references(registry)
     for law_id in registry.law_ids:
-        try:
-            law = registry.get_law(law_id)
-            added_edges += _add_law_edges(graph, law_id, law, citation_index)
-        except (OSError, ValueError, LexFlowError):
-            logger.warning("Could not process references for %s", law_id, exc_info=True)
+        references = references_by_law.get(law_id)
+        if references is None:
+            continue
+        added_edges += _add_law_edges(graph, law_id, references, citation_index)
     enrich_graph_analytics(graph)
     logger.info("Graph built: %d nodes, %d edges", graph.node_count(), added_edges)
     return graph
@@ -109,7 +156,12 @@ def apply_diff_to_graph(graph: LegalGraph, registry: LawRegistry, diff: CorpusDi
     enrich_graph_analytics(graph)
 
 
-def _add_law_edges(graph: LegalGraph, law_id: str, law: Law, citation_index: dict[str, str]) -> int:
+def _add_law_edges(
+    graph: LegalGraph,
+    law_id: str,
+    references: list[Reference],
+    citation_index: dict[str, str],
+) -> int:
     """Add one law's outgoing edges; park unresolved targets as dangling.
 
     Returns the number of edges actually inserted (target was a known node).
@@ -122,7 +174,7 @@ def _add_law_edges(graph: LegalGraph, law_id: str, law: Law, citation_index: dic
     ``inferred`` when the target came from ``citation_index`` (#64).
     """
     added = 0
-    for ref in law.references:
+    for ref in references:
         if ref.target_id:
             target_id = ref.target_id
             resolution = EdgeResolution.BOE_ID
@@ -184,7 +236,8 @@ def _upsert_law(graph: LegalGraph, registry: LawRegistry, law_id: str, citation_
     graph.add_law(registry.get_metadata(law_id))
     graph.drop_source_from_dangling(law_id)
     graph.clear_outgoing(law_id)
-    _add_law_edges(graph, law_id, registry.get_law(law_id), citation_index)
+    law = registry.get_law(law_id)
+    _add_law_edges(graph, law_id, law.references, citation_index)
 
 
 def _resolve_incoming(graph: LegalGraph, law_id: str) -> None: