fix(vector): close local pgvector final review

harness/tests/l0/test_pgvector_store.py

@@ -245,8 +245,9 @@ def test_pgvector_writer_health_requires_sequence_usage(vector_configs):
     assert health.write_detail == (
         "vector schema incomplete: missing sequence privileges evidence, memory, schema_records"
     )
-    with pytest.raises(InsufficientPrivilege):
+    with pytest.raises(VectorWriteUnavailable) as error:
         store.upsert("memory", [_record("needs-sequence", [1.0, 0.0])])
+    assert isinstance(error.value.__cause__, InsufficientPrivilege)
 
     admin_engine = create_engine(
         f"postgresql+psycopg2://{admin_config.user}:{admin_config.password}"
@@ -262,6 +263,53 @@ def test_pgvector_writer_health_requires_sequence_usage(vector_configs):
     assert store.upsert("memory", [_record("has-sequence", [1.0, 0.0])]) == 1
 
 
+def test_pgvector_health_requires_schema_usage_for_reader_and_writer(vector_configs):
+    from tht.adapters.vector.pgvector import PgVectorStore
+
+    admin_config, reader_config, writer_config, _ = vector_configs
+    admin_engine = create_engine(
+        f"postgresql+psycopg2://{admin_config.user}:{admin_config.password}"
+        f"@{admin_config.host}:{admin_config.port}/{admin_config.database}"
+    )
+    store = PgVectorStore(reader_config, writer_config, expected_dimension=2)
+    with admin_engine.begin() as connection:
+        connection.exec_driver_sql(
+            f"REVOKE USAGE ON SCHEMA vectors FROM {reader_config.user}, {writer_config.user}"
+        )
+    health = store.health()
+    assert health.read_reachable is False and health.write_reachable is False
+    assert "missing schema usage" in health.read_detail
+    assert "missing schema usage" in health.write_detail
+    with pytest.raises(VectorReadUnavailable, match="Vector read operation unavailable"):
+        store.search(["memory"], [1.0, 0.0], limit=1)
+    with pytest.raises(VectorWriteUnavailable, match="Vector write operation unavailable"):
+        store.upsert("memory", [_record("blocked", [1.0, 0.0])])
+    with admin_engine.begin() as connection:
+        connection.exec_driver_sql(
+            f"GRANT USAGE ON SCHEMA vectors TO {reader_config.user}, {writer_config.user}"
+        )
+    admin_engine.dispose()
+    assert store.health().ok is True
+
+
+def test_pgvector_maps_unavailable_connections_without_leaking_password(vector_configs):
+    from tht.adapters.vector.pgvector import PgVectorStore
+
+    _, reader_config, writer_config, _ = vector_configs
+    password = "never-leak-this"
+    reader = reader_config.model_copy(update={"port": 1, "password": password})
+    writer = writer_config.model_copy(update={"port": 1, "password": password})
+    with pytest.raises(VectorReadUnavailable) as read_error:
+        PgVectorStore(reader, None).search(["memory"], [1.0, 0.0], limit=1)
+    with pytest.raises(VectorWriteUnavailable) as hash_error:
+        PgVectorStore(None, writer).existing_hashes("memory", ["memory"])
+    with pytest.raises(VectorWriteUnavailable) as write_error:
+        PgVectorStore(None, writer).upsert("memory", [_record("x", [1.0, 0.0])])
+    assert password not in str(read_error.value)
+    assert password not in str(hash_error.value)
+    assert password not in str(write_error.value)
+
+
 def test_pgvector_health_reports_dimension_and_each_connection(vector_configs):
     from tht.adapters.vector.pgvector import PgVectorStore
 

harness/tests/test_config_resources.py

@@ -1,4 +1,7 @@
+import pytest
+
 from tht.config import (
+    ConfigError,
     PgvectorDirectConfig,
     PostgresDwhConfig,
     ThothRestDwhConfig,
@@ -7,6 +10,44 @@ from tht.config import (
 )
 
 
+def test_direct_vector_passwords_load_from_file_references(monkeypatch, tmp_path):
+    reader = tmp_path / "reader"
+    writer = tmp_path / "writer"
+    reader.write_text("reader-secret")
+    writer.write_text("writer-secret")
+    monkeypatch.setenv("READER_FILE", str(reader))
+    monkeypatch.setenv("WRITER_FILE", str(writer))
+    workspace = tmp_path / "workspace.yaml"
+    workspace.write_text("""
+dwh:
+  type: postgres_direct
+  connection: {database: d, schema: public, user: u, password: p}
+vectors:
+  type: pgvector_direct
+  reader: {database: d, schema: vectors, user: r, password_file: '${READER_FILE}'}
+  writer: {database: d, schema: vectors, user: w, password_file: '${WRITER_FILE}'}
+""")
+    config = load_config(workspace)
+    assert config.vectors.reader.password == "reader-secret"
+    assert config.vectors.writer.password == "writer-secret"
+
+
+def test_direct_vector_secret_file_rejects_whitespace(tmp_path):
+    secret = tmp_path / "reader"
+    secret.write_text("bad secret")
+    workspace = tmp_path / "workspace.yaml"
+    workspace.write_text(f"""
+dwh:
+  type: postgres_direct
+  connection: {{database: d, schema: public, user: u, password: p}}
+vectors:
+  type: pgvector_direct
+  reader: {{database: d, schema: vectors, user: r, password_file: {secret}}}
+""")
+    with pytest.raises(ConfigError, match="secret file"):
+        load_config(workspace)
+
+
 def test_loads_discriminated_dwh_and_vector_resources(tmp_path):
     workspace = tmp_path / "workspace.yaml"
     workspace.write_text(

harness/tht/adapters/vector/pgvector.py

@@ -97,6 +97,13 @@ class PgVectorStore:
             try:
                 with raw.cursor() as cursor:
                     cursor.execute("SELECT 1")
+                    cursor.execute(
+                        "SELECT has_schema_privilege(current_user, %s, 'USAGE')",
+                        (self._schema,),
+                    )
+                    schema_usage = bool(cursor.fetchone()[0])
+                    if not schema_usage:
+                        return False, "vector schema incomplete: missing schema usage", set()
                     cursor.execute(
                         """SELECT c.relname, format_type(a.atttypid, a.atttypmod),
                                   has_table_privilege(current_user, c.oid, 'SELECT'),
@@ -244,8 +251,9 @@ class PgVectorStore:
         if kinds:
             _validate_known_kinds(kinds)
         hits: list[VectorHit] = []
-        raw = self._reader.raw_connection()
+        raw = None
         try:
+            raw = self._reader.raw_connection()
             with raw.cursor() as cursor:
                 for collection in collections:
                     table = _collection(self._schema, collection)
@@ -272,8 +280,13 @@ class PgVectorStore:
                     params.extend((_vector_literal(embedding), limit))
                     cursor.execute(query, params)
                     hits.extend(hit_from_metadata(row[1], row[0]) for row in cursor.fetchall())
+        except VectorStoreError:
+            raise
+        except Exception as exc:
+            raise VectorReadUnavailable("Vector read operation unavailable") from exc
         finally:
-            raw.close()
+            if raw is not None:
+                raw.close()
         return sorted(hits, key=lambda hit: (-hit.similarity, hit.id))[:limit]
 
     def _require_writer(self) -> Engine:
@@ -285,8 +298,9 @@ class PgVectorStore:
         engine = self._require_writer()
         table = _collection(self._schema, collection)
         _validate_collection_kinds(collection, kinds)
-        raw = engine.raw_connection()
+        raw = None
         try:
+            raw = engine.raw_connection()
             with raw.cursor() as cursor:
                 cursor.execute(
                     sql.SQL("SELECT record_key, content_hash FROM {} WHERE kind = ANY(%s)").format(
@@ -295,8 +309,13 @@ class PgVectorStore:
                     (kinds,),
                 )
                 return dict(cursor.fetchall())
+        except VectorStoreError:
+            raise
+        except Exception as exc:
+            raise VectorWriteUnavailable("Vector write operation unavailable") from exc
         finally:
-            raw.close()
+            if raw is not None:
+                raw.close()
 
     def upsert(self, collection: str, records: list[VectorWriteRecord]) -> int:
         engine = self._require_writer()
@@ -317,8 +336,9 @@ class PgVectorStore:
             "UPDATE {} SET kind = %s, content_hash = %s, metadata = %s::jsonb, "
             "embedding = %s::{}, indexed_at = pg_catalog.now() WHERE record_key = %s"
         ).format(table, _vector_type(self._schema))
-        raw = engine.raw_connection()
+        raw = None
         try:
+            raw = engine.raw_connection()
             with raw.cursor() as cursor:
                 for write_record in records:
                     record = write_record.record
@@ -348,11 +368,17 @@ class PgVectorStore:
                             ),
                         )
             raw.commit()
-        except Exception:
-            raw.rollback()
+        except VectorStoreError:
+            if raw is not None:
+                raw.rollback()
             raise
+        except Exception as exc:
+            if raw is not None:
+                raw.rollback()
+            raise VectorWriteUnavailable("Vector write operation unavailable") from exc
         finally:
-            raw.close()
+            if raw is not None:
+                raw.close()
         return len(records)
 
 

harness/tht/config.py

@@ -36,6 +36,26 @@ def _expand_env(value: Any) -> Any:
     return value
 
 
+def _resolve_secret_files(value: Any) -> Any:
+    if isinstance(value, dict):
+        resolved = {key: _resolve_secret_files(item) for key, item in value.items()}
+        if "password_file" in resolved:
+            if "password" in resolved:
+                raise ConfigError("password and password_file are mutually exclusive")
+            path = Path(resolved.pop("password_file"))
+            try:
+                secret = path.read_text()
+            except (OSError, UnicodeError) as exc:
+                raise ConfigError(f"Cannot read secret file: {path}") from exc
+            if not secret or any(char.isspace() for char in secret) or "\x00" in secret:
+                raise ConfigError(f"Invalid secret file: {path}")
+            resolved["password"] = secret
+        return resolved
+    if isinstance(value, list):
+        return [_resolve_secret_files(item) for item in value]
+    return value
+
+
 class DatabaseConfig(BaseModel):
     host: str = "localhost"
     port: int = 5432
@@ -254,7 +274,7 @@ def load_config(path: Path) -> Config:
     raw = yaml.safe_load(path.read_text())
     if not isinstance(raw, dict):
         raise ConfigError(f"Configurazione non valida (atteso un mapping YAML): {path}")
-    expanded = _expand_env(raw)
+    expanded = _resolve_secret_files(_expand_env(raw))
     translated, used_legacy = translate_legacy_config(expanded)
     _populate_legacy_views(translated)
     try: