refactor(config): centralize indexed item loading

Consolidate validation and duplicate detection for models, use cases, and workflows while preserving loader ordering and error contracts.

Add characterization coverage for collection semantics, source paths, duplicate precedence, and workflow extension ordering.

Author: sturlese

src/flightdeck/config.py

@@ -10,8 +10,10 @@
 ledger.jsonl) lives under .flightdeck/ and is never committed.
 """
 
+from collections.abc import Callable, Iterable
 from dataclasses import dataclass, field
 from pathlib import Path
+from typing import TypeVar
 
 import yaml
 from pydantic import ValidationError
@@ -26,6 +28,8 @@
 WORKFLOWS_DIR = "workflows"
 STATE_DIR = ".flightdeck"
 
+ConfigItem = TypeVar("ConfigItem", ModelSpec, UseCase, Workflow)
+
 
 class ConfigError(Exception):
     """A human-actionable configuration problem: message always names the file."""
@@ -100,6 +104,37 @@ def _validation_error(path: Path, exc: ValidationError) -> ConfigError:
     return ConfigError(f"{path}: invalid configuration\n" + "\n".join(lines))
 
 
+def _load_indexed_items(
+    raw_items: Iterable[object],
+    *,
+    path: Path,
+    item_type: type[ConfigItem],
+    kind: str,
+    items: dict[str, ConfigItem] | None = None,
+    before_index: Callable[[ConfigItem], None] | None = None,
+) -> dict[str, ConfigItem]:
+    """Validate and index config items, rejecting duplicate IDs at their source."""
+    indexed = items if items is not None else {}
+    for raw_item in raw_items:
+        try:
+            item = item_type.model_validate(raw_item)
+        except ValidationError as exc:
+            raise _validation_error(path, exc) from None
+        if item.id in indexed:
+            raise ConfigError(f"{path}: duplicate {kind} id '{item.id}'")
+        if before_index:
+            before_index(item)
+        indexed[item.id] = item
+    return indexed
+
+
+def _validate_workflow_use_case(
+    workflow: Workflow, path: Path, usecases: dict[str, UseCase]
+) -> None:
+    if workflow.use_case and workflow.use_case not in usecases:
+        raise ConfigError(f"{path}: use_case '{workflow.use_case}' not found in {USECASES_FILE}")
+
+
 def load_org(root: Path | str) -> Org:
     """Load and cross-validate an org directory. Workflows and use cases are
     optional (a fresh org starts empty); the org file and model registry are not."""
@@ -116,45 +151,39 @@ def load_org(root: Path | str) -> Org:
         raise _validation_error(org_path, exc) from None
 
     models_path = root / MODELS_FILE
-    models: dict[str, ModelSpec] = {}
-    for item in _read_yaml(models_path).get("models") or []:
-        try:
-            spec = ModelSpec.model_validate(item)
-        except ValidationError as exc:
-            raise _validation_error(models_path, exc) from None
-        if spec.id in models:
-            raise ConfigError(f"{models_path}: duplicate model id '{spec.id}'")
-        models[spec.id] = spec
+    models = _load_indexed_items(
+        _read_yaml(models_path).get("models") or [],
+        path=models_path,
+        item_type=ModelSpec,
+        kind="model",
+    )
     if not models:
         raise ConfigError(f"{models_path}: the model registry is empty")
 
     usecases: dict[str, UseCase] = {}
     usecases_path = root / USECASES_FILE
     if usecases_path.exists():
-        for item in _read_yaml(usecases_path).get("usecases") or []:
-            try:
-                case = UseCase.model_validate(item)
-            except ValidationError as exc:
-                raise _validation_error(usecases_path, exc) from None
-            if case.id in usecases:
-                raise ConfigError(f"{usecases_path}: duplicate use case id '{case.id}'")
-            usecases[case.id] = case
+        usecases = _load_indexed_items(
+            _read_yaml(usecases_path).get("usecases") or [],
+            path=usecases_path,
+            item_type=UseCase,
+            kind="use case",
+        )
 
     workflows: dict[str, Workflow] = {}
     workflows_dir = root / WORKFLOWS_DIR
     if workflows_dir.is_dir():
         for path in sorted(workflows_dir.glob("*.yaml")) + sorted(workflows_dir.glob("*.yml")):
-            try:
-                workflow = Workflow.model_validate(_read_yaml(path))
-            except ValidationError as exc:
-                raise _validation_error(path, exc) from None
-            if workflow.id in workflows:
-                raise ConfigError(f"{path}: duplicate workflow id '{workflow.id}'")
-            if workflow.use_case and workflow.use_case not in usecases:
-                raise ConfigError(
-                    f"{path}: use_case '{workflow.use_case}' not found in {USECASES_FILE}"
-                )
-            workflows[workflow.id] = workflow
+            _load_indexed_items(
+                [_read_yaml(path)],
+                path=path,
+                item_type=Workflow,
+                kind="workflow",
+                items=workflows,
+                before_index=lambda workflow, workflow_path=path: _validate_workflow_use_case(
+                    workflow, workflow_path, usecases
+                ),
+            )
 
     # Optional SSO directory snapshot (like usecases.yaml, absent is fine).
     directory = Directory.from_file(root / DIRECTORY_FILE)

tests/test_store_config.py

@@ -5,7 +5,7 @@
 
 from flightdeck.config import ConfigError, load_org
 from flightdeck.schemas import Feedback, Run
-from tests.conftest import ORG, SUPPORT_WORKFLOW, write_org
+from tests.conftest import MODELS, ORG, SUPPORT_WORKFLOW, write_org
 
 
 def _run(run_id: str, when: datetime, **overrides) -> Run:
@@ -67,6 +67,23 @@ def test_month_cost_sums_only_that_month(store):
 # ------------------------------------------------------------------ config loading
 
 
+def _use_case(**overrides):
+    fields = {
+        "id": "ticket-triage",
+        "name": "Ticket triage",
+        "department": "Support",
+        "task_minutes": 8,
+        "tasks_per_month": 200,
+        "automation_potential": 0.6,
+        "data_readiness": 4,
+        "process_stability": 4,
+        "risk": 2,
+        "effort_weeks": 3,
+    }
+    fields.update(overrides)
+    return fields
+
+
 def test_missing_org_file_suggests_init(tmp_path):
     with pytest.raises(ConfigError, match="flightdeck init"):
         load_org(tmp_path)
@@ -82,8 +99,13 @@ def test_unknown_keys_fail_loudly(tmp_path):
 def test_dangling_use_case_reference_fails(tmp_path):
     workflow = dict(SUPPORT_WORKFLOW)
     workflow["use_case"] = "does-not-exist"
-    with pytest.raises(ConfigError, match="does-not-exist"):
-        load_org(write_org(tmp_path / "org", workflows=[workflow]))
+    root = write_org(tmp_path / "org", workflows=[workflow])
+    path = root / "workflows" / "support-reply.yaml"
+
+    with pytest.raises(ConfigError) as excinfo:
+        load_org(root)
+
+    assert str(excinfo.value) == f"{path}: use_case 'does-not-exist' not found in usecases.yaml"
 
 
 def test_empty_model_registry_fails(tmp_path):
@@ -93,6 +115,127 @@ def test_empty_model_registry_fails(tmp_path):
         load_org(root)
 
 
+@pytest.mark.parametrize("registry", [{}, {"models": None}])
+def test_missing_or_null_model_collection_is_an_empty_registry(tmp_path, registry):
+    root = write_org(tmp_path / "org")
+    (root / "models.yaml").write_text(yaml.safe_dump(registry), encoding="utf-8")
+
+    with pytest.raises(ConfigError, match=r"models\.yaml: the model registry is empty"):
+        load_org(root)
+
+
+@pytest.mark.parametrize("collection", [{}, {"usecases": None}])
+def test_missing_or_null_use_case_collection_loads_empty(tmp_path, collection):
+    root = write_org(tmp_path / "org", workflows=[])
+    (root / "usecases.yaml").write_text(yaml.safe_dump(collection), encoding="utf-8")
+
+    assert load_org(root).usecases == {}
+
+
+def test_absent_workflow_directory_loads_empty(tmp_path):
+    root = write_org(tmp_path / "org", workflows=[])
+
+    assert not (root / "workflows").exists()
+    assert load_org(root).workflows == {}
+
+
+def test_invalid_model_reports_the_model_registry_path(tmp_path):
+    model = {**MODELS[0], "tier": "unsupported"}
+    root = write_org(tmp_path / "org", models=[model])
+
+    with pytest.raises(ConfigError) as excinfo:
+        load_org(root)
+
+    message = str(excinfo.value)
+    assert str(root / "models.yaml") in message
+    assert "invalid configuration" in message
+    assert "tier" in message
+
+
+def test_duplicate_model_id_reports_kind_and_registry_path(tmp_path):
+    root = write_org(tmp_path / "org", models=[dict(MODELS[0]), dict(MODELS[0])])
+
+    with pytest.raises(ConfigError) as excinfo:
+        load_org(root)
+
+    assert str(excinfo.value) == f"{root / 'models.yaml'}: duplicate model id 'mock-fast-eu'"
+
+
+def test_invalid_use_case_reports_the_use_case_path(tmp_path):
+    root = write_org(tmp_path / "org", workflows=[])
+    (root / "usecases.yaml").write_text(
+        yaml.safe_dump({"usecases": [_use_case(task_minutes=0)]}), encoding="utf-8"
+    )
+
+    with pytest.raises(ConfigError) as excinfo:
+        load_org(root)
+
+    message = str(excinfo.value)
+    assert str(root / "usecases.yaml") in message
+    assert "invalid configuration" in message
+    assert "task_minutes" in message
+
+
+def test_duplicate_use_case_id_reports_kind_and_use_case_path(tmp_path):
+    root = write_org(tmp_path / "org", workflows=[])
+    cases = [_use_case(), _use_case(name="Another triage")]
+    (root / "usecases.yaml").write_text(yaml.safe_dump({"usecases": cases}), encoding="utf-8")
+
+    with pytest.raises(ConfigError) as excinfo:
+        load_org(root)
+
+    assert str(excinfo.value) == f"{root / 'usecases.yaml'}: duplicate use case id 'ticket-triage'"
+
+
+@pytest.mark.parametrize("suffix", [".yaml", ".yml"])
+def test_invalid_workflow_reports_its_workflow_path(tmp_path, suffix):
+    workflow = {
+        **SUPPORT_WORKFLOW,
+        "baseline": {**SUPPORT_WORKFLOW["baseline"], "minutes_per_task": 0},
+    }
+    root = write_org(tmp_path / "org", workflows=[])
+    workflows_dir = root / "workflows"
+    workflows_dir.mkdir()
+    path = workflows_dir / f"invalid{suffix}"
+    path.write_text(yaml.safe_dump(workflow), encoding="utf-8")
+
+    with pytest.raises(ConfigError) as excinfo:
+        load_org(root)
+
+    message = str(excinfo.value)
+    assert str(path) in message
+    assert "invalid configuration" in message
+    assert "baseline.minutes_per_task" in message
+
+
+def test_duplicate_workflow_id_wins_over_dangling_use_case_in_second_file(tmp_path):
+    root = write_org(tmp_path / "org", workflows=[])
+    workflows_dir = root / "workflows"
+    workflows_dir.mkdir()
+    first = workflows_dir / "first.yaml"
+    second = workflows_dir / "second.yaml"
+    duplicate_with_dangling_use_case = {**SUPPORT_WORKFLOW, "use_case": "does-not-exist"}
+    first.write_text(yaml.safe_dump(SUPPORT_WORKFLOW), encoding="utf-8")
+    second.write_text(yaml.safe_dump(duplicate_with_dangling_use_case), encoding="utf-8")
+
+    with pytest.raises(ConfigError) as excinfo:
+        load_org(root)
+
+    assert str(excinfo.value) == f"{second}: duplicate workflow id 'support-reply'"
+
+
+def test_mixed_workflow_extensions_use_yaml_then_yml_order(tmp_path):
+    root = write_org(tmp_path / "org", workflows=[])
+    workflows_dir = root / "workflows"
+    workflows_dir.mkdir()
+    yaml_workflow = {**SUPPORT_WORKFLOW, "id": "yaml-first"}
+    yml_workflow = {**SUPPORT_WORKFLOW, "id": "yml-second"}
+    (workflows_dir / "z.yaml").write_text(yaml.safe_dump(yaml_workflow), encoding="utf-8")
+    (workflows_dir / "a.yml").write_text(yaml.safe_dump(yml_workflow), encoding="utf-8")
+
+    assert list(load_org(root).workflows) == ["yaml-first", "yml-second"]
+
+
 def test_invalid_redact_pattern_is_a_loud_config_error(tmp_path):
     # A bad regex must fail at LOAD, naming the org file and the pattern — never
     # at run time, inside the redactor, mid-run.