fix: preserve Task 2 command loading contracts

harness/tests/test_preprocess_cli.py

@@ -160,3 +160,34 @@ def test_preprocess_evidence_uses_runtime_identity_for_dev_fd_config(monkeypatch
     monkeypatch.setattr("tht.corpus.pipeline.CorpusPipeline", FakePipeline)
     command.run_from_config(Path("/dev/fd/3"))
     assert captured["workspace_id"] == "runtime-workspace"
+
+
+def test_preprocess_gc_uses_runtime_identity_for_dev_fd_config(monkeypatch, tmp_path):
+    import tht.cli.preprocess_cmd as command
+
+    cfg = SimpleNamespace(
+        runtime_identity=SimpleNamespace(workspace_id="runtime-workspace"),
+        embeddings=SimpleNamespace(model="m", dim=4),
+        vector=SimpleNamespace(max_chunk_chars=10, retain_published_generations=1),
+        paths=SimpleNamespace(artifacts=tmp_path / "artifacts"),
+    )
+    captured = {}
+
+    class FakePipeline:
+        def __init__(self, **kwargs):
+            captured.update(kwargs)
+
+        def gc(self, **kwargs):
+            captured.update(kwargs)
+            captured["workspace_id"] = self.workspace_id
+            return {"status": "succeeded", "dry_run": True, "evicted": [], "failures": []}
+
+    monkeypatch.setattr(command, "_load_config_or_exit", lambda _: cfg)
+    monkeypatch.setattr("tht.adapters.factory.build_evidence_sources", lambda _: [])
+    monkeypatch.setattr("tht.adapters.factory.build_vector_store", lambda *_args, **_kwargs: object())
+    monkeypatch.setattr("tht.cli.vector_cmd.make_embedder", lambda _: object())
+    monkeypatch.setattr("tht.corpus.pipeline.CorpusPipeline", FakePipeline)
+
+    command.gc_from_config(Path("/dev/fd/3"), dry_run=True)
+    assert captured["dry_run"] is True
+    assert captured["workspace_id"] == "runtime-workspace"

harness/tests/test_schema_fk_annotations.py

@@ -1,4 +1,8 @@
 # ruff: noqa: DTZ001
+import json
+import subprocess
+import sys
+import textwrap
 import warnings
 from datetime import datetime
 
@@ -510,3 +514,90 @@ def test_staged_sql_oversize_read_is_bounded(tmp_path, monkeypatch):
     with pytest.raises(_MachineSchemaError, match="staged_sql_too_large"):
         _read_staged_sql([staged])
     assert reads == [(1 << 20) + 1]
+
+
+def test_staged_sql_deduplicates_overlapping_roots_and_repeated_files(tmp_path):
+    from tht.cli.schema_cmd import _staged_sql_files
+
+    root = tmp_path / "approved"
+    nested = root / "nested"
+    nested.mkdir(parents=True)
+    first = root / "first.sql"
+    second = nested / "second.sql"
+    first.write_text("SELECT 1")
+    second.write_text("SELECT 1")
+
+    files = _staged_sql_files([root, nested, first, root])
+    assert files == sorted({first.resolve(), second.resolve()})
+
+
+def test_staged_sql_file_limit_counts_distinct_paths_only(tmp_path):
+    from tht.cli.schema_cmd import _MachineSchemaError, _staged_sql_files
+
+    path = tmp_path / "same.sql"
+    path.write_text("SELECT 1")
+    assert _staged_sql_files([path] * 100) == [path.resolve()]
+
+    root = tmp_path / "many"
+    root.mkdir()
+    for index in range(33):
+        (root / f"q{index:02d}.sql").write_text("SELECT 1")
+    with pytest.raises(_MachineSchemaError, match="staged_sql_too_many"):
+        _staged_sql_files([root])
+
+
+def test_fresh_process_human_warning_cardinality_is_one_across_failure_and_write_paths(tmp_path):
+    cfg = _write_workspace(tmp_path)
+    physical = tmp_path / "artifacts" / "mschema" / "physical.yaml"
+    annotations = tmp_path / "artifacts" / "mschema" / "annotations.yaml"
+    annotations.parent.mkdir(parents=True, exist_ok=True)
+    Annotations().to_yaml(annotations)
+    probe = textwrap.dedent(
+        """
+        import json, sys, warnings
+        from typer.testing import CliRunner
+        from tht.cli import app
+
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            result = CliRunner().invoke(app, sys.argv[1:])
+        print(json.dumps({
+            "warnings": sum(issubclass(w.category, FutureWarning) for w in caught),
+            "exit": result.exit_code,
+        }))
+        """
+    )
+
+
+    physical.unlink()
+    branches = [
+        ["schema", "check"],
+        ["schema", "suggest-fks"],
+        ["vector", "index-schema"],
+    ]
+    for branch in branches:
+        response = subprocess.run(
+            [sys.executable, "-c", probe, *branch, "-c", str(cfg)],
+            check=True, capture_output=True, text=True,
+        )
+        assert json.loads(response.stdout) == {"warnings": 1, "exit": 1}
+
+    _physical().to_yaml(physical)
+    response = subprocess.run(
+        [sys.executable, "-c", probe, "schema", "suggest-fks", "--write", "-c", str(cfg)],
+        check=True, capture_output=True, text=True,
+    )
+    assert json.loads(response.stdout) == {"warnings": 1, "exit": 0}
+
+    physical.unlink()
+    for branch in branches:
+        response = subprocess.run(
+            [sys.executable, "-c", probe, *branch, "--json", "-c", str(cfg)],
+            check=True, capture_output=True, text=True,
+        )
+        assert json.loads(response.stdout)["warnings"] == 0
+    response = subprocess.run(
+        [sys.executable, "-c", probe, "schema", "suggest-fks", "--write", "--json", "-c", str(cfg)],
+        check=True, capture_output=True, text=True,
+    )
+    assert json.loads(response.stdout)["warnings"] == 0