Story S2-01 — Bench import-path resolution (load_task_class)¶
Step: Step 2 — Build harness internals: loader, cache, audit chain extension, canary + cost-tag shims
Status: HARDENED
Effort: S
Depends on: S1-01 (errors — TaskClassNotFound, BenchCaseLoadError, TaskClassAlreadyRegistered; this story adds BenchRootNotFound, TaskClassRegistrationFailed, TaskClassRootConflict, InvalidTaskClassName), S1-03 (TaskClass, TaskClassRegistry, default_registry, @register_task_class), S1-05 (locked 9-name public surface — this story must not widen it)
ADRs honored: ADR-0001 (no in-process rubric import surface), Phase 5 ADR-0006 (Protocol convention upstream of registry)
Validation notes¶
Validated: 2026-05-26
Verdict: HARDENED
Findings addressed: 55 total — 14 block, ~30 harden, ~11 nit
Critic reports: Coverage (15), Test-Quality (15), Consistency (12), Design-Patterns (13). No NEEDS RESEARCH — every pattern is precedented in this repo.
Conflict resolutions (priority order: Consistency > Coverage > Test-Quality > Design-Patterns):
- Public-surface seam (F-COV-1 / F-CON-3 / F-CON-10). Consistency wins. The original AC-1 phrase
codegenie.eval.__init__'s loader-internal seam' contradicted S1-05 AC-1 (locked__all__of exactly 9 names; the tenth fails CI). Rewrote AC-1 to:load_task_classis importable fromcodegenie.eval.loader(sub-module path); it is NOT added tocodegenie.eval.__init__.__all__. Deleted the Files-to-touch row forsrc/codegenie/eval/__init__.py. Removed the inventedloader-internal seamphrase. - Typed-error / exit-code mapping (F-COV-2 / F-TQ-4 / F-CON-4 / F-DP-3). Consistency wins via High-level-impl.md line 97 exit-code table. Pinned: missing
registration.pyfile →BenchCaseLoadError(case_dir, field="registration.py", reason="file not found")→ CLI exit code 4;registration.pyran but did not registername→TaskClassNotFound(name, looked_up_in=..., available_names=...)→ CLI exit code 3;registration.pyraised during import → newTaskClassRegistrationFailed(name, cause_type, cause_message)chained viaraise ... from→ CLI exit code 1 (generic). All three carry machine-readable attributes (not just message text) for the S4-02 CLI exit-code mapping. AC-5 ambiguity ('pick one') is closed. __all__placeholder (F-DP-2). Pinned:__all__ = ("load_task_class",)— single-tuple form. S2-02 will add"load_cases"when it lands the function. Namingload_casestoday breaksfrom codegenie.eval.loader import *andpydocintrospection.Depends oncorrection (F-CON-7). WasS1-02, S1-03; corrected toS1-01, S1-03, S1-05. The implementation outline imports nothing from S1-02 (wire models). Same correction shape S1-03 itself shipped.- Fixture directory naming (F-CON-8). Changed
tests/fixtures/bench/stub_task_class/(underscore) →tests/fixtures/bench/stub-task-class/(hyphen). The hyphen→underscore translation is import-time only; on-disk directory MUST stay hyphenated to match the registered slug (matches arch line 1016, High-level-impl.md line 77). - DI
registry=kwarg (F-DP-4). Added:load_task_class(name, bench_root=..., *, registry: TaskClassRegistry | None = None). Mirrorsregister_task_class(..., registry=...)in S1-03 andregister_plugin(..., registry=...)inplugins/registry.py. Enables clean test isolation without monkeypatching the module-leveldefault_registry. - Concrete TDD assertions (F-TQ-1 through F-TQ-15). Comment-only test stubs rewritten as runnable Python with explicit
assert/pytest.raises/ parametrize lines. Identity (is), sys.path index (== 0) and count (== 1),looked_up_inattribute, registry-call-count spy, fixture-counter-EXTERNAL-to-module-being-imported (F-TQ-13), and Hypothesis property test for name→module-name translation all pinned. - Test isolation (F-TQ-7 / F-CON-9). Added an autouse
tests/unit/eval/conftest.pyfixture that snapshots+restoressys.modulesandsys.patharound each test, plus monkeypatchesdefault_registryto a freshTaskClassRegistry(). The loader itself does NOT provide a teardown helper — global-state hygiene is a test-fixture concern. - Structural defense for ADR-0001 (F-CON-5 / F-DP-9). Added
tests/fence/test_eval_loader_no_rubric_import.pyAST-walkingsrc/codegenie/eval/loader.pyfor anybench.*.rubricreference. Pinned as AC; mirrorstests/fence/_phase4_scanner.pyprecedent. namevalidation (F-COV-6). Pinned regex^[a-z][a-z0-9-]*[a-z0-9]$(lowercase alphanumeric + hyphen; must start with letter, must not end with hyphen, min length 2). Rejects"","a","-foo","foo-","Foo","foo/bar","foo.bar","../etc/passwd","registration","_internal". NewInvalidTaskClassName(name, reason)typed exit, raised BEFORE any sys.path or import side effect.- Different bench_root same name (F-COV-5). Pinned: second call with same
namebut different resolvedbench_rootraisesTaskClassRootConflict(name, first_root, second_root). Mechanism: module-level_loaded: dict[str, Path]tracks the resolved bench_root per name; second call with mismatching resolved path raises before any import. - Registered-different-name (F-COV-11 / F-DP-8). Pinned: snapshot
default_registry's name set before/after the import; the diff feedsTaskClassNotFound(name, looked_up_in=..., available_names=tuple(sorted(diff))). Surfaces the typo case cheaply — piggy-backs on S1-03 AC-11'savailable_namescontract. - Resolved module path verification (F-COV-4). Pinned: AC asserts
sys.modules[f"bench.{module_name}.registration"].__file__equals(bench_root / name / "registration.py").resolve().as_posix(). - Structured-log events (F-COV-12 / F-CON-11 / F-DP-10). Pinned three event IDs (all matching the
^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$convention):loader.task_class_loaded(first-load success),loader.task_class_cache_hit(second call), and per-failure-path warnings/errors (loader.registration_file_missing,loader.task_class_not_registered_after_import,loader.registration_import_failed,loader.bench_root_not_found,loader.task_class_root_conflict,loader.invalid_task_class_name). - Surfaced doc drift (no auto-edit).
final-design.md §loader.pyline 186/289 still saysvuln_remediation.registration(nobench.prefix) — stale wording; phase-arch-design.md Gap 2 Option A is canonical.phase-arch-design.mdline 1159 incorrectly says "bench/ becomes a__init__.py-bearing implicit namespace package" — PEP 420 implicit namespace packages have NO__init__.py.phase-arch-design.mdline 866 says "registry rejects duplicates" but S1-03 made the registry RAISE on duplicate; the actual no-op mechanism issys.modulescaching preventing module-body re-execution. All three are flagged for a doc-sweep follow-on, not auto-fixed. - Design endorsements deferred (Rule 2 — no premature abstraction).
TaskClassNamenewtype (F-DP-1): deferred to identifier-consolidation work, per S1-03 precedent.slugify_taskclass_namehelper (F-DP-5): only one consumer today (this loader); fence-CI walks the filesystem directly.LoaderProtocol(F-DP-7): three downstream consumers exist on paper but none need an injected fake yet; trigger is "first test that wants an in-memory fake loader". Context-managersys.pathshape (F-DP-6): prepend-and-leave is correct for short-lived CLI; a context manager would breaksys.modulescaching that AC-3 depends on. All four surfaced in Notes-for-implementer as explicit deferrals with their trigger conditions.
Full audit log: _validation/S2-01-bench-import-path-resolution.md
Context¶
The synthesis docs hand-wave _codegenie_bench.{name}.registration (final-design.md §Components → loader.py); bench/ lives at repo root and isn't inside src/codegenie/, so the import does not resolve as written. phase-arch-design.md §Gap analysis & improvements §Gap 2 picks Option A: prepend the parent of bench/ to sys.path and import bench.{name}.registration directly (no synthesized prefix), so bench/ becomes a PEP 420 implicit namespace package (no __init__.py). This story implements that contract — the first concrete loader entry point, with the side-effect import that triggers @register_task_class("<name>") exactly once and returns the resolved TaskClass.
The loader is internal scaffolding: it lives at codegenie.eval.loader, is callable from Runner.plan() / audit verify / PromotionGate, but is NOT added to codegenie.eval.__init__.__all__ (S1-05 locks that surface at 9 names).
References — where to look¶
- Architecture:
../phase-arch-design.md §Component design — src/codegenie/eval/loader.py(line 556) — public-interface signatures (load_task_class,load_cases); side-effect-import idempotence note../phase-arch-design.md §Gap analysis & improvements §Gap 2(line 1153) — full rationale for Option A vs MetaPathFinder; the OQ #3 fallback if packaging conflicts surface../phase-arch-design.md §Control flow(line 826, Happy path narrative) — theRunner.plan()call site that invokesload_task_class../phase-arch-design.md §Failure modes(lines 944–963) — typed-exception mapping for case-load + task-class failures- Phase ADRs:
../ADRs/0001-rubric-execution-isolation-via-subprocess.md— loader must never importbench/{name}/rubric.py; onlyregistration.pyis in-process- Source design:
../final-design.md §Components → loader.py(line 182) — original (hand-wavy) statement of the import target; stale wording flagged in Validation notes- High-level-impl:
../High-level-impl.md §Step 2(line 48) — loader scope; Step 4 line 97 — CLI exit-code partitioning that this story's typed exits feed.- Existing code (registered by sibling stories):
src/codegenie/eval/registry.py(S1-03) —default_registry,@register_task_class; the side-effect targetsrc/codegenie/eval/models.py(S1-02) —TaskClassshape (NOT directly imported by this story, but the return type)src/codegenie/eval/errors.py(S1-01) —TaskClassNotFound,TaskClassAlreadyRegistered,BenchCaseLoadError(this story adds four new typed errors — see Implementation outline)src/codegenie/eval/__init__.py(S1-05) — locked 9-name__all__surface; this story does NOT widen it- Precedents in this repo:
src/codegenie/probes/__init__.py— explicit-imports collection point (S1-05 precedent for the eval package shape)src/codegenie/probes/registry.py:139-158— caller-frame origin capture for collision diagnosticssrc/codegenie/plugins/registry.py:189-202—register_plugin(plugin, *, registry=None)DI kwarg pattern (mirrored by this story'sregistry=kwarg)tests/fence/_phase4_scanner.py:walk_imports— single AST-kernel fortests/fence/walks (reused by F-CON-5 / F-DP-9 fence test)docs/phases/06.5-per-task-class-eval-harness/stories/_validation/S1-03-*.md— sibling validation discipline (registry-shape, immutability normalization,Finaldiscipline)
Goal¶
codegenie.eval.loader.load_task_class(name, bench_root, *, registry=None) resolves bench/{name}/registration.py via sys.path prep (Option A), triggers @register_task_class exactly once via importlib.import_module(f"bench.{module_name}.registration"), and returns the registered TaskClass. Second call with the same (name, bench_root) returns the cached TaskClass without re-executing the module body (no TaskClassAlreadyRegistered raised). Six typed failure modes (name validation, bench-root missing, registration-file missing, registration-import-raised, registration-did-not-register-name, bench-root-conflict-across-calls) carry machine-readable attributes that S4-02's CLI maps to exit codes 1/3/4 per High-level-impl.md Step 4.
Acceptance criteria¶
- [ ] AC-1 (public-surface seam):
load_task_class(name: str, bench_root: Path = Path("bench"), *, registry: TaskClassRegistry | None = None) -> TaskClassis importable asfrom codegenie.eval.loader import load_task_class(sub-module path). It is NOT added tocodegenie.eval.__init__.__all__— S1-05's locked 9-name public surface remains unchanged. A fence test asserts"load_task_class" not in codegenie.eval.__all__to prevent accidental promotion. - [ ] AC-2 (module-level
__all__):codegenie.eval.loader.__all__ == ("load_task_class",)— a single-element tuple."load_cases"is NOT listed (S2-02 adds it). Test pins the exact tuple viaassert codegenie.eval.loader.__all__ == ("load_task_class",). Rationale: namingload_casestoday breaksfrom codegenie.eval.loader import *until S2-02 lands. - [ ] AC-3 (name validation, fail-fast):
load_task_class(name, ...)validatesnameagainstre.compile(r"^[a-z][a-z0-9-]*[a-z0-9]$")BEFORE anysys.pathmutation,importlibcall, or filesystem stat. Invalid input raisesInvalidTaskClassName(name, reason)withreason ∈ {"empty", "too short", "wrong case", "leading hyphen", "trailing hyphen", "non-slug character", "reserved name"}. Parametrized test covers:""(empty),"a"(too short),"-foo"(leading hyphen),"foo-"(trailing hyphen),"Foo"(uppercase),"foo/bar"(slash),"foo.bar"(dot),"../etc/passwd"(path traversal),"registration"(would shadow the submodule),"_internal"(leading underscore),123(non-str) →TypeError,None→TypeError,b"foo"(bytes) →TypeError. - [ ] AC-4 (bench-root validation): If
bench_rootdoes not exist or is not a directory, raiseBenchRootNotFound(bench_root=bench_root.resolve())BEFORE any sys.path mutation or import. Test covers: nonexistent path, regular file masquerading as bench_root, broken symlink. - [ ] AC-5 (first-call side effect runs exactly once): On the first
load_task_class(name, bench_root)call: (a)bench/{name}/registration.py's module body executes exactly once — verified by an EXTERNAL counter (a file undertmp_path / "side_effects.txt"thatregistration.pyappends to; the counter MUST NOT live in the imported module's namespace becauseimportlib.reloadwould silently reset it — see F-TQ-13). (b)sys.path[0] == str(bench_root.resolve().parent)(parent-of-bench prepended, NOT bench itself); (c) the resolved entry appears insys.pathexactly once (sys.path.count(...) == 1); (d)sys.modules[f"bench.{module_name}.registration"]is set; (e)sys.modules[f"bench.{module_name}.registration"].__file__ == (bench_root / name / "registration.py").resolve().as_posix()— pins the resolved module identity, prevents accidental wrong-module import. - [ ] AC-6 (second-call cache hit, no re-execution, no duplicate-registration raise): On the second
load_task_class(name, bench_root)call with the same(name, resolved bench_root): (a) the external counter is unchanged (module body did not re-run); (b)TaskClassAlreadyRegisteredis NOT raised (the test wraps the second call intry/exceptand assertspytest.failif raised) — guards the path that arch line 866 hand-waves as "registry rejects duplicates" but which actually relies onsys.modulescaching preventing decorator re-firing; (c)result_first is result_second(identity, not equality); (d)result_first is registry.get(name)(the registry, not the wrapper, is the source of truth); (e)sys.path.count(str(bench_root.resolve().parent)) == 1(no growth); (f) aloader.task_class_cache_hitstructlog event is emitted (distinct from the first-callloader.task_class_loaded). - [ ] AC-7 (different bench_root, same name → conflict):
load_task_class("foo", root_A)thenload_task_class("foo", root_B)withroot_A.resolve() != root_B.resolve()raisesTaskClassRootConflict(name="foo", first_root=root_A.resolve(), second_root=root_B.resolve())— does NOT silently return the first root's TaskClass (a load-bearing footgun for--bench-rootCLI flag tests). Mechanism: module-level_loaded: dict[str, Path]tracks resolved bench_root per name; second call with mismatching resolved path raises BEFORE any sys.path or import. - [ ] AC-8 (different bench_root, different name → both work):
load_task_class("foo", root_A)thenload_task_class("bar", root_B)(different names AND different roots) both succeed; both parent-of-bench entries appear insys.pathexactly once each;result_foo is registry.get("foo")andresult_bar is registry.get("bar")both hold. - [ ] AC-9 (hyphen → underscore translation, parametrized): Parametrized over
("foo", "foo"),("vuln-remediation", "vuln_remediation"),("migration-chainguard-distroless", "migration_chainguard_distroless"),("a-b-c-d-e", "a_b_c_d_e"). For each(name, expected_module):sys.modules[f"bench.{expected_module}.registration"]exists after the call ANDsys.modules.get(f"bench.{name}.registration")is None (i.e., the original hyphenated form was NOT used as a module key). Property-based test (Hypothesis or hand-rolled): for anynamematching the AC-3 regex,module_name = name.replace("-", "_")ANDmodule_name.isidentifier(). - [ ] AC-10 (missing
registration.pyfile → typed exit, CLI exit code 4): Ifbench/{name}/registration.pydoes not exist, raiseBenchCaseLoadError(case_dir=bench_root.resolve() / name, field="registration.py", reason="file not found"). Asserted via attribute access (NOT message regex):assert exc.case_dir == bench_root.resolve() / name;assert exc.field == "registration.py";assert exc.reason == "file not found". Distinct test from "directorybench/{name}/itself missing" (handled identically — both surface asModuleNotFoundErroron the top-levelbench.{module}.registrationimport; both produceBenchCaseLoadError). - [ ] AC-11 (
registration.pyimports but does not registername→ typed exit, CLI exit code 3): Ifimportlib.import_module(...)succeeds butnameis absent fromregistryafter the import, raiseTaskClassNotFound(name, looked_up_in=f"bench.{module_name}.registration", available_names=tuple(sorted(<delta of registry names before vs after import>))). Mechanism: snapshotset(registry._by_name.keys())before the import, recompute after, take the symmetric difference. Tests: (a)registration.pywith no decorator call at all →available_names == (); (b)registration.pywith@register_task_class("typo")→available_names == ("typo",). Asserted via attribute access:assert exc.name == "vuln-remediation";assert exc.looked_up_in == "bench.vuln_remediation.registration";assert exc.available_names == ("typo",). - [ ] AC-12 (
registration.pyraises during import → typed exit, CLI exit code 1): If theregistration.pymodule body raises any exception (SyntaxError, transitive ImportError of a sibling module, arbitrary RuntimeError), raiseTaskClassRegistrationFailed(name, cause_type=type(e).__name__, cause_message=str(e)[:200])chained viaraise ... from e. AModuleNotFoundErrorwhose.name == f"bench.{module_name}.registration"is the missing-file case (AC-10); aModuleNotFoundErrorwhose.nameis a transitive missing dep is classified asTaskClassRegistrationFailed(AC-12). Tests: (a)registration.pywithraise RuntimeError("boom")→exc.cause_type == "RuntimeError",exc.cause_message == "boom",exc.__cause__is the originalRuntimeError; (b)registration.pywithimport nonexistent_module→exc.cause_type == "ModuleNotFoundError",exc.cause_message.startswith("No module named 'nonexistent_module'"). - [ ] AC-13 (symlinked
bench_rootresolves identically): Giventmp_path/real/bench/foo/registration.pyand a symlinktmp_path/link → tmp_path/real,load_task_class("foo", tmp_path/real/bench)andload_task_class("foo", tmp_path/link/bench)produce the sameTaskClassidentity ANDsys.path.count(str((tmp_path/real).resolve())) == 1(no double-import under two paths). Mechanism: the loader resolvesbench_root.resolve().parentbefore any sys.path mutation or_loadedlookup. - [ ] AC-14 (relative vs absolute
bench_rootequivalence): Bothload_task_class("foo", Path("bench"))(relative, CWD-dependent) andload_task_class("foo", tmp_path / "bench")(absolute) produce the same module identity and the samesys.moduleskey. The resolved-absolute-string is what lands onsys.path; the input form does not change the cache key. - [ ] AC-15 (machine-readable exception attributes on all six typed exits): Each of
InvalidTaskClassName,BenchRootNotFound,BenchCaseLoadError,TaskClassNotFound,TaskClassRegistrationFailed,TaskClassRootConflictexposes its diagnostic fields as named attributes (not just.args[i]). The CLI in S4-02 maps these to exit codes byisinstance(exc, …)+exc.<field>. Test asserts attribute access for each. - [ ] AC-16 (DI registry= kwarg):
load_task_class("foo", root, registry=fresh_registry)registers intofresh_registryand reads back from it. The module-leveldefault_registryis not touched. Mirrors the DI pattern in S1-03 (register_task_class(..., registry=...)) andplugins/registry.py:189-202(register_plugin(..., registry=...)). Whenregistry is None, falls back tocodegenie.eval.registry.default_registry. Test exercises both paths. - [ ] AC-17 (sys.path mutation is bounded — idempotent insert): Repeated
load_task_classcalls do not growsys.path. After N=5 calls with the same(name, bench_root),sys.path.count(str(bench_root.resolve().parent)) == 1ANDsys.path[0] == str(bench_root.resolve().parent). - [ ] AC-18 (registry-side side-effect verification — INTENT, Rule 9): The
@register_task_classside effect must actually fire on the registry. Test:before = set(registry.all_task_classes()); result = load_task_class("foo", root, registry=registry); after = set(registry.all_task_classes()); assert after - before == {result}; assert result is registry.get("foo"). Guards a mutant impl that builds and returns aTaskClassdirectly without calling the decorator (the registry would stay empty, but a behavior-only test would pass). - [ ] AC-19 (structured-log events on success and every failure path): Success path emits exactly one
structlog.infoeventloader.task_class_loadedwithname=<name>,bench_root=<resolved-absolute>,module=<f"bench.{module_name}.registration">. Cache-hit emitsloader.task_class_cache_hitwith the same keys. Each failure path emits exactly onestructlog.error(orwarn) with a distinct event ID matching^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$:loader.invalid_task_class_name,loader.bench_root_not_found,loader.registration_file_missing,loader.task_class_not_registered_after_import,loader.registration_import_failed,loader.task_class_root_conflict. Test usesstructlog.testing.capture_logs()to assert event-ID + key attribute presence. - [ ] AC-20 (concurrency contract — caller-serialized): The loader docstring documents that concurrent
load_task_classcalls within a single process are NOT supported — the caller (Runner.plan()) must serialize. The loader does NOT acquire any threading lock. Phase 16 may revisit if multi-task-class concurrent loads land. Test: this is a documentation AC; the docstring contains the literal substringcaller-serializedandRunner.plan(verified viainspect.getdoc(load_task_class)substring assertion). - [ ] AC-21 (ADR-0001 structural defense — fence test):
tests/fence/test_eval_loader_no_rubric_import.pyAST-walkssrc/codegenie/eval/loader.pyvia the sharedwalk_importskernel (tests/fence/_phase4_scanner.py) and asserts: (a) noImport/ImportFromnode references any module path containing the substringrubric; (b) noimportlib.import_module(...)literal argument contains the substringrubric(extracted viaast.Constantwalk). Fence test runs on every PR. Mirrorstests/fence/test_pyproject_fence_phase4.pyprecedent. - [ ] AC-22 (test isolation autouse fixture):
tests/unit/eval/conftest.pyprovides an autouse fixture that (a) snapshotssys.pathanddict(sys.modules)on entry; (b) yields control to the test; (c) restoressys.pathfrom snapshot and removes any newbench.*keys fromsys.moduleson exit; (d) monkeypatchescodegenie.eval.registry.default_registryto a freshTaskClassRegistry()for the duration of the test. Loader tests run in arbitrary order without cross-contamination. - [ ] AC-23 (typecheck + lint clean):
ruff format --check,ruff check, andmypy --strictare clean onsrc/codegenie/eval/loader.py,src/codegenie/eval/errors.py(new exception classes), and the test file. All new exception classes carry@dataclass(frozen=True)semantics or explicit__init__storing the diagnostic fields as instance attributes. - [ ] AC-24 (TDD red→green transition): The full red test suite (every AC above with at least one runnable test) is committed in a single commit BEFORE any production code lands, with a passing import of
codegenie.eval.loaderblocked (because the module does not yet exist). The green commit implements the loader; all tests pass.
Implementation outline¶
- Add four new typed errors to
src/codegenie/eval/errors.py(S1-01 amendment via additive surface): BenchRootNotFound(bench_root: Path)— exit code 4.InvalidTaskClassName(name: object, reason: str)— exit code 1 (validation error;nametypedobjectto accept the123/Nonebad-arg cases without coercion).TaskClassRegistrationFailed(name: str, cause_type: str, cause_message: str)— exit code 1.TaskClassRootConflict(name: str, first_root: Path, second_root: Path)— exit code 1. All four expose their fields as named instance attributes (not just.args).- Create
src/codegenie/eval/loader.pywith module docstring quoting Gap #2 Option A, thecaller-serializedconcurrency contract, and theRunner.plancall-site reference. - Module-level state:
_loaded: dict[str, Path] = {}— tracks the resolved bench_root per name (drives AC-7 conflict detection)._NAME_RE: Final[re.Pattern[str]] = re.compile(r"^[a-z][a-z0-9-]*[a-z0-9]$").__all__ = ("load_task_class",). - Pure helpers (functional core):
_validate_name(name: object) -> str— raisesTypeErrorfor non-str,InvalidTaskClassNamefor regex/reserved-name failures, otherwise returns the validatedstr._translate(name: str) -> str—name.replace("-", "_"). Pure; one line.- Impure helper (imperative shell):
_prep_bench_sys_path(bench_root_parent: Path) -> None—bench_root_parentis alreadybench_root.resolve().parent. Inserts atsys.path[0]only if missing. No return.- Implement
load_task_class(name, bench_root=Path("bench"), *, registry=None) -> TaskClass: name = _validate_name(name)(AC-3 — fail-fast BEFORE any I/O).registry = registry or default_registry.bench_root_resolved = bench_root.resolve().- If
not bench_root_resolved.is_dir():structlog.error("loader.bench_root_not_found", ...); raiseBenchRootNotFound(bench_root_resolved). - AC-7 conflict check:
if name in _loaded and _loaded[name] != bench_root_resolved:raiseTaskClassRootConflict(name, _loaded[name], bench_root_resolved). - AC-6 cache-hit short-circuit:
if name in _loaded:structlog.info("loader.task_class_cache_hit", ...); returnregistry.get(name). _prep_bench_sys_path(bench_root_resolved.parent).module_name = _translate(name).- Snapshot
before_names = set(registry.all_names())(or whatever the S1-03 accessor is — use the public accessor, NOT_by_name). - Try
importlib.import_module(f"bench.{module_name}.registration"):- On
ModuleNotFoundErrorwhose.name == f"bench.{module_name}.registration": emitloader.registration_file_missing; raiseBenchCaseLoadError(bench_root_resolved / name, "registration.py", "file not found")chained viafrom e. - On any other exception
e: emitloader.registration_import_failed; raiseTaskClassRegistrationFailed(name, type(e).__name__, str(e)[:200])frome.
- On
- AC-11 post-import check:
after_names = set(registry.all_names());delta = tuple(sorted(after_names - before_names)); ifname not in after_names: emitloader.task_class_not_registered_after_import; raiseTaskClassNotFound(name, looked_up_in=f"bench.{module_name}.registration", available_names=delta). _loaded[name] = bench_root_resolved.- emit
loader.task_class_loaded; returnregistry.get(name). - NOT touched:
src/codegenie/eval/__init__.py— S1-05's__all__stays locked at 9 names. The loader is reached viacodegenie.eval.loader.load_task_class.
TDD plan — red / green / refactor¶
Red — test files¶
tests/unit/eval/conftest.py(new) — autouse_isolate_eval_globalsfixture (AC-22).tests/unit/eval/test_loader_import_path.py(new) — ACs 1–20, 23–24.tests/unit/eval/test_loader_errors.py(new) — AC-3, 4, 7, 10, 11, 12 (every failure path; attribute-shape assertions).tests/fence/test_eval_loader_no_rubric_import.py(new) — AC-21.tests/fixtures/bench/stub-task-class/registration.py(new — hyphen NOT underscore; AC-9 + F-CON-8).tests/unit/eval/_bench_factory.py(new helper — builds parameterizedbench/<name>/registration.pytrees undertmp_path; reusable by S3-01 per F-TQ-14).
Sample concrete tests (NOT comment-only stubs)¶
# tests/unit/eval/_bench_factory.py
from pathlib import Path
import textwrap
def make_bench(
tmp_path: Path,
*,
name: str = "stub-task-class",
register_name: str | None = None, # None → use `name`; "" → omit decorator
side_effect_log: Path | None = None,
body_raises: str | None = None,
) -> Path:
"""Builds bench/<name>/registration.py under tmp_path. Returns bench_root."""
bench_root = tmp_path / "bench"
pkg_dir = bench_root / name.replace("-", "_")
pkg_dir.mkdir(parents=True, exist_ok=True)
register_call = (
""
if register_name == ""
else f'@register_task_class({register_name or name!r}, ...)' # ... filled in by caller via S1-03 kwargs
)
body = textwrap.dedent(f"""
from pathlib import Path
from codegenie.eval.registry import register_task_class
{f"Path({str(side_effect_log)!r}).open('a').write('x\\n')" if side_effect_log else ""}
{f"raise {body_raises}" if body_raises else ""}
{register_call}
class StubRubric: ...
""")
(pkg_dir / "registration.py").write_text(body)
return bench_root
# tests/unit/eval/test_loader_import_path.py
import sys
import pytest
from pathlib import Path
from codegenie.eval.loader import load_task_class
from codegenie.eval.registry import TaskClassRegistry
from ._bench_factory import make_bench
def test_first_call_runs_module_body_exactly_once_with_external_counter(tmp_path):
log = tmp_path / "side_effects.txt"
bench_root = make_bench(tmp_path, name="stub-task-class", side_effect_log=log)
registry = TaskClassRegistry()
result1 = load_task_class("stub-task-class", bench_root, registry=registry)
assert log.read_text().count("x\n") == 1 # AC-5(a)
result2 = load_task_class("stub-task-class", bench_root, registry=registry)
assert log.read_text().count("x\n") == 1 # AC-6(a) — body did NOT re-run
assert result1 is result2 # AC-6(c) — identity, not equality
assert result1 is registry.get("stub-task-class") # AC-6(d) — registry is source of truth
def test_sys_path_prepend_is_idempotent_at_index_zero(tmp_path):
bench_root = make_bench(tmp_path, name="stub-task-class")
registry = TaskClassRegistry()
for _ in range(5):
load_task_class("stub-task-class", bench_root, registry=registry)
expected = str(bench_root.resolve().parent)
assert sys.path[0] == expected # AC-5(b), AC-17
assert sys.path.count(expected) == 1 # AC-5(c), AC-17
def test_resolved_module_file_matches_registration_path(tmp_path):
bench_root = make_bench(tmp_path, name="stub-task-class")
load_task_class("stub-task-class", bench_root, registry=TaskClassRegistry())
mod = sys.modules["bench.stub_task_class.registration"] # AC-5(d)
expected = (bench_root / "stub-task-class" / "registration.py").resolve().as_posix()
assert Path(mod.__file__).resolve().as_posix() == expected # AC-5(e), AC-4 (COV-4)
@pytest.mark.parametrize("name, module", [
("foo", "foo"),
("vuln-remediation", "vuln_remediation"),
("migration-chainguard-distroless", "migration_chainguard_distroless"),
("a-b-c-d-e", "a_b_c_d_e"),
])
def test_hyphen_to_underscore_translation_parametrized(tmp_path, name, module):
bench_root = make_bench(tmp_path, name=name)
load_task_class(name, bench_root, registry=TaskClassRegistry())
assert f"bench.{module}.registration" in sys.modules # AC-9
assert sys.modules.get(f"bench.{name}.registration") is None # the hyphen form must NOT be a key
def test_second_call_with_different_bench_root_raises_root_conflict(tmp_path):
root_a = make_bench(tmp_path / "a", name="foo")
root_b = make_bench(tmp_path / "b", name="foo")
registry = TaskClassRegistry()
load_task_class("foo", root_a, registry=registry)
with pytest.raises(TaskClassRootConflict) as exc:
load_task_class("foo", root_b, registry=registry)
assert exc.value.name == "foo" # AC-7
assert exc.value.first_root == root_a.resolve()
assert exc.value.second_root == root_b.resolve()
def test_intent_side_effect_fires_on_registry(tmp_path):
# Rule 9 / AC-18: verifies INTENT (decorator fired) not just behavior (TaskClass returned).
bench_root = make_bench(tmp_path, name="stub-task-class")
registry = TaskClassRegistry()
before = set(registry.all_task_classes())
result = load_task_class("stub-task-class", bench_root, registry=registry)
after = set(registry.all_task_classes())
assert after - before == {result} # AC-18
assert result is registry.get("stub-task-class")
# tests/unit/eval/test_loader_errors.py
import pytest
from codegenie.eval.errors import (
BenchCaseLoadError, BenchRootNotFound, InvalidTaskClassName,
TaskClassNotFound, TaskClassRegistrationFailed,
)
from codegenie.eval.loader import load_task_class
from codegenie.eval.registry import TaskClassRegistry
from ._bench_factory import make_bench
@pytest.mark.parametrize("bad", ["", "a", "-foo", "foo-", "Foo", "foo/bar", "foo.bar", "../etc", "_internal", "registration"])
def test_invalid_name_raises_typed(tmp_path, bad):
with pytest.raises(InvalidTaskClassName) as exc:
load_task_class(bad, tmp_path, registry=TaskClassRegistry())
assert exc.value.name == bad # AC-3, AC-15
@pytest.mark.parametrize("bad", [123, None, b"foo", ["foo"], 1.5])
def test_non_str_name_raises_typeerror(tmp_path, bad):
with pytest.raises(TypeError):
load_task_class(bad, tmp_path, registry=TaskClassRegistry()) # AC-3
def test_missing_bench_root_raises_typed(tmp_path):
with pytest.raises(BenchRootNotFound) as exc:
load_task_class("foo", tmp_path / "nonexistent", registry=TaskClassRegistry())
assert exc.value.bench_root == (tmp_path / "nonexistent").resolve() # AC-4
def test_missing_registration_file_raises_bench_case_load_error(tmp_path):
bench_root = tmp_path / "bench"
(bench_root / "foo").mkdir(parents=True) # directory exists but no registration.py
with pytest.raises(BenchCaseLoadError) as exc:
load_task_class("foo", bench_root, registry=TaskClassRegistry())
assert exc.value.case_dir == (bench_root / "foo").resolve() # AC-10, AC-15
assert exc.value.field == "registration.py"
assert exc.value.reason == "file not found"
def test_registration_imports_but_doesnt_register_name_raises_typed(tmp_path):
bench_root = make_bench(tmp_path, name="foo", register_name="") # no decorator at all
with pytest.raises(TaskClassNotFound) as exc:
load_task_class("foo", bench_root, registry=TaskClassRegistry())
assert exc.value.name == "foo" # AC-11, AC-15
assert exc.value.looked_up_in == "bench.foo.registration"
assert exc.value.available_names == ()
def test_registration_registers_typo_name_surfaces_in_available_names(tmp_path):
bench_root = make_bench(tmp_path, name="vuln-remediation", register_name="typo")
with pytest.raises(TaskClassNotFound) as exc:
load_task_class("vuln-remediation", bench_root, registry=TaskClassRegistry())
assert exc.value.available_names == ("typo",) # AC-11 + F-DP-8
def test_registration_raises_at_import_time_classified(tmp_path):
bench_root = make_bench(tmp_path, name="foo", body_raises='RuntimeError("boom")')
with pytest.raises(TaskClassRegistrationFailed) as exc:
load_task_class("foo", bench_root, registry=TaskClassRegistry())
assert exc.value.cause_type == "RuntimeError" # AC-12, AC-15
assert "boom" in exc.value.cause_message
assert isinstance(exc.__cause__, RuntimeError) or isinstance(exc.value.__cause__, RuntimeError)
# tests/fence/test_eval_loader_no_rubric_import.py
import ast
from pathlib import Path
from tests.fence._phase4_scanner import walk_imports # reused AST kernel
LOADER = Path("src/codegenie/eval/loader.py")
def test_loader_does_not_import_rubric():
src = LOADER.read_text()
tree = ast.parse(src)
for imp in walk_imports(tree):
assert "rubric" not in imp.module_path # AC-21
# also walk all importlib.import_module(...) literal args
for node in ast.walk(tree):
if (isinstance(node, ast.Call)
and isinstance(node.func, ast.Attribute)
and node.func.attr == "import_module"):
for arg in node.args:
if isinstance(arg, ast.Constant) and isinstance(arg.value, str):
assert "rubric" not in arg.value
Green — smallest impl¶
The seven-step body in §Implementation outline §6, plus the four typed-error additions to errors.py (~30 lines total) and the autouse conftest fixture (~15 lines). Total ~80 lines of code + ~150 lines of test.
Refactor¶
- Add Sphinx-style docstrings to
load_task_class,_validate_name,_translate,_prep_bench_sys_pathquoting Gap #2 Option A and thecaller-serializedconcurrency contract. - Inline-comment the hyphen→underscore translation as the ONE place we cross the slug/module-name boundary (F-DP-5 deferral).
- Ensure every structlog event ID matches the
^[a-z][a-z0-9_]*\.[a-z][a-z0-9_]*$regex; add a module-level_LOG_EVENTS: Final[frozenset[str]]collection for self-documentation (not enforced, just discoverable). - Run
mypy --stricton the module; ensurePath,dict[str, Path],re.Pattern[str],TaskClassRegistry | Noneall typecheck cleanly.
Files to touch¶
| Path | Why |
|---|---|
src/codegenie/eval/loader.py |
New module — Option A sys.path prep + load_task_class |
src/codegenie/eval/errors.py |
Add BenchRootNotFound, InvalidTaskClassName, TaskClassRegistrationFailed, TaskClassRootConflict (additive — S1-01 surface widening) |
tests/unit/eval/conftest.py |
Autouse fixture for sys.modules + sys.path + default_registry isolation (AC-22) |
tests/unit/eval/test_loader_import_path.py |
Red tests for happy paths + caching + AC-5..AC-9, AC-13..AC-20 |
tests/unit/eval/test_loader_errors.py |
Red tests for the six typed exits (AC-3, 4, 7, 10, 11, 12) |
tests/unit/eval/_bench_factory.py |
Helper builder for bench/<name>/registration.py trees (replaces hard-coded fixture; F-TQ-14) |
tests/fence/test_eval_loader_no_rubric_import.py |
Fence test for ADR-0001 (AC-21) |
tests/fixtures/bench/stub-task-class/registration.py |
Minimal hyphen-named fixture; reused by S3-01 runner story (F-CON-8) |
NOT touched: src/codegenie/eval/__init__.py — S1-05's 9-name __all__ lock stands.
Out of scope¶
- Case loading and digest verification — handled by S2-02 (
load_cases). - MetaPathFinder fallback (Option B) — surfaces only if Option A causes packaging conflicts in CI; tracked as OQ #3.
- Bench-root discovery from CWD — caller passes
bench_root; auto-discovery is a CLI concern (S4-01/S4-02). ThePath("bench")default is a test-convenience only; production callers MUST passbench_rootexplicitly. TaskClassNamenewtype extraction — deferred per S1-03's identifier-consolidation precedent. Revisit when Phase 7 ships the second task class.slugify_taskclass_name(name)helper extract — deferred until a second consumer materializes (fence-CI walks the filesystem directly, not the slug→module-name translation). Rule of three not crossed.LoaderProtocolextraction — deferred until the first consumer needs to inject a fake loader (likely Phase 7 multi-task-class runner tests).- Context-manager
sys.pathshape — prepend-and-leave is correct for the short-lived CLI and matches AC-3'ssys.modules-caching dependency. - Threading lock for concurrent calls — deferred; caller-serialized contract documented in AC-20. Phase 16 may revisit.
- Doc-sweep for stale
final-design.mdline 186/289 andphase-arch-design.mdline 866, 1159 wording — flagged in Validation notes; spawn-task candidate, not a blocker for this story.
Notes for the implementer¶
bench/does not need__init__.py— implicit namespace packages (PEP 420) work as long as the parent dir is onsys.path.phase-arch-design.mdline 1159's wording (__init__.py-bearing implicit namespace package) is internally contradictory; this story implements the no-__init__.pyform.- ADR-0001 hard line: Don't import
bench/{name}/rubric.pyfrom anywhere reachable here, transitively or otherwise. The decorator'srubric_classkwarg captures a class object (data), not an import path; the runner (S3-01) reaches the rubric via a subprocess only. AC-21's fence test is the structural defense. - Test fixture name is hyphenated (
tests/fixtures/bench/stub-task-class/); the underscore form (stub_task_class) only exists as the on-the-fly translated module name insidesys.modules. This matches the registered slug convention acrossbench/vuln-remediation/,bench/migration-chainguard-distroless/, etc. name.replace("-", "_")is the ONLY place insrc/codegenie/eval/that crosses between user-facing slug and Python module name. Fence-CI (S7-01 assertion #1) walksbench/<hyphenated-slug>/directly — it does NOT do this translation. Keep them separate; if a third consumer materializes (curators-CLI scaffolder, S5-07), extractslugify_taskclass_name(name)tocodegenie.eval.namingthen.sys.pathmutation is intentionally prepend-and-leave, not a context manager. A context manager would forceimportlib.invalidate_caches()on every call and break thesys.modulescache that AC-6 depends on. Rule of three not crossed.- Test isolation: every test that calls
load_task_classMUST use the autouse fixture intests/unit/eval/conftest.py. The loader does NOT provide a teardown helper —sys.path/sys.modules/default_registryhygiene is a test-fixture concern, mirroring S1-03's discipline (no mutation of singleton private state). - The loader READS
default_registry(whenregistry=None) but never MUTATES it directly. Registration is the side effect ofimportlib.import_module(...)running@register_task_class(...). Do NOT touchdefault_registry._by_namefrom the loader — same discipline S1-03 enforces. default_registry: Final[TaskClassRegistry]is locked by S1-03 AC-4a. The loader usesfrom codegenie.eval.registry import default_registry(read-only access via theFinalannotation); mypy--strictblocks any accidental reassignment.- Functional core / imperative shell:
_validate_nameand_translateare pure;_prep_bench_sys_pathandimportlib.import_moduleare impure. Don't blend them. A future refactor wrapping the impure surface in a_load_module(name, bench_root)private helper is welcome; inlining I/O into the pure helpers is not. - The
Path("bench")default is a test convenience; production callers MUST passbench_rootexplicitly (CLI resolves it in S4-02). If a future refactor surfaces CWD-coupling, drop the default to a required kwarg. - Caller-serialized concurrency.
Runner.plan()is async but the loader contract is single-threaded: callers must serialize. The docstring documents this; the loader does not acquire any threading lock. (Verified by AC-20.) - Phase 0's
codegenie/probes/registry pattern (@register_probe) is the closest precedent overall; the difference isbench/lives outsidesrc/codegenie/, which is exactly what Gap #2 calls out.plugins/registry.py:189-202'sregister_plugin(..., registry=...)is the closest precedent for the DI kwarg shape adopted in AC-16.