diff --git a/docs/superpowers/plans/2026-07-11-adapter-foundations.md b/docs/superpowers/plans/2026-07-11-adapter-foundations.md index a32f475e..b11fc8ea 100644 --- a/docs/superpowers/plans/2026-07-11-adapter-foundations.md +++ b/docs/superpowers/plans/2026-07-11-adapter-foundations.md @@ -289,6 +289,11 @@ git commit -m "feat(config): add typed resource schema" - Produces: `build_dwh(cfg: Config) -> DwhAdapter` and `build_vector_store(cfg: Config, *, require_write: bool = False) -> VectorStore`. - Consumes: resource configs from Task 4 and wrappers from Tasks 2-3. +Correction: `DwhAdapter.distinct_values(table, column, *, limit)` requires an explicit +positive limit, and direct DWH construction injects `cfg.execution.statement_timeout_ms`. +Targeted vector writes consume the factory-returned `VectorStore` and pass +`VectorWriteRecord` objects to `upsert`. + - [ ] **Step 1: Write exact factory selection and missing-writer tests** ```python diff --git a/harness/tests/l2/test_memory_save_one_real.py b/harness/tests/l2/test_memory_save_one_real.py index bc98c283..c7e2aa74 100644 --- a/harness/tests/l2/test_memory_save_one_real.py +++ b/harness/tests/l2/test_memory_save_one_real.py @@ -39,7 +39,10 @@ def test_save_one_upserts_to_real_pgvector(l2_env): detail="ablazione", rationale="L2 self-test (idempotent)", question_context="ablazione 2025", tables=["fct_ricoveri"], concepts=[], ) - upserted = save_one_memory([record], decision_seq=999, writer=writer, embedder=embedder) + from tht.adapters.vector import ThothHttpVectorStore + + store = ThothHttpVectorStore(reader=writer, writer=writer) + upserted = save_one_memory([record], decision_seq=999, store=store, embedder=embedder) assert upserted >= 0 # idempotent: 0 on unchanged, >=1 on new/updated # read it back via the READER key (vector_rest, path /vector/v1/) diff --git a/harness/tests/test_adapter_factory.py b/harness/tests/test_adapter_factory.py index cf0f2f4b..52cf75ed 100644 --- a/harness/tests/test_adapter_factory.py +++ b/harness/tests/test_adapter_factory.py @@ -84,3 +84,9 @@ def test_factory_selects_direct_vector_reader(): assert isinstance(build_vector_store(config), LegacyDirectVectorStore) with pytest.raises(ConfigError, match="writer"): build_vector_store(config, require_write=True) + + +def test_factory_propagates_non_default_statement_timeout(): + config = _config(dwh_type="postgres_direct") + config.execution.statement_timeout_ms = 12_345 + assert build_dwh(config)._statement_timeout_ms == 12_345 diff --git a/harness/tests/test_dwh_adapters.py b/harness/tests/test_dwh_adapters.py index fdb88b62..985b9d4b 100644 --- a/harness/tests/test_dwh_adapters.py +++ b/harness/tests/test_dwh_adapters.py @@ -85,14 +85,17 @@ def test_postgres_sampling_delegates_to_paired_sampling_functions(monkeypatch): ) or ["A", "B"], ) + distinct_calls = [] monkeypatch.setattr( "tht.adapters.dwh.postgres.sampling.distinct_values", - lambda engine, schema, table, column: expected, + lambda engine, schema, table, column, *, max_values: distinct_calls.append(max_values) + or expected, ) assert adapter.sample_column("sales", "region", limit=2) == ["A", "B"] assert calls == [(adapter._engine, "dw", "sales", "region", 2)] - assert adapter.distinct_values("sales", "region") is expected + assert adapter.distinct_values("sales", "region", limit=17) is expected + assert distinct_calls == [17] def test_rest_sampling_delegates_and_translates_transport_errors(monkeypatch): @@ -104,10 +107,10 @@ def test_rest_sampling_delegates_and_translates_transport_errors(monkeypatch): ) monkeypatch.setattr( "tht.adapters.dwh.thoth_rest.sampling.distinct_values_rest", - lambda client, schema, table, column: expected, + lambda client, schema, table, column, *, max_values: expected, ) assert adapter.sample_column("sales", "region", limit=2) == ["A", "B"] - assert adapter.distinct_values("sales", "region") is expected + assert adapter.distinct_values("sales", "region", limit=17) is expected def fail(*args, **kwargs): raise RestError("transport failed") @@ -117,7 +120,7 @@ def test_rest_sampling_delegates_and_translates_transport_errors(monkeypatch): with pytest.raises(ExecutionError, match="transport failed"): adapter.sample_column("sales", "region", limit=2) with pytest.raises(ExecutionError, match="transport failed"): - adapter.distinct_values("sales", "region") + adapter.distinct_values("sales", "region", limit=17) def test_rest_distinct_values_reports_transport_truncation(): diff --git a/harness/tests/test_dwh_port_contract.py b/harness/tests/test_dwh_port_contract.py index 8899d608..55e7bac9 100644 --- a/harness/tests/test_dwh_port_contract.py +++ b/harness/tests/test_dwh_port_contract.py @@ -31,7 +31,7 @@ class FakeDwhAdapter: def sample_column(self, table: str, column: str, *, limit: int) -> list[object]: raise NotImplementedError - def distinct_values(self, table: str, column: str) -> DistinctValues: + def distinct_values(self, table: str, column: str, *, limit: int) -> DistinctValues: raise NotImplementedError diff --git a/harness/tests/test_memory_metadata.py b/harness/tests/test_memory_metadata.py index 673bb488..4e5ed79d 100644 --- a/harness/tests/test_memory_metadata.py +++ b/harness/tests/test_memory_metadata.py @@ -56,12 +56,12 @@ def test_save_one_memory_preserves_subject_through_upsert_row(): from tht.memory import save_one_memory writer = MagicMock() - writer.upsert_records.return_value = 1 + writer.upsert.return_value = 1 embedder = MagicMock() embedder.embed_documents.return_value = [[0.0] * 8] - save_one_memory([_record(decision_seq=1)], decision_seq=1, writer=writer, embedder=embedder) - row = writer.upsert_records.call_args[0][1][0] - md = row["metadata"] + save_one_memory([_record(decision_seq=1)], decision_seq=1, store=writer, embedder=embedder) + row = writer.upsert.call_args[0][1][0] + md = row.record.metadata assert md["subject"] == "dim_pazienti" assert md["detail"] == "promossa" assert md["rationale"] == "perche' serve" diff --git a/harness/tests/test_memory_save_one.py b/harness/tests/test_memory_save_one.py index e686cc6a..7a679668 100644 --- a/harness/tests/test_memory_save_one.py +++ b/harness/tests/test_memory_save_one.py @@ -8,11 +8,8 @@ pgvector as a one-row upsert. This test pins the pure core of that behavior: - writer.sync is NEVER called (that is the full-resync path) """ from datetime import datetime -from pathlib import Path from unittest.mock import MagicMock -import pytest - from tht.memory import MemoryRecord, memory_vector_record_for_decision, save_one_memory @@ -47,31 +44,31 @@ def test_returns_none_for_unknown_decision_seq(): def test_save_one_calls_upsert_with_single_row_never_sync(): records = [_record(seq=7)] writer = MagicMock() - writer.upsert_records.return_value = 1 + writer.upsert.return_value = 1 embedder = MagicMock() embedder.embed_documents.return_value = [[0.1] * 8] - upserted = save_one_memory(records, decision_seq=7, writer=writer, embedder=embedder) + upserted = save_one_memory(records, decision_seq=7, store=writer, embedder=embedder) assert upserted == 1 writer.sync.assert_not_called() # the whole point of D11: no full resync - writer.upsert_records.assert_called_once() - args = writer.upsert_records.call_args + writer.upsert.assert_called_once() + args = writer.upsert.call_args # table is memory, exactly one row assert args[0][0] == "memory" rows = args[0][1] assert len(rows) == 1 - assert rows[0]["record_key"] == "memory:mem-0007" - assert "embedding" in rows[0] + assert rows[0].record.id == "memory:mem-0007" + assert rows[0].embedding def test_save_one_no_record_for_seq_is_noop(): records = [_record(seq=7)] writer = MagicMock() embedder = MagicMock() - upserted = save_one_memory(records, decision_seq=42, writer=writer, embedder=embedder) + upserted = save_one_memory(records, decision_seq=42, store=writer, embedder=embedder) assert upserted == 0 - writer.upsert_records.assert_not_called() + writer.upsert.assert_not_called() writer.sync.assert_not_called() embedder.embed_documents.assert_not_called() @@ -81,9 +78,9 @@ def test_save_one_uses_writer_key_for_upsert(): Verified indirectly: save_one_memory takes the writer as its client argument.""" records = [_record(seq=7)] writer = MagicMock() - writer.upsert_records.return_value = 1 + writer.upsert.return_value = 1 embedder = MagicMock() embedder.embed_documents.return_value = [[0.0] * 4] - save_one_memory(records, decision_seq=7, writer=writer, embedder=embedder) + save_one_memory(records, decision_seq=7, store=writer, embedder=embedder) # one upsert call, single row, table=memory - assert writer.upsert_records.call_count == 1 + assert writer.upsert.call_count == 1 diff --git a/harness/tests/test_schema_introspect_guard.py b/harness/tests/test_schema_introspect_guard.py index 056e361f..80fe6f2a 100644 --- a/harness/tests/test_schema_introspect_guard.py +++ b/harness/tests/test_schema_introspect_guard.py @@ -4,6 +4,8 @@ from typer.testing import CliRunner from tht.cli import app from tht.mschema.models import ColumnPhysical, PhysicalSchema, TablePhysical +from tht.config import ExamplesConfig +from tht.cli.schema_cmd import _add_examples def _write_catalog(tmp_path): @@ -68,3 +70,23 @@ def test_render_without_catalog_guides_fallback(tmp_path): res = CliRunner().invoke(app, ["schema", "render", "-c", str(cfg)]) assert res.exit_code == 1 assert "Esegui prima" in res.output + + +def test_examples_skip_one_unreadable_column_and_continue(caplog): + physical = PhysicalSchema( + database="d", schema="s", introspected_at=datetime(2026, 1, 1), + tables={"t": TablePhysical(columns={ + "bad": ColumnPhysical(type="text"), "good": ColumnPhysical(type="text") + })}, + ) + + class Dwh: + def sample_column(self, table, column, *, limit): + if column == "bad": + raise RuntimeError("denied") + return ["kept"] + + _add_examples(Dwh(), physical, ExamplesConfig(max_per_column=3)) + assert physical.tables["t"].columns["bad"].examples == [] + assert physical.tables["t"].columns["good"].examples == ["kept"] + assert "Campionamento saltato" in caplog.text diff --git a/harness/tests/test_solved_question.py b/harness/tests/test_solved_question.py index 54c4fb62..91c9e5d0 100644 --- a/harness/tests/test_solved_question.py +++ b/harness/tests/test_solved_question.py @@ -43,18 +43,18 @@ def test_solved_kind_maps_to_memory_table(): def test_save_upserts_single_row_into_memory_table(): writer = MagicMock() writer.existing_hashes.return_value = {} - writer.upsert_records.return_value = 1 + writer.upsert.return_value = 1 embedder = MagicMock() embedder.embed_documents.return_value = [[0.1] * 8] - assert save_solved_question(_rec(), writer=writer, embedder=embedder) == 1 + assert save_solved_question(_rec(), store=writer, embedder=embedder) == 1 writer.sync.assert_not_called() - table, rows = writer.upsert_records.call_args[0] + table, rows = writer.upsert.call_args[0] assert table == "memory" assert len(rows) == 1 - assert rows[0]["record_key"] == "solved:s1" - assert rows[0]["metadata"]["kind"] == SOLVED_KIND - assert rows[0]["metadata"]["sql"].startswith("SELECT") + assert rows[0].record.id == "solved:s1" + assert rows[0].record.kind == SOLVED_KIND + assert rows[0].record.metadata["sql"].startswith("SELECT") def test_save_skips_when_question_and_sql_unchanged(): @@ -62,9 +62,9 @@ def test_save_skips_when_question_and_sql_unchanged(): writer = MagicMock() writer.existing_hashes.return_value = {r.id: _solved_hash(r)} embedder = MagicMock() - assert save_solved_question(r, writer=writer, embedder=embedder) == 0 + assert save_solved_question(r, store=writer, embedder=embedder) == 0 embedder.embed_documents.assert_not_called() - writer.upsert_records.assert_not_called() + writer.upsert.assert_not_called() def test_sql_change_alone_triggers_reupsert(): @@ -72,7 +72,7 @@ def test_sql_change_alone_triggers_reupsert(): new = _rec(sql="SELECT 1") # stessa domanda, SQL diverso writer = MagicMock() writer.existing_hashes.return_value = {old.id: _solved_hash(old)} - writer.upsert_records.return_value = 1 + writer.upsert.return_value = 1 embedder = MagicMock() embedder.embed_documents.return_value = [[0.0] * 4] - assert save_solved_question(new, writer=writer, embedder=embedder) == 1 + assert save_solved_question(new, store=writer, embedder=embedder) == 1 diff --git a/harness/tht/adapters/dwh/postgres.py b/harness/tht/adapters/dwh/postgres.py index 2041ba82..58356b6b 100644 --- a/harness/tht/adapters/dwh/postgres.py +++ b/harness/tht/adapters/dwh/postgres.py @@ -4,7 +4,7 @@ from tht.config import DatabaseConfig from sqlalchemy.exc import SQLAlchemyError from tht.db import execute, sampling -from tht.db.connection import make_engine, ping +from tht.db.connection import can_create_in_schema, make_engine, ping, writable_tables from tht.db.introspect import introspect from tht.execute import ExecResult, PlanSummary from tht.mschema.models import PhysicalSchema @@ -14,32 +14,39 @@ from tht.ports.dwh import DistinctValues, DwhCapabilities, DwhHealth class PostgresDwhAdapter: capabilities = DwhCapabilities() - def __init__(self, config: DatabaseConfig): + def __init__(self, config: DatabaseConfig, *, statement_timeout_ms: int = 30_000): self._config = config self._engine = make_engine(config) + self._statement_timeout_ms = statement_timeout_ms def health(self) -> DwhHealth: try: ping(self._engine) except SQLAlchemyError as exc: return DwhHealth(ok=False, detail=str(exc)) - return DwhHealth(ok=True) + writable = tuple(writable_tables(self._engine, self._config.db_schema)) + can_create = can_create_in_schema(self._engine, self._config.db_schema) + return DwhHealth(ok=True, database=self._config.database, schema=self._config.db_schema, + read_only=not writable and not can_create, + writable_tables=writable, can_create=can_create) def introspect(self) -> PhysicalSchema: return introspect(self._engine, self._config.database, self._config.db_schema) def run_query(self, sql: str, *, limit: int) -> ExecResult: - return execute.run_query(self._engine, sql, limit=limit) + return execute.run_query( + self._engine, sql, limit=limit, timeout_ms=self._statement_timeout_ms + ) def explain(self, sql: str) -> PlanSummary: - return execute.explain(self._engine, sql) + return execute.explain(self._engine, sql, timeout_ms=self._statement_timeout_ms) def sample_column(self, table: str, column: str, *, limit: int) -> list[object]: return sampling.sample_column( self._engine, self._config.db_schema, table, column, limit=limit ) - def distinct_values(self, table: str, column: str) -> DistinctValues: + def distinct_values(self, table: str, column: str, *, limit: int) -> DistinctValues: return sampling.distinct_values( - self._engine, self._config.db_schema, table, column + self._engine, self._config.db_schema, table, column, max_values=limit ) diff --git a/harness/tht/adapters/dwh/thoth_rest.py b/harness/tht/adapters/dwh/thoth_rest.py index beac1781..e1f1c321 100644 --- a/harness/tht/adapters/dwh/thoth_rest.py +++ b/harness/tht/adapters/dwh/thoth_rest.py @@ -22,8 +22,10 @@ class ThothRestDwhAdapter: result = self._client.ping() except RestError as exc: return DwhHealth(ok=False, detail=str(exc)) - ok = bool(result.get("db_connected", True)) - return DwhHealth(ok=ok, detail=None if ok else str(result)) + ok = bool(result.get("db_connected") and result.get("schema_accessible")) + return DwhHealth(ok=ok, detail=None if ok else str(result), + database=self._database.database, schema=self._database.db_schema, + endpoint=self._client.cfg.base_url, read_only=True) def introspect(self) -> PhysicalSchema: return introspect_rest( @@ -44,10 +46,10 @@ class ThothRestDwhAdapter: except RestError as exc: raise ExecutionError(str(exc)) from exc - def distinct_values(self, table: str, column: str) -> DistinctValues: + def distinct_values(self, table: str, column: str, *, limit: int) -> DistinctValues: try: return sampling.distinct_values_rest( - self._client, self._database.db_schema, table, column + self._client, self._database.db_schema, table, column, max_values=limit ) except RestError as exc: raise ExecutionError(str(exc)) from exc diff --git a/harness/tht/adapters/factory.py b/harness/tht/adapters/factory.py index 61fffa60..a30fdc17 100644 --- a/harness/tht/adapters/factory.py +++ b/harness/tht/adapters/factory.py @@ -14,7 +14,10 @@ def build_dwh(cfg: Config) -> DwhAdapter: resource = cfg.dwh match resource.type: case "postgres_direct": - return PostgresDwhAdapter(resource.connection) + return PostgresDwhAdapter( + resource.connection, + statement_timeout_ms=cfg.execution.statement_timeout_ms, + ) case "thoth_rest": return ThothRestDwhAdapter(resource.database, resource.endpoint) case other: # pragma: no cover - Pydantic's discriminator rejects this first. diff --git a/harness/tht/cli/db_cmd.py b/harness/tht/cli/db_cmd.py index 28ed309a..a1d06b65 100644 --- a/harness/tht/cli/db_cmd.py +++ b/harness/tht/cli/db_cmd.py @@ -1,43 +1,14 @@ from pathlib import Path import typer -from sqlalchemy.exc import OperationalError - -from tht.adapters.dwh import ThothRestDwhAdapter from tht.adapters.factory import build_dwh from tht.cli.config_cmd import CONFIG_OPT from tht.config import ConfigError, load_config -from tht.db.connection import can_create_in_schema, make_engine, ping, writable_tables from tht.db.fetch_ca import CaFetchError, describe_pem, fetch_chain_pem, parse_host_port db_app = typer.Typer(help="Operazioni sul database target") -def _ping_rest(adapter: ThothRestDwhAdapter, cfg, schema: str) -> None: - """Health check via REST. Il read-only รจ garantito strutturalmente dall'API - (ammette solo SELECT/WITH): non serve il controllo dei privilegi di scrittura.""" - from tht.rest.client import RestError - - try: - info = adapter._client.ping() - except RestError as e: - typer.secho(f"ERRORE di connessione: {e}", fg=typer.colors.RED, err=True) - raise typer.Exit(code=1) - if not info.get("db_connected") or not info.get("schema_accessible"): - typer.secho( - f"ERRORE: DWH non accessibile via REST (risposta: {info}).", - fg=typer.colors.RED, - err=True, - ) - raise typer.Exit(code=1) - typer.secho( - f"OK: connesso via REST a {cfg.rest.base_url} (schema {schema})", fg=typer.colors.GREEN - ) - typer.secho( - "OK: accesso read-only garantito dall'API (solo SELECT/WITH).", fg=typer.colors.GREEN - ) - - @db_app.command("ping") def ping_cmd(config: Path = CONFIG_OPT) -> None: """Testa la connessione e verifica che l'utente sia effettivamente read-only.""" @@ -46,29 +17,26 @@ def ping_cmd(config: Path = CONFIG_OPT) -> None: except ConfigError as e: typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) - schema = cfg.database.db_schema - adapter = build_dwh(cfg) - if isinstance(adapter, ThothRestDwhAdapter): - _ping_rest(adapter, cfg, schema) - return - engine = make_engine(cfg.database) - try: - ping(engine) - except OperationalError as e: - typer.secho(f"ERRORE di connessione: {e.orig}", fg=typer.colors.RED, err=True) + health = build_dwh(cfg).health() + if not health.ok: + typer.secho(f"ERRORE di connessione: {health.detail}", fg=typer.colors.RED, err=True) raise typer.Exit(code=1) - typer.secho(f"OK: connesso a {cfg.database.database} (schema {schema})", fg=typer.colors.GREEN) - - writable = writable_tables(engine, schema) - can_create = can_create_in_schema(engine, schema) - if writable or can_create: + if health.endpoint: + typer.secho(f"OK: connesso via REST a {health.endpoint} (schema {health.schema})", + fg=typer.colors.GREEN) + typer.secho("OK: accesso read-only garantito dall'API (solo SELECT/WITH).", + fg=typer.colors.GREEN) + return + typer.secho(f"OK: connesso a {health.database} (schema {health.schema})", + fg=typer.colors.GREEN) + if not health.read_only: typer.secho( f"ERRORE: l'utente '{cfg.database.user}' NON e' read-only.", fg=typer.colors.RED, err=True ) - if writable: - typer.echo(f" Tabelle scrivibili: {', '.join(writable[:10])}", err=True) - if can_create: - typer.echo(f" L'utente puo' creare oggetti nello schema {schema}.", err=True) + if health.writable_tables: + typer.echo(f" Tabelle scrivibili: {', '.join(health.writable_tables[:10])}", err=True) + if health.can_create: + typer.echo(f" L'utente puo' creare oggetti nello schema {health.schema}.", err=True) typer.echo(" Crea un ruolo read-only con scripts/create_readonly_role.sql.", err=True) raise typer.Exit(code=2) typer.secho("OK: l'utente e' read-only sullo schema target.", fg=typer.colors.GREEN) diff --git a/harness/tht/cli/lsh_cmd.py b/harness/tht/cli/lsh_cmd.py index a275633f..f1916fcc 100644 --- a/harness/tht/cli/lsh_cmd.py +++ b/harness/tht/cli/lsh_cmd.py @@ -47,7 +47,9 @@ def build_cmd(config: Path = CONFIG_OPT) -> None: if not is_text_type(column.type) or not effective_eligibility(column, ann_col)[0]: continue try: - distinct = dwh.distinct_values(table_name, column_name) + distinct = dwh.distinct_values( + table_name, column_name, limit=cfg.lsh.max_values_per_column + ) except Exception as e: skipped.append(SkippedColumn(table_name, column_name, f"errore: {e}")) continue diff --git a/harness/tht/cli/memory_cmd.py b/harness/tht/cli/memory_cmd.py index 9ba16ca6..982f1abb 100644 --- a/harness/tht/cli/memory_cmd.py +++ b/harness/tht/cli/memory_cmd.py @@ -156,7 +156,6 @@ def save_one_cmd( from tht.adapters.factory import build_vector_store from tht.cli.vector_cmd import make_embedder from tht.memory import load_registry, promote, save_one_memory - from tht.vectorstore.rest_client import VectorRestClient cfg = _load_config_or_exit(config) manifest = load_session_or_exit(cfg, session) @@ -168,7 +167,7 @@ def save_one_cmd( fg=typer.colors.RED, err=True, ) raise typer.Exit(code=4) - build_vector_store(cfg, require_write=True) + store = build_vector_store(cfg, require_write=True) sdir = session_dir(cfg, session) # Promuove la decisione scelta nel registro locale (idempotente: salta se gia' presente @@ -176,9 +175,8 @@ def save_one_cmd( promote(sdir, manifest, seqs=[decision], registry_path=registry_path(cfg)) records = [r for r in load_registry(registry_path(cfg)) if r.session_id == manifest.id] - writer = VectorRestClient(cfg.vector_write_rest) embedder = make_embedder(cfg.embeddings) - count = save_one_memory(records, decision, writer=writer, embedder=embedder) + count = save_one_memory(records, decision, store=store, embedder=embedder) msg = ( f"{count} memoria salvata su pgvector (decision_seq {decision})." @@ -455,21 +453,20 @@ def index_solved_session(cfg, session_id: str) -> int: from tht.cli.sql_cmd import promoted_tables_for from tht.cli.vector_cmd import make_embedder from tht.solved import build_solved_record, save_solved_question - from tht.vectorstore.rest_client import VectorRestClient if not has_vector_write_rest(cfg): raise RuntimeError( "vector_write_rest assente: la coppia domanda->SQL si indicizza solo con la " "writer key configurata nel workspace yaml" ) - build_vector_store(cfg, require_write=True) + store = build_vector_store(cfg, require_write=True) manifest = load_session_or_exit(cfg, session_id) record = build_solved_record( session_dir(cfg, session_id), manifest, promoted_tables_for(cfg, session_id) ) return save_solved_question( record, - writer=VectorRestClient(cfg.vector_write_rest), + store=store, embedder=make_embedder(cfg.embeddings), ) diff --git a/harness/tht/cli/schema_cmd.py b/harness/tht/cli/schema_cmd.py index f8fe23aa..c0cfacd5 100644 --- a/harness/tht/cli/schema_cmd.py +++ b/harness/tht/cli/schema_cmd.py @@ -1,4 +1,5 @@ from pathlib import Path +import logging import typer from tht.adapters.factory import build_dwh @@ -8,6 +9,23 @@ from tht.db.sampling import is_text_type from tht.mschema.eligibility import classify_all schema_app = typer.Typer(help="Gestione mschema (rappresentazione canonica dello schema)") +logger = logging.getLogger(__name__) + + +def _add_examples(dwh, phys, examples) -> None: + for table_name, table in phys.tables.items(): + for column_name, column in table.columns.items(): + if not is_text_type(column.type): + continue + try: + sampled = dwh.sample_column( + table_name, column_name, limit=examples.max_per_column + ) + except Exception as exc: + logger.warning("Campionamento saltato per %s.%s: %s", + table_name, column_name, exc) + continue + column.examples = [str(value) for value in sampled if value not in (None, "")] def _load_config_or_exit(config: Path): @@ -64,18 +82,7 @@ def introspect_cmd( try: dwh = build_dwh(cfg) phys = dwh.introspect() - for table_name, table in phys.tables.items(): - for column_name, column in table.columns.items(): - if is_text_type(column.type): - column.examples = [ - str(value) - for value in dwh.sample_column( - table_name, - column_name, - limit=cfg.examples.max_per_column, - ) - if value not in (None, "") - ] + _add_examples(dwh, phys, cfg.examples) classify_all(phys, cfg.eligibility) except Exception as e: typer.secho(f"ERRORE: {e}", fg=typer.colors.RED, err=True) diff --git a/harness/tht/db/execute.py b/harness/tht/db/execute.py index 61e74c8f..75fab3c6 100644 --- a/harness/tht/db/execute.py +++ b/harness/tht/db/execute.py @@ -13,15 +13,15 @@ from tht.execute import ( DEFAULT_TIMEOUT_MS = 30_000 -def run_query(engine: Engine, sql: str, *, limit: int) -> ExecResult: +def run_query(engine: Engine, sql: str, *, limit: int, timeout_ms: int = DEFAULT_TIMEOUT_MS) -> ExecResult: limit = require_positive_int(limit, name="limit") return run_controlled( engine, sql, limit=limit, - timeout_ms=DEFAULT_TIMEOUT_MS, + timeout_ms=timeout_ms, ) -def explain(engine: Engine, sql: str) -> PlanSummary: - return _explain(engine, sql, timeout_ms=DEFAULT_TIMEOUT_MS) +def explain(engine: Engine, sql: str, *, timeout_ms: int = DEFAULT_TIMEOUT_MS) -> PlanSummary: + return _explain(engine, sql, timeout_ms=timeout_ms) diff --git a/harness/tht/memory.py b/harness/tht/memory.py index e081b478..8a6acc78 100644 --- a/harness/tht/memory.py +++ b/harness/tht/memory.py @@ -255,7 +255,7 @@ def memory_vector_record_for_decision( def save_one_memory( - records: list[MemoryRecord], decision_seq: int, *, writer, embedder + records: list[MemoryRecord], decision_seq: int, *, store, embedder ) -> int: """Targeted one-row upsert of a promoted decision to pgvector via the writer key (spec D11). This is NOT a full vectorstore resync: it embeds and pushes a single @@ -267,27 +267,22 @@ def save_one_memory( the writer's existing_vector_hashes; the embedding (Ollama round-trip) and the upsert are skipped when the content is unchanged. Idempotent by construction. - `writer` is a VectorRestClient (writer key); `embedder` an embeddings client. + `store` is the configured writable VectorStore; `embedder` an embeddings client. The destructive cleanup (sync's delete-stale step) is intentionally absent: it remains a server-side-only operation via the direct vectordb connection. """ - from tht.vectorstore.rest_writer import pack_metadata + from tht.ports.vector import VectorWriteRecord from tht.vectorstore.store import content_hash record = memory_vector_record_for_decision(records, decision_seq) if record is None: return 0 new_hash = content_hash(record.content) - existing = writer.existing_hashes("memory", ["memory"]) + existing = store.existing_hashes("memory", ["memory"]) if existing.get(record.id) == new_hash: return 0 # unchanged: skip embedding + upsert embedding = embedder.embed_documents([record.content])[0] - row = { - "record_key": record.id, - "kind": record.kind, - "content_hash": new_hash, - "metadata": pack_metadata(record), - "embedding": embedding, - } - return writer.upsert_records("memory", [row]) - + return store.upsert( + "memory", + [VectorWriteRecord(record=record, embedding=embedding, content_hash=new_hash)], + ) diff --git a/harness/tht/ports/dwh.py b/harness/tht/ports/dwh.py index 0e511008..d30dc71a 100644 --- a/harness/tht/ports/dwh.py +++ b/harness/tht/ports/dwh.py @@ -19,6 +19,12 @@ class DwhCapabilities: class DwhHealth: ok: bool detail: str | None = None + database: str | None = None + schema: str | None = None + endpoint: str | None = None + read_only: bool | None = None + writable_tables: tuple[str, ...] = () + can_create: bool = False @dataclass(frozen=True) @@ -46,4 +52,4 @@ class DwhAdapter(Protocol): def sample_column(self, table: str, column: str, *, limit: int) -> list[object]: ... - def distinct_values(self, table: str, column: str) -> DistinctValues: ... + def distinct_values(self, table: str, column: str, *, limit: int) -> DistinctValues: ... diff --git a/harness/tht/solved.py b/harness/tht/solved.py index 96de27e5..e62910de 100644 --- a/harness/tht/solved.py +++ b/harness/tht/solved.py @@ -43,25 +43,22 @@ def _solved_hash(record: VectorRecord) -> str: return content_hash(record.content + "\n" + str(record.metadata.get("sql", ""))) -def save_solved_question(record: VectorRecord, *, writer, embedder) -> int: +def save_solved_question(record: VectorRecord, *, store, embedder) -> int: """Upsert one-row della coppia domanda->SQL via writer key (stesso pattern di save_one_memory, spec D11): hash dedup client-side, embedding solo se domanda o SQL sono cambiati. `writer` e' un VectorRestClient (writer key). Ritorna il numero di righe upsertate (0 = invariata).""" - from tht.vectorstore.rest_writer import pack_metadata + from tht.ports.vector import VectorWriteRecord new_hash = _solved_hash(record) - existing = writer.existing_hashes("memory", [SOLVED_KIND]) + existing = store.existing_hashes("memory", [SOLVED_KIND]) if existing.get(record.id) == new_hash: return 0 embedding = embedder.embed_documents([record.content])[0] - return writer.upsert_records("memory", [{ - "record_key": record.id, - "kind": record.kind, - "content_hash": new_hash, - "metadata": pack_metadata(record), - "embedding": embedding, - }]) + return store.upsert( + "memory", + [VectorWriteRecord(record=record, embedding=embedding, content_hash=new_hash)], + ) class SolvedIndexError(Exception):