fix(cli): prevent doctor Note/Hint columns from wrapping into unreadable fragments (#1428) (#1475)

* fix(cli): prevent doctor Note/Hint columns from wrapping into unreadable fragments (#1428)

* fix(cli): keep doctor check labels readable

---------

Co-authored-by: Sameer Kadam <sskadam6305@gmail.com>
Co-authored-by: Sameer Kadam <sameerkadam@Sameers-MacBook-Air.local>

.github/workflows/release.yml

@@ -67,6 +67,12 @@ jobs:
         run: |
           pip install -r .github/requirements/twine.txt --require-hashes
           twine check dist/*
+      # pypi-publish uploads everything under packages-dir (default: dist/) with
+      # no glob/include filter, so it must run before anything else writes a
+      # non-distribution file into dist/ - the Sigstore step below does exactly
+      # that (dist/*.sigstore.json), and pypi-publish fails on it with
+      # "InvalidDistribution: Unknown distribution format" if it runs after.
+      - uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # release/v1
       - name: Attest build provenance
         uses: actions/attest-build-provenance@4d101475d8b20a2381f78447822ac1eab6504dd8 # v4
         with:
@@ -75,7 +81,7 @@ jobs:
       # OpenSSF Scorecard's Signed-Releases check does not inspect - it looks for
       # signature files attached as release assets. Sign here too so
       # `dist/*.sigstore.json` bundles ship alongside the wheel/sdist on the
-      # GitHub Release itself.
+      # GitHub Release itself. This must run after pypi-publish (see above).
       - name: Sign artifacts with Sigstore
         uses: sigstore/gh-action-sigstore-python@790bc6befb9d733738f18d8f895854b453640ec9 # v3.5.0
         with:
@@ -88,4 +94,3 @@ jobs:
             dist/*.whl
             dist/*.tar.gz
             dist/*.sigstore.json
-      - uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # release/v1

semantica/cli.py

@@ -917,11 +917,11 @@ def doctor(cli_ctx: CLIContext, local_json: bool, deep_embeddings: bool) -> None
                     for lbl, st, note, hint in checks])
             return
 
-        tbl = Table(box=_TABLE_BOX, show_edge=False, padding=(0, 2))
-        tbl.add_column("Check",  style=_KEY, no_wrap=True, min_width=16)
-        tbl.add_column("Status", no_wrap=True, min_width=6)
-        tbl.add_column("Note",   style=_DIM)
-        tbl.add_column("Hint",   style=_DIM)
+        tbl = Table(box=_TABLE_BOX, show_edge=False, padding=(0, 2), expand=True)
+        tbl.add_column("Check",  style=_KEY, no_wrap=True, min_width=34)
+        tbl.add_column("Status", no_wrap=True, min_width=4)
+        tbl.add_column("Note",   style=_DIM, min_width=15, ratio=2, overflow="fold")
+        tbl.add_column("Hint",   style=_DIM, min_width=20, ratio=3, overflow="fold")
 
         icons = {"ok": f"[{_SUCCESS}] ✓[/{_SUCCESS}]",
                  "warn": f"[{_WARN_STY}] ⚠[/{_WARN_STY}]",
@@ -1155,6 +1155,57 @@ def _get_graph_store(cli_ctx: CLIContext) -> Any:
     return GraphStore(backend=backend, **graph_db)
 
 
+def _load_rule_definitions(path: str) -> List[str]:
+    """Load reasoning rule definitions from a YAML or plain-text rules file.
+
+    YAML files may hold a list of rule strings or a mapping with a ``rules``
+    list; anything else (e.g. Datalog) is read as one rule per non-comment
+    line. The strings are handed to ``Reasoner.add_rule()`` untouched.
+    """
+    text = Path(path).read_text(encoding="utf-8")
+    try:
+        data = yaml.safe_load(text)
+    except yaml.YAMLError:
+        data = None
+    if isinstance(data, dict):
+        rules_value = data.get("rules")
+        if rules_value is None and "rules" not in data:
+            raise click.ClickException(
+                f"Rules file '{path}' is a YAML mapping but has no 'rules' key. "
+                "Expected either a YAML list or a mapping with a 'rules' list."
+            )
+        data = rules_value
+    if isinstance(data, list):
+        return [str(item) for item in data]
+    return [line.strip() for line in text.splitlines()
+            if line.strip() and not line.lstrip().startswith("#")]
+
+
+def _graph_store_facts(cli_ctx: CLIContext) -> List[str]:
+    """Read the configured graph store into Reasoner fact strings.
+
+    Follows the same conventions ``Reasoner.add_fact()`` applies to
+    KG-style dicts: nodes become ``Label(name)`` and relationships become
+    ``TYPE(source, target)``, with internal node ids resolved to names.
+    """
+    gs = _get_graph_store(cli_ctx)
+    nodes = gs.get_nodes(limit=sys.maxsize)
+    relationships = gs.get_relationships(limit=sys.maxsize)
+    names: Dict[Any, Any] = {}
+    facts: List[str] = []
+    for node in nodes:
+        props = node.get("properties") or {}
+        name = props.get("name") or props.get("id") or node.get("id")
+        names[node.get("id")] = name
+        for label in node.get("labels") or ["Entity"]:
+            facts.append(f"{label}({name})")
+    for rel in relationships:
+        source = names.get(rel.get("start_node_id"), rel.get("start_node_id"))
+        target = names.get(rel.get("end_node_id"), rel.get("end_node_id"))
+        facts.append(f"{rel.get('type', 'RELATED_TO')}({source}, {target})")
+    return facts
+
+
 # ─── Output helpers ──────────────────────────────────────────────────────────
 
 
@@ -1415,6 +1466,31 @@ _INGEST_TYPES = [
 ]
 
 _INGEST_FORMATS = ["pdf", "docx", "csv", "excel", "html", "json", "parquet", "xml", "rdf"]
+_GRAPH_STORE_ENV_BACKEND_HINTS = {
+    "GRAPH_STORE_NEO4J_URI": "neo4j",
+    "GRAPH_STORE_FALKORDB_HOST": "falkordb",
+    "GRAPH_STORE_NEPTUNE_ENDPOINT": "neptune",
+    "GRAPH_STORE_AGE_CONNECTION_STRING": "age",
+}
+
+
+def _configured_ingest_graph_backend(
+    cli_ctx: CLIContext, store_override: Optional[str]
+) -> Optional[str]:
+    graph_db = dict(cli_ctx.config.to_dict().get("graph_db", {}))
+    backend = store_override or cli_ctx.store_backend or graph_db.get("backend")
+    if backend:
+        return str(backend)
+
+    env_backend = os.environ.get("GRAPH_STORE_DEFAULT_BACKEND")
+    if env_backend:
+        return env_backend
+
+    for env_var, hinted_backend in _GRAPH_STORE_ENV_BACKEND_HINTS.items():
+        if os.environ.get(env_var):
+            return hinted_backend
+
+    return None
 
 
 @main.command()
@@ -1427,8 +1503,13 @@ _INGEST_FORMATS = ["pdf", "docx", "csv", "excel", "html", "json", "parquet", "xm
 @click.option("--watch", is_flag=True, default=False, help="Re-ingest on file changes.")
 @click.option("--batch-size", default=500, type=int, show_default=True)
 @click.option("--store", "store_override", default=None,
-              help="Target graph backend: neo4j falkordb age neptune")
-@click.option("--output", default=None, type=click.Path(), help="Write to file instead of graph store.")
+              help="Target graph backend: neo4j falkordb age neptune. "
+                   "Not yet implemented — ingest cannot persist to a graph "
+                   "store, so this only determines whether the command "
+                   "refuses to report false success; use --output instead.")
+@click.option("--output", default=None, type=click.Path(),
+              help="Write ingested content to a .json/.jsonl/.csv file "
+                   "instead of the (unimplemented) graph store.")
 @click.option("--dry-run", "local_dry", is_flag=True, default=False)
 @click.option("--json", "local_json", is_flag=True, default=False)
 @click.pass_obj
@@ -1452,6 +1533,19 @@ def ingest(
             _dry(cli_ctx, "ingest", json_out=_is_json(cli_ctx, local_json),
                  source=source, type=ingestor_type, format=fmt)
             return
+        graph_backend = _configured_ingest_graph_backend(cli_ctx, store_override)
+        if graph_backend and graph_backend.lower() != "memory" and not output:
+            raise click.ClickException(
+                f"A graph backend is configured ({graph_backend}), but "
+                "semantica ingest does not write to graph stores yet — no CLI "
+                "command currently does (tracked in issues #1351, #1352). "
+                "Pass --output <file>.json to save the ingested content "
+                "instead, or build a GraphStore/GraphBuilder directly in "
+                "Python."
+            )
+        # NOTE: --store/GRAPH_STORE_DEFAULT_BACKEND are read only to decide
+        # whether to raise the error above — nothing downstream of this point
+        # writes to a graph store, so neither is forwarded as an ingest kwarg.
         kwargs: Dict[str, Any] = {"batch_size": batch_size}
         if ingestor_type:
             kwargs["source_type"] = ingestor_type
@@ -1461,10 +1555,6 @@ def ingest(
             kwargs["recursive"] = True
         if watch:
             kwargs["watch"] = True
-        if store_override or cli_ctx.store_backend:
-            kwargs["store"] = store_override or cli_ctx.store_backend
-        if output:
-            kwargs["output"] = output
         try:
             from .ingest import ingest as _ingest
             label = Path(source).name if Path(source).exists() else source
@@ -1478,7 +1568,10 @@ def ingest(
                     result = _ingest(source, **kwargs)
         except ImportError as exc:
             raise click.ClickException(f"Ingest module not available: {exc}") from exc
-        if _is_json(cli_ctx, local_json):
+        if output:
+            _write_result_output(Path(output), result)
+            _ok(cli_ctx, f"Wrote {output}")
+        elif _is_json(cli_ctx, local_json):
             _jecho(result if isinstance(result, dict) else {"status": "ok"})
         else:
             _ok(cli_ctx, f"Ingested: {source}")
@@ -1735,9 +1828,19 @@ def embed(ctx: click.Context) -> None:
 def _json_default(obj) -> object:
     """JSON serialiser that converts NumPy scalars/arrays to native Python types.
 
+    Also expands dataclasses (e.g. ``FileObject`` from ``ingest``) to plain
+    dicts and decodes ``bytes`` as UTF-8 text where possible, so a domain
+    object round-trips through ``--output`` as data instead of a repr string.
     Falls back to ``str()`` for everything else so the writer never crashes on
-    unexpected types (e.g. ``datetime``, custom domain objects).
+    unexpected types (e.g. ``datetime``).
     """
+    if is_dataclass(obj) and not isinstance(obj, type):
+        return asdict(obj)
+    if isinstance(obj, bytes):
+        try:
+            return obj.decode("utf-8")
+        except UnicodeDecodeError:
+            return obj.hex()
     try:
         import numpy as np  # local import — only needed when result contains numpy
         if isinstance(obj, np.ndarray):
@@ -2160,17 +2263,43 @@ def reason_run(cli_ctx: CLIContext, engine: str, rules: Optional[str],
     cli_ctx = _require_ctx(cli_ctx)
 
     def _action() -> None:
+        # Only the forward-chaining production-rule engines run through
+        # Reasoner.infer_facts(); the other engines take different inputs
+        # (SPARQL/Datalog queries, observations, premises) and are not wired
+        # to this command yet. Fail honestly instead of silently
+        # forward-chaining under another engine's name.
+        if engine not in ("rete", "forward-chain"):
+            hint = (" Use 'semantica reason query' for SPARQL/Datalog queries."
+                    if engine in ("sparql", "datalog") else "")
+            raise click.ClickException(
+                f"Engine '{engine}' is not wired to 'reason run' yet; "
+                f"supported engines: rete, forward-chain.{hint}")
         try:
             from .reasoning import Reasoner
+            # Reasoner has no run() method (#1354); dispatch to its real
+            # API: facts from the configured graph store + rules from the
+            # optional --rules file into infer_facts().
             r = Reasoner(engine=engine, config=cli_ctx.config.to_dict())
+            rule_defs = _load_rule_definitions(rules) if rules else None
+            facts = _graph_store_facts(cli_ctx)
+
+            def _infer() -> Dict[str, Any]:
+                inferred = r.infer_facts(facts, rule_defs)
+                return {
+                    "engine": engine,
+                    "facts": len(facts),
+                    "inferred_count": len(inferred),
+                    "inferred_facts": inferred,
+                }
+
             if cli_ctx.quiet or cli_ctx.json_output:
-                result = r.run(rules_file=rules)
+                result = _infer()
             else:
                 with console.status(
                     f"[{_DIM}]Running {engine} reasoning engine…[/{_DIM}]",
                     spinner="dots",
                 ):
-                    result = r.run(rules_file=rules)
+                    result = _infer()
         except ImportError as exc:
             raise click.ClickException(f"Reasoning module not available: {exc}") from exc
         if _is_json(cli_ctx, local_json):
@@ -3661,14 +3790,17 @@ def store_connect(cli_ctx: CLIContext, backend: str, uri: Optional[str], local_j
 
     def _action() -> None:
         try:
-            from .graph_store import get_graph_store_method
-            store_cls = get_graph_store_method(backend)
+            # get_graph_store_method(task, method_name) is the method
+            # registry, not a backend factory (#1354); build the store
+            # through GraphStore, which resolves the backend by name.
+            from .graph_store import GraphStore
             cfg = dict(cli_ctx.config.to_dict().get("graph_db", {}))
+            cfg.pop("backend", None)
             if uri:
                 cfg["uri"] = uri
-            # Attempt instantiation as the minimal connectivity probe; backends
-            # that require a live connection will fail here if unreachable.
-            store_instance = store_cls(config=cfg)
+            # Instantiation only wires the backend; the probe below performs
+            # the live connectivity check and raises if unreachable.
+            store_instance = GraphStore(backend=backend, **cfg)
             for probe in ("health_check", "ping", "connect"):
                 fn = getattr(store_instance, probe, None)
                 if callable(fn):

semantica/context/erasure.py

@@ -34,6 +34,7 @@ Example:
     'unsupported'
 """
 
+import copy
 import inspect
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
@@ -117,13 +118,21 @@ class ErasureReceipt:
         ]
 
     def to_dict(self) -> Dict[str, Any]:
-        """Serialize the receipt, deep-copying the per-store results."""
+        """Serialize the receipt, deep-copying the per-store results.
+
+        Each store result is copied recursively so the returned payload shares
+        no mutable objects with the live receipt: mutating
+        ``payload["stores"][name][...]`` (including nested dicts such as a
+        vector backend's ``backend_result``) cannot corrupt the audit record.
+        """
         return {
             "entity_id": self.entity_id,
             "reason": self.reason,
             "erased_at": self.erased_at,
             "complete": self.complete,
-            "stores": {name: dict(result) for name, result in self.stores.items()},
+            "stores": {
+                name: copy.deepcopy(result) for name, result in self.stores.items()
+            },
         }
 
 

semantica/explorer/routes/ontology.py

@@ -35,6 +35,10 @@ router = APIRouter(prefix="/api/ontology", tags=["ontology"])
 _MAX_FETCH_BYTES = 20 * 1024 * 1024  # 20 MB
 _MAX_ANALYSIS_NODES = 5_000   # cap for health/suggest/shacl node scans to avoid OOM
 _MAX_ENTITIES_PER_SIDE = 500  # per-ontology cap for the O(n²) pairwise suggestion loop
+_GRAPH_TOO_LARGE_DETAIL = (
+    "Ontology editor graph exceeds the maximum size "
+    f"({_MAX_ANALYSIS_NODES} nodes or edges)."
+)
 
 
 class GraphTruncationError(Exception):
@@ -72,6 +76,20 @@ _ONTOLOGY_TYPES = frozenset({
 }) | _SCHEME_TYPES
 
 _SEARCHABLE_TYPES = _CLASS_TYPES | _PROPERTY_TYPES | _INDIVIDUAL_TYPES | _CONCEPT_TYPES | _SCHEME_TYPES
+_SCHEMA_NODE_TYPES = _CLASS_TYPES | _PROPERTY_TYPES | _CONCEPT_TYPES | _ONTOLOGY_TYPES
+_STRUCTURE_EDGE_TYPES = frozenset({
+    "rdf:type",
+    "rdfs:subClassOf",
+    "rdfs:domain",
+    "rdfs:range",
+    "owl:disjointWith",
+    "owl:equivalentClass",
+    "owl:equivalentProperty",
+    "owl:inverseOf",
+    "skos:broader",
+    "skos:narrower",
+    "skos:related",
+})
 
 _URI_PREFIX_MAP = {
     "http://www.w3.org/2002/07/owl#": "owl:",
@@ -1821,6 +1839,64 @@ async def search_entities(
     return results
 
 
+def _known_ontology_uris(
+    session: GraphSession, registry: Dict[str, OntologyEntry]
+) -> set[str]:
+    known = set(registry)
+    for node_type in _ONTOLOGY_TYPES:
+        for node in session.iter_nodes(node_type=node_type):
+            node_id = str(node.get("id", ""))
+            if node_id:
+                known.add(node_id)
+    return known
+
+
+def _collect_core_nodes(
+    session: GraphSession, uri: str, known_ontology_uris: set[str]
+) -> Dict[str, Dict[str, Any]]:
+    """Stream schema nodes, keeping only the ones this ontology owns.
+
+    Filtering as each node arrives makes _MAX_ANALYSIS_NODES bound the work and
+    not merely the response: foreign nodes are discarded instead of materialized,
+    and the scan stops once the owned ones pass the cap. The ownership filter has
+    to stay ahead of that check — thousands of *other* ontologies' nodes must
+    never make this one too large to open. Requesting pages instead would bound
+    nothing: paginate_nodes normalizes the whole matching set on every call.
+    """
+    core_nodes_by_id: Dict[str, Dict[str, Any]] = {}
+    for node_type in _SCHEMA_NODE_TYPES:
+        for node in session.iter_nodes(node_type=node_type):
+            node_id = str(node.get("id", ""))
+            if not node_id or not _node_belongs_to_ontology(
+                node, uri, known_ontology_uris
+            ):
+                continue
+            core_nodes_by_id[node_id] = node
+            if len(core_nodes_by_id) > _MAX_ANALYSIS_NODES:
+                raise GraphTruncationError(_GRAPH_TOO_LARGE_DETAIL)
+    return core_nodes_by_id
+
+
+def _select_structure_edges(
+    session: GraphSession, core_node_ids: set[str]
+) -> List[Dict[str, Any]]:
+    """Stream structural edges, keeping only those leaving a core node.
+
+    The requested ontology may reference outward (e.g. rdfs:range to an external
+    vocabulary), but an unrelated ontology's property pointing at a core class
+    must not leak inward.
+    """
+    selected_edges: List[Dict[str, Any]] = []
+    for edge_type in _STRUCTURE_EDGE_TYPES:
+        for edge in session.iter_edges(edge_type=edge_type):
+            if str(edge.get("source", "")) not in core_node_ids:
+                continue
+            selected_edges.append(edge)
+            if len(selected_edges) > _MAX_ANALYSIS_NODES:
+                raise GraphTruncationError(_GRAPH_TOO_LARGE_DETAIL)
+    return selected_edges
+
+
 @router.get("/graph", response_model=OntologyGraphResponse)
 async def get_ontology_graph(
     request: Request,
@@ -1828,88 +1904,39 @@ async def get_ontology_graph(
     session: GraphSession = Depends(get_session),
 ):
     """Return the editable schema subgraph for one registered ontology."""
-    registry = _get_registry(request)
-    ontology_nodes: List[Dict[str, Any]] = []
-    for node_type in _ONTOLOGY_TYPES:
-        nodes, _ = await asyncio.to_thread(
-            session.get_nodes, node_type=node_type, skip=0, limit=2**63 - 1
-        )
-        ontology_nodes.extend(nodes)
-    known_ontology_uris = set(registry) | {
-        str(node.get("id", "")) for node in ontology_nodes if node.get("id")
-    }
+    known_ontology_uris = await asyncio.to_thread(
+        _known_ontology_uris, session, _get_registry(request)
+    )
     if uri not in known_ontology_uris:
         raise HTTPException(status_code=404, detail="Ontology not found in registry.")
 
-    schema_types = _CLASS_TYPES | _PROPERTY_TYPES | _CONCEPT_TYPES | _ONTOLOGY_TYPES
-    candidates_by_id: Dict[str, Dict[str, Any]] = {}
-    for node_type in schema_types:
-        nodes, _ = await asyncio.to_thread(
-            session.get_nodes, node_type=node_type, skip=0, limit=2**63 - 1
+    try:
+        core_nodes_by_id = await asyncio.to_thread(
+            _collect_core_nodes, session, uri, known_ontology_uris
         )
-        candidates_by_id.update(
-            (str(node.get("id", "")), node) for node in nodes if node.get("id")
+        if not core_nodes_by_id:
+            raise HTTPException(status_code=404, detail="Ontology graph not found.")
+        core_node_ids = set(core_nodes_by_id)
+        selected_edges = await asyncio.to_thread(
+            _select_structure_edges, session, core_node_ids
         )
+    except GraphTruncationError as exc:
+        raise HTTPException(status_code=413, detail=str(exc)) from exc
 
-    core_node_ids = {
-        str(node.get("id", ""))
-        for node in candidates_by_id.values()
-        if _node_belongs_to_ontology(node, uri, known_ontology_uris)
-    }
-    if not core_node_ids:
-        raise HTTPException(status_code=404, detail="Ontology graph not found.")
+    # Invariant: the helpers raise the moment their accumulation passes
+    # _MAX_ANALYSIS_NODES, so core_nodes_by_id and selected_edges are both
+    # within the cap here; a post-filter re-check would be unreachable.
+    external_node_ids = {
+        node_id
+        for edge in selected_edges
+        for node_id in (str(edge.get("source", "")), str(edge.get("target", "")))
+    } - core_node_ids
+    external_nodes = await asyncio.gather(
+        *(asyncio.to_thread(session.get_node, node_id) for node_id in external_node_ids)
+    )
 
-    structure_edge_types = {
-        "rdf:type",
-        "rdfs:subClassOf",
-        "rdfs:domain",
-        "rdfs:range",
-        "owl:disjointWith",
-        "owl:equivalentClass",
-        "owl:equivalentProperty",
-        "owl:inverseOf",
-        "skos:broader",
-        "skos:narrower",
-        "skos:related",
-    }
-    selected_edges: List[Dict[str, Any]] = []
-    for edge_type in structure_edge_types:
-        edges, _ = await asyncio.to_thread(
-            session.get_edges,
-            edge_type=edge_type,
-            skip=0,
-            limit=2**63 - 1,
-        )
-        # Keep only edges whose source is a core node: the requested ontology
-        # may reference outward (e.g. rdfs:range to an external vocabulary),
-        # but an unrelated ontology's property pointing at a core class must
-        # not leak inward.
-        selected_edges.extend(
-            edge for edge in edges
-            if str(edge.get("source", "")) in core_node_ids
-        )
-    if (
-        len(core_node_ids) > _MAX_ANALYSIS_NODES
-        or len(selected_edges) > _MAX_ANALYSIS_NODES
-    ):
-        raise HTTPException(
-            status_code=413,
-            detail=(
-                "Ontology editor graph exceeds the maximum size "
-                f"({_MAX_ANALYSIS_NODES} nodes or edges)."
-            ),
-        )
-
-    selected_node_ids = set(core_node_ids)
-    for edge in selected_edges:
-        selected_node_ids.add(str(edge.get("source", "")))
-        selected_node_ids.add(str(edge.get("target", "")))
-
-    selected_nodes = [candidates_by_id[node_id] for node_id in core_node_ids]
-    for node_id in selected_node_ids - core_node_ids:
-        external = await asyncio.to_thread(session.get_node, node_id)
-        if external is not None:
-            selected_nodes.append(external)
+    selected_nodes = list(core_nodes_by_id.values())
+    selected_nodes.extend(node for node in external_nodes if node is not None)
     selected_nodes.sort(key=lambda node: str(node.get("id", "")))
     selected_edges.sort(
         key=lambda edge: (

semantica/explorer/session.py

@@ -9,7 +9,7 @@ import threading
 import time
 import uuid
 from datetime import UTC, datetime
-from typing import Any, Dict, Iterable, List, Optional
+from typing import Any, Dict, Iterable, Iterator, List, Optional
 
 from ..context.context_graph import ContextGraph, _resolve_edge_identity
 from .search_index import GraphSearchIndex
@@ -375,6 +375,52 @@ class GraphSession:
         )
         return page, total
 
+    def iter_nodes(self, node_type: Optional[str] = None) -> Iterator[Dict[str, Any]]:
+        """Yield matching nodes one at a time, in the same order as ``paginate_nodes``.
+
+        ``paginate_nodes`` normalizes and holds the entire matching set before it
+        slices out a page, so a caller that filters the result down itself cannot
+        bound its cost by asking for smaller pages — it would re-pay that full
+        cost per page. Streaming lets such a caller retain only what it selects
+        and stop scanning as soon as it has enough.
+
+        Only the id list is snapshotted under the lock; nodes are read one at a
+        time, so a concurrent mutation can be observed mid-iteration and ids that
+        disappear are skipped. ``paginate_nodes`` is the atomic alternative.
+        """
+        with self._lock:
+            source_ids = (
+                self.graph.node_type_index.get(node_type, set())
+                if node_type
+                else self.graph.nodes.keys()
+            )
+            node_ids = sorted(
+                (node_id for node_id in source_ids if node_id is not None),
+                key=lambda value: str(value),
+            )
+        for node_id in node_ids:
+            with self._lock:
+                raw = self.graph.find_node(node_id)
+            if raw is None:
+                continue
+            yield self.normalize_node(raw)
+
+    def iter_edges(self, edge_type: Optional[str] = None) -> Iterator[Dict[str, Any]]:
+        """Yield matching edges one at a time, in raw graph order.
+
+        Same rationale as ``iter_nodes``. Edge normalization derives an identity
+        hash per edge, which ``paginate_edges`` pays for every matching edge (and
+        then sorts) before paging; a filtering caller only needs it for the edges
+        it keeps. Callers that need a stable order sort the subset they select.
+        """
+        with self._lock:
+            raw_edges = self.graph.find_edges(edge_type=edge_type)
+        for edge in raw_edges:
+            normalized = self.normalize_edge(edge)
+            if not normalized["source"] or not normalized["target"]:
+                continue
+            yield normalized
+
     def get_raw_counts(self) -> tuple[int, int]:
         """O(1) node/edge counts from the raw collections, with no per-item
         normalization.

tests/context/test_erasure_coordinator.py

@@ -399,6 +399,121 @@ class TestReceipt(unittest.TestCase):
 
         self.assertEqual(receipt.stores["graph"]["status"], STATUS_ERASED)
 
+    def test_to_dict_deep_copies_nested_store_results(self):
+        """A nested dict in a store result is not shared with the live receipt.
+
+        ``backend_result`` from a vector backend is a dict of its own, so a
+        shallow per-store copy leaves it referenced by both the payload and the
+        receipt -- sanitizing the payload for a user-facing response would
+        silently corrupt the audit record.
+        """
+        receipt = ErasureReceipt(
+            entity_id="customer-4471",
+            stores={
+                "vectors": {
+                    "status": STATUS_ERASED,
+                    "backend": "qdrant",
+                    "backend_result": {"status": "completed"},
+                },
+            },
+        )
+
+        payload = receipt.to_dict()
+        payload["stores"]["vectors"]["backend_result"]["status"] = "redacted"
+
+        self.assertEqual(
+            receipt.stores["vectors"]["backend_result"]["status"], "completed"
+        )
+
+    def test_to_dict_store_results_are_distinct_objects(self):
+        """The per-store dict and any nested dict in the payload must not be
+        the same objects as the ones in the live receipt.
+
+        A mutation test proves *isolation* only when the copy actually
+        happened; identity checks prove *that* a copy was made.
+        """
+        receipt = ErasureReceipt(
+            entity_id="customer-4471",
+            stores={
+                "vectors": {
+                    "status": STATUS_ERASED,
+                    "backend": "qdrant",
+                    # Realistic Qdrant-shaped backend_result with rendered enum
+                    "backend_result": {"status": "UpdateStatus.COMPLETED", "points": 1},
+                    "vector_ids": 2,
+                    "via": "delete_vectors",
+                },
+                "graph": {"status": STATUS_ERASED, "nodes": 1, "edges": 3},
+            },
+        )
+
+        payload = receipt.to_dict()
+
+        # The stores container itself is a new dict.
+        self.assertIsNot(payload["stores"], receipt.stores)
+
+        # Each per-store result dict is a new object.
+        self.assertIsNot(
+            payload["stores"]["vectors"], receipt.stores["vectors"]
+        )
+        self.assertIsNot(
+            payload["stores"]["graph"], receipt.stores["graph"]
+        )
+
+        # The nested backend_result dict is also a new object.
+        self.assertIsNot(
+            payload["stores"]["vectors"]["backend_result"],
+            receipt.stores["vectors"]["backend_result"],
+        )
+
+        # Values are equal (correct copy), not just distinct references.
+        self.assertEqual(
+            payload["stores"]["vectors"]["backend_result"],
+            {"status": "UpdateStatus.COMPLETED", "points": 1},
+        )
+        self.assertEqual(payload["stores"]["graph"], {"status": STATUS_ERASED, "nodes": 1, "edges": 3})
+
+    def test_to_dict_isolation_across_multiple_stores(self):
+        """Mutations to any store in the payload must not affect any other
+        store in either the payload or the live receipt.
+
+        This catches a hypothetical implementation that shares a single deep
+        copy across all stores rather than copying each independently.
+        """
+        receipt = ErasureReceipt(
+            entity_id="e1",
+            stores={
+                "vectors": {
+                    "status": STATUS_UNSUPPORTED,
+                    "backend": "faiss",
+                    "vector_ids": 1,
+                    "detail": "backend exposes no delete()",
+                },
+                "memory": {"status": STATUS_ERASED, "items": 4},
+                "graph": {
+                    "status": STATUS_ERASED,
+                    "nodes": 1,
+                    "edges": 2,
+                },
+            },
+        )
+
+        payload = receipt.to_dict()
+
+        # Mutate every store in the payload.
+        payload["stores"]["vectors"]["status"] = "tampered"
+        payload["stores"]["memory"]["items"] = 0
+        payload["stores"]["graph"]["nodes"] = 99
+
+        # None of the live receipt's stores are affected.
+        self.assertEqual(receipt.stores["vectors"]["status"], STATUS_UNSUPPORTED)
+        self.assertEqual(receipt.stores["memory"]["items"], 4)
+        self.assertEqual(receipt.stores["graph"]["nodes"], 1)
+
+        # The other stores in the payload are also unaffected (no aliasing).
+        self.assertEqual(payload["stores"]["memory"]["items"], 0)   # our mutation
+        self.assertEqual(receipt.stores["memory"]["items"], 4)       # unchanged
+
     def test_receipt_and_tombstone_agree_on_when_the_erasure_happened(self):
         graph = _graph()
         receipt = ErasureCoordinator(graph=graph).erase_entity(

tests/explorer/test_ontology_subissue3.py

@@ -13,6 +13,7 @@ pytest.importorskip("fastapi")
 
 from semantica.explorer.app import create_app  # noqa: E402
 from semantica.explorer.routes.ontology import (  # noqa: E402
+    _MAX_ANALYSIS_NODES,
     OntologyEntry,
     _convert_ontology_to_graph,
     _node_belongs_to_ontology,
@@ -310,6 +311,66 @@ def test_ontology_graph_ignores_unrelated_data_when_enforcing_size_limit(client)
     }
 
 
+def test_ontology_graph_rejects_oversized_core_and_stops_scanning(client, monkeypatch):
+    graph = client.app.state.session.graph
+    for index in range(5_001):
+        graph.add_node(
+            f"http://example.org/onto-a#Bulk{index:05d}",
+            node_type="owl:Class",
+            content="Bulk",
+            scheme_uri="http://example.org/onto-a",
+        )
+    for index in range(3_000):
+        graph.add_node(
+            f"urn:unrelated:{index}",
+            node_type="owl:Class",
+            content="Unrelated",
+            scheme_uri="http://example.org/onto-b",
+        )
+
+    streamed = 0
+    original_iter_nodes = GraphSession.iter_nodes
+
+    def counting_iter_nodes(self, node_type=None):
+        nonlocal streamed
+        for node in original_iter_nodes(self, node_type=node_type):
+            streamed += 1
+            yield node
+
+    monkeypatch.setattr(GraphSession, "iter_nodes", counting_iter_nodes)
+
+    response = client.get(
+        "/api/ontology/graph",
+        params={"uri": "http://example.org/onto-a"},
+    )
+
+    assert response.status_code == 413
+    assert str(_MAX_ANALYSIS_NODES) in response.json()["detail"]
+    # The graph holds 8,001 owl:Class nodes and onto-a's own sort first, so a
+    # scan that abandons at the cap sees far fewer than the whole type.
+    assert streamed < 6_000
+
+
+def test_ontology_graph_hydrates_external_edge_targets_in_sorted_order(client):
+    graph = client.app.state.session.graph
+    external = "http://external.example/Thing"
+    also_external = "http://external.example/Aardvark"
+    graph.add_node(external, node_type="owl:Class", content="External Thing")
+    graph.add_node(also_external, node_type="owl:Class", content="External Aardvark")
+    graph.add_edge("http://example.org/onto-a#name", external, edge_type="rdfs:range")
+    graph.add_edge("http://example.org/onto-a#name", also_external, edge_type="rdfs:range")
+
+    response = client.get(
+        "/api/ontology/graph",
+        params={"uri": "http://example.org/onto-a"},
+    )
+
+    assert response.status_code == 200
+    node_ids = [node["id"] for node in response.json()["nodes"]]
+    assert {external, also_external} <= set(node_ids)
+    assert node_ids == sorted(node_ids)
+
+
 def test_shacl_generate_and_shapes(client):
     response = client.post(
         "/api/ontology/shacl/generate",

tests/test_cli_commands.py

@@ -69,6 +69,19 @@ def _json_output(result) -> Any:
     return json.loads(result.output.strip())
 
 
+def _flatten(output: str) -> str:
+    """Undo Rich panel wrapping for substring assertions on error text.
+
+    Rich wraps long messages across multiple bordered lines (each with its
+    own leading/trailing "│"), so a naive whitespace join still leaves those
+    border characters between words that were split across lines. Strip the
+    box-drawing characters first, then collapse whitespace.
+    """
+    for ch in "┌┐└┘│─":
+        output = output.replace(ch, " ")
+    return " ".join(output.split())
+
+
 # ─── Global flags ─────────────────────────────────────────────────────────────
 
 
@@ -336,6 +349,79 @@ class TestIngest:
         assert captured["sources"] == "README.md"
         assert captured["kwargs"]["method"] == "file"
 
+    def test_configured_graph_backend_does_not_report_false_success(
+        self, runner, monkeypatch
+    ):
+        monkeypatch.setenv("GRAPH_STORE_DEFAULT_BACKEND", "neo4j")
+        monkeypatch.setattr(
+            "semantica.ingest.methods.ingest_file",
+            lambda sources, **kwargs: [{"path": sources}],
+        )
+
+        result = runner.invoke(cli_module.main, ["ingest", "README.md"])
+        output = _flatten(result.output)
+
+        assert result.exit_code != 0
+        assert "does not write to graph stores" in output
+        assert "Ingested:" not in output
+
+    def test_configured_graph_backend_error_does_not_recommend_broken_kg_build(
+        self, runner, monkeypatch
+    ):
+        # kg build also does not persist to a configured graph store
+        # (tracked separately as #1352), so the error must not send users to
+        # a command that will silently no-op the same way.
+        monkeypatch.setenv("GRAPH_STORE_DEFAULT_BACKEND", "neo4j")
+        monkeypatch.setattr(
+            "semantica.ingest.methods.ingest_file",
+            lambda sources, **kwargs: [{"path": sources}],
+        )
+
+        result = runner.invoke(cli_module.main, ["ingest", "README.md"])
+        output = _flatten(result.output)
+
+        assert result.exit_code != 0
+        assert "kg build" not in output
+
+    def test_output_flag_writes_real_content_and_bypasses_graph_error(
+        self, runner, monkeypatch, tmp_path
+    ):
+        monkeypatch.setenv("GRAPH_STORE_DEFAULT_BACKEND", "neo4j")
+        monkeypatch.setattr(
+            "semantica.ingest.methods.ingest_file",
+            lambda sources, **kwargs: [{"path": sources}],
+        )
+        out_path = tmp_path / "out.json"
+
+        result = runner.invoke(
+            cli_module.main, ["ingest", "README.md", "--output", str(out_path)]
+        )
+
+        _ok(result, substr="Wrote")
+        written = json.loads(out_path.read_text(encoding="utf-8"))
+        assert written == {"files": [{"path": "README.md"}]}
+
+    def test_store_and_output_are_not_forwarded_to_ingest_backend(
+        self, runner, monkeypatch, tmp_path
+    ):
+        captured = {}
+
+        def fake_ingest_file(sources, **kwargs):
+            captured["kwargs"] = kwargs
+            return [{"path": sources}]
+
+        monkeypatch.setattr("semantica.ingest.methods.ingest_file", fake_ingest_file)
+        out_path = tmp_path / "out.json"
+
+        result = runner.invoke(
+            cli_module.main,
+            ["ingest", "README.md", "--store", "neo4j", "--output", str(out_path)],
+        )
+
+        _ok(result)
+        assert "store" not in captured["kwargs"]
+        assert "output" not in captured["kwargs"]
+
     def test_import_error_is_clean(self, runner, monkeypatch):
         monkeypatch.setattr(cli_module, "__import__", _import_side_effect, raising=False)
         original_import = __import__
@@ -771,6 +857,156 @@ class TestReason:
         assert result.exit_code != 0
         assert "Traceback" not in result.output
 
+    def test_run_infers_from_graph_store_facts(self, runner, monkeypatch, tmp_path):
+        # reason run used to call Reasoner.run(), which does not exist
+        # (#1354); it must feed graph store facts + --rules into
+        # Reasoner.infer_facts().
+        pytest.importorskip("numpy", reason="semantica.reasoning needs numpy")
+        rules_file = tmp_path / "rules.yaml"
+        rules_file.write_text(
+            '- IF Person(?x) THEN Human(?x)\n'
+            '- IF MANAGES(?x, ?y) THEN Manager(?x)\n'
+            '- IF Employee(?x) THEN Staff(?x)\n',
+            encoding="utf-8")
+
+        class _FakeStore:
+            # Same dict schema as the real backends: nodes carry
+            # labels/properties, relationships carry start_node_id/end_node_id.
+            def get_nodes(self, limit=None):
+                return [{"id": 1, "labels": ["Person"],
+                         "properties": {"name": "Alice"}},
+                        {"id": 2, "labels": ["Person", "Employee"],
+                         "properties": {"name": "Bob"}}]
+
+            def get_relationships(self, limit=None):
+                return [{"id": 9, "type": "MANAGES",
+                         "start_node_id": 1, "end_node_id": 2}]
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _FakeStore())
+        result = runner.invoke(
+            cli_module.main,
+            ["--json", "reason", "run", "--rules", str(rules_file)],
+        )
+        _ok(result)
+        data = json.loads(result.output.strip())
+        # Person(Alice), Person(Bob), Employee(Bob), MANAGES(Alice, Bob)
+        assert data["facts"] == 4
+        assert "Human(Alice)" in data["inferred_facts"]
+        # Relationship endpoints resolve node ids to names.
+        assert "Manager(Alice)" in data["inferred_facts"]
+        # Secondary labels also become facts.
+        assert "Staff(Bob)" in data["inferred_facts"]
+        assert data["inferred_count"] == len(data["inferred_facts"])
+
+    def test_run_rejects_unwired_engine(self, runner):
+        result = runner.invoke(cli_module.main,
+                               ["reason", "run", "--engine", "sparql"])
+        assert result.exit_code != 0
+        assert "not wired" in result.output
+        assert "reason query" in result.output
+        assert "Traceback" not in result.output
+
+    def test_load_rule_definitions_formats(self, tmp_path):
+        yaml_list = tmp_path / "list.yaml"
+        yaml_list.write_text('- IF A(?x) THEN B(?x)\n- IF B(?x) THEN C(?x)\n',
+                             encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(yaml_list)) == [
+            "IF A(?x) THEN B(?x)", "IF B(?x) THEN C(?x)"]
+
+        yaml_map = tmp_path / "map.yaml"
+        yaml_map.write_text('rules:\n  - IF A(?x) THEN B(?x)\n', encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(yaml_map)) == [
+            "IF A(?x) THEN B(?x)"]
+
+        plain = tmp_path / "rules.dl"
+        plain.write_text('# comment\nIF A(?x) THEN B(?x)\n\n', encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(plain)) == [
+            "IF A(?x) THEN B(?x)"]
+
+    def test_run_empty_graph_returns_zero_facts(self, runner, monkeypatch):
+        """reason run with an empty graph store should not crash and report 0 facts."""
+        pytest.importorskip("numpy", reason="semantica.reasoning needs numpy")
+
+        class _EmptyStore:
+            def get_nodes(self, limit=None): return []
+            def get_relationships(self, limit=None): return []
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _EmptyStore())
+        result = runner.invoke(cli_module.main, ["--json", "reason", "run"])
+        _ok(result)
+        data = json.loads(result.output.strip())
+        assert data["facts"] == 0
+        assert data["inferred_count"] == 0
+        assert data["inferred_facts"] == []
+
+    def test_run_graph_store_error_surfaces_cleanly(self, runner, monkeypatch):
+        """A graph-store connectivity error must surface as a clean error, not a Traceback."""
+
+        def _bad_store(ctx):
+            raise RuntimeError("connection refused")
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", _bad_store)
+        result = runner.invoke(cli_module.main, ["reason", "run"])
+        assert result.exit_code != 0
+        assert "Traceback" not in result.output
+        assert "connection refused" in result.output
+
+    def test_run_no_rules_uses_empty_ruleset(self, runner, monkeypatch):
+        """reason run without --rules should still succeed (zero rules -> zero inferences)."""
+        pytest.importorskip("numpy", reason="semantica.reasoning needs numpy")
+
+        class _FakeStore:
+            def get_nodes(self, limit=None):
+                return [{"id": 1, "labels": ["Person"], "properties": {"name": "Alice"}}]
+
+            def get_relationships(self, limit=None):
+                return []
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _FakeStore())
+        result = runner.invoke(cli_module.main, ["--json", "reason", "run"])
+        _ok(result)
+        data = json.loads(result.output.strip())
+        assert data["facts"] == 1
+        assert data["inferred_count"] == 0
+
+    def test_load_rule_definitions_yaml_mapping_without_rules_key_raises(self, tmp_path):
+        """A YAML mapping with no 'rules' key must raise ClickException, not silently
+        pass the raw YAML lines as rules."""
+        bad = tmp_path / "bad.yaml"
+        bad.write_text("some_key: some_value\nother_key: other_value\n", encoding="utf-8")
+        import click as _click
+        with pytest.raises(_click.ClickException, match="no 'rules' key"):
+            cli_module._load_rule_definitions(str(bad))
+
+    def test_load_rule_definitions_empty_file_returns_empty_list(self, tmp_path):
+        empty = tmp_path / "empty.yaml"
+        empty.write_text("", encoding="utf-8")
+        assert cli_module._load_rule_definitions(str(empty)) == []
+
+    def test_load_rule_definitions_yaml_rules_null_falls_to_plaintext(self, tmp_path):
+        """rules: null is valid YAML with the key present; the null value is
+        not a list, so the function falls through to plain-text parsing and
+        returns the literal line (one no-op rule).  This documents the edge
+        case rather than asserting a specific useful behaviour."""
+        f = tmp_path / "null_rules.yaml"
+        f.write_text("rules: null\n", encoding="utf-8")
+        result = cli_module._load_rule_definitions(str(f))
+        # Plain-text fallback: the non-comment, non-blank line becomes a rule.
+        assert result == ["rules: null"]
+
+    def test_run_rejects_deductive_engine(self, runner, monkeypatch):
+        """Engines other than rete/forward-chain must be rejected with a helpful message."""
+
+        class _EmptyStore:
+            def get_nodes(self, limit=None): return []
+            def get_relationships(self, limit=None): return []
+
+        monkeypatch.setattr(cli_module, "_get_graph_store", lambda ctx: _EmptyStore())
+        result = runner.invoke(cli_module.main, ["reason", "run", "--engine", "deductive"])
+        assert result.exit_code != 0
+        assert "not wired" in result.output
+        assert "Traceback" not in result.output
+
     def test_explain_requires_conclusion(self, runner):
         result = runner.invoke(cli_module.main, ["reason", "explain"])
         assert result.exit_code != 0
@@ -1350,6 +1586,65 @@ class TestStore:
         result = runner.invoke(cli_module.main, ["store", "connect", "--backend", "neo4j"])
         _ok(result)
 
+    def test_connect_dispatches_through_graph_store(self, runner, monkeypatch):
+        # store connect used to call get_graph_store_method(backend) — the
+        # method registry, which needs (task, method_name) — so it raised a
+        # TypeError before any connection attempt (#1354).
+        calls = {}
+
+        class _FakeGraphStore:
+            def __init__(self, backend=None, **cfg):
+                calls["backend"] = backend
+                calls["cfg"] = cfg
+
+            def connect(self):
+                calls["connected"] = True
+                return True
+
+        import semantica.graph_store as gs_mod
+        monkeypatch.setattr(gs_mod, "GraphStore", _FakeGraphStore)
+        result = runner.invoke(cli_module.main, [
+            "store", "connect", "--backend", "neo4j",
+            "--uri", "bolt://example:7687", "--json"])
+        _ok(result)
+        data = _json_output(result)
+        assert data == {"backend": "neo4j", "connected": True}
+        assert calls["backend"] == "neo4j"
+        assert calls["cfg"].get("uri") == "bolt://example:7687"
+        assert calls.get("connected") is True
+
+    def test_connect_invalid_backend_reports_error_not_dispatch_error(self, runner):
+        """An unknown backend name must produce a meaningful backend error, not a
+        Python TypeError from the old get_graph_store_method() dispatch (#1354)."""
+        result = runner.invoke(cli_module.main,
+                               ["store", "connect", "--backend", "does-not-exist"])
+        # Exit 0 because store_connect always catches and reports errors gracefully.
+        _ok(result)
+        # The output must mention the backend, not a Python internal error.
+        assert "does-not-exist" in result.output
+        assert "TypeError" not in result.output
+        assert "Traceback" not in result.output
+
+    def test_connect_backend_error_surfaces_in_json(self, runner, monkeypatch):
+        """A connect() failure must appear in JSON output as connected=False with an error field."""
+
+        class _FailingStore:
+            def __init__(self, backend=None, **cfg):
+                pass
+
+            def connect(self):
+                raise RuntimeError("auth failed")
+
+        import semantica.graph_store as gs_mod
+        monkeypatch.setattr(gs_mod, "GraphStore", _FailingStore)
+        result = runner.invoke(cli_module.main, [
+            "store", "connect", "--backend", "neo4j", "--json"])
+        _ok(result)
+        data = _json_output(result)
+        assert data["connected"] is False
+        assert "auth failed" in data.get("error", "")
+        assert data["backend"] == "neo4j"
+
     def test_migrate_dry_run(self, runner):
         result = runner.invoke(cli_module.main, ["store", "migrate",
                                       "--from", "faiss", "--to", "qdrant", "--dry-run"])
@@ -2071,6 +2366,160 @@ class TestDoctorEmbeddingHintsAndEnv:
         assert "hash fallback" in st["note"], "padded/caps env value must enable deep mode"
 
 
+class TestDoctorTableLayout:
+    """#1428 + Qodo review: doctor table must keep Check labels and Hint text
+    readable at a normal 80-column terminal.
+
+    These tests render the *human-readable* (non-JSON) doctor table into a
+    captured 80-column Rich console so they cover the actual column-width
+    arithmetic, not just the JSON data.
+
+    Two regressions are protected:
+
+    A. #1428 — Hint (and Note) columns must not collapse into unreadable
+       single-character fragments or be silently truncated with a layout '…'.
+       overflow="fold" on both columns ensures content wraps across lines while
+       remaining fully present.
+
+    B. Qodo — Long Check labels such as "Embeddings (sentence-transformers)"
+       must not be truncated/ellipsized.  Assigning ratio=1 to the Check column
+       (as the original PR did) caused Rich to squeeze it below its min_width
+       at narrow terminals, so the fix removes ratio from the fixed-size columns.
+    """
+
+    def _render_doctor_at_80(self, runner, monkeypatch):
+        """Return the plain-text (ANSI-stripped) doctor table rendered at 80 cols."""
+        import io
+        import re
+        from rich.console import Console
+
+        # Unset LLM-provider env vars so the warn rows (with hints) are always present.
+        for var in ("OPENAI_API_KEY", "ANTHROPIC_API_KEY", "GROQ_API_KEY"):
+            monkeypatch.delenv(var, raising=False)
+
+        buf = io.StringIO()
+        narrow_console = Console(
+            file=buf, width=80, highlight=False, force_terminal=True, no_color=True
+        )
+        monkeypatch.setattr(cli_module, "console", narrow_console)
+
+        result = runner.invoke(cli_module.main, ["doctor"])
+        assert result.exit_code == 0, f"doctor exited non-zero: {result.output!r}"
+
+        return re.sub(r"\x1b\[[0-9;]*m", "", buf.getvalue())
+
+    def _hint_column_parts(self, output: str) -> "list[str]":
+        """Extract non-blank Hint-column segments from each rendered line.
+
+        Locates the Hint column start from the header row and slices that
+        suffix from every subsequent line, so the test is insensitive to the
+        exact widths of the other columns.
+        """
+        lines = output.splitlines()
+        # Line 0 is blank (console.print() blank line before table).
+        hdr = next((l for l in lines if "Hint" in l and "Check" in l), None)
+        assert hdr is not None, "Could not find table header in doctor output"
+        hint_start = hdr.index("Hint")
+
+        parts = []
+        for line in lines:
+            if len(line) > hint_start:
+                seg = line[hint_start:].rstrip()
+                if seg and seg != "Hint" and not set(seg).issubset({"─", " "}):
+                    parts.append(seg)
+        return parts
+
+    # ── B: Qodo regression ────────────────────────────────────────────────────
+
+    def test_long_check_label_not_truncated_at_80_cols(self, runner, monkeypatch):
+        """'Embeddings (sentence-transformers)' must appear verbatim at 80 cols.
+
+        Before the fix, ratio=1 on the Check column let Rich squeeze it below
+        its min_width, turning the label into 'Embedd…' or similar.
+        """
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        assert "Embeddings (sentence-transformers)" in output, (
+            "Check label 'Embeddings (sentence-transformers)' was truncated in "
+            "the 80-column doctor table — the ratio= constraint on the Check "
+            "column must be removed so min_width=34 is always honoured."
+        )
+
+    def test_all_check_labels_not_truncated_at_80_cols(self, runner, monkeypatch):
+        """Every standard Check label must appear verbatim at 80 cols."""
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        for label in (
+            "Python",
+            "semantica",
+            "rich",
+            "Graph store",
+            "Vector store",
+            "Embeddings (sentence-transformers)",
+            "Embeddings (fastembed)",
+            "OpenAI",
+            "Anthropic",
+            "Groq",
+            "Config file",
+            "Log directory",
+        ):
+            assert label in output, (
+                f"Check label {label!r} was truncated or missing in the "
+                "80-column doctor table."
+            )
+
+    # ── A: #1428 regression ───────────────────────────────────────────────────
+
+    def test_hint_content_fully_present_at_80_cols(self, runner, monkeypatch):
+        """The LLM-provider hints must be fully present (folded, not ellipsized).
+
+        With overflow='fold' the full hint text wraps across lines; no
+        characters are discarded.  Joining the Hint-column segments (stripping
+        whitespace) must reconstruct each complete hint string.
+        """
+        import re
+
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        parts = self._hint_column_parts(output)
+        hint_joined = re.sub(r"\s+", "", "".join(parts))
+
+        # Each LLM-provider hint must be fully recoverable from the folded lines.
+        for expected in (
+            "exportOPENAI_API_KEY=\u2026",       # export OPENAI_API_KEY=…
+            "exportANTHROPIC_API_KEY=\u2026",    # export ANTHROPIC_API_KEY=…
+            "exportGROQ_API_KEY=\u2026",         # export GROQ_API_KEY=…
+        ):
+            assert expected in hint_joined, (
+                f"Hint content {expected!r} is missing from the 80-column "
+                "doctor table — overflow='fold' must be set on the Hint column "
+                "so no content is silently discarded."
+            )
+
+    def test_hint_column_has_no_single_char_fragments_at_80_cols(
+        self, runner, monkeypatch
+    ):
+        """No Hint-column line must be a single alphabetic character.
+
+        The original #1428 bug produced outputs like:
+            export
+            O
+            P
+            E
+            N
+            A
+            I
+            ...
+        because Rich allocated the Hint column only 1–2 characters of content
+        width.  overflow='fold' on a properly-wide column eliminates this.
+        """
+        output = self._render_doctor_at_80(runner, monkeypatch)
+        parts = self._hint_column_parts(output)
+        single_char_alpha = [p for p in parts if len(p.strip()) == 1 and p.strip().isalpha()]
+        assert not single_char_alpha, (
+            f"Hint column contains single-character lines {single_char_alpha!r} "
+            "at 80 columns — the Hint column is too narrow; check min_width and "
+            "ratio settings."
+        )
+
+
 class TestEmbedGenerateOutput:
     """#994: `embed generate --output` must write files `embed index` can read."""