perf(turn): defer semantic preference recall during preview

loopx/capabilities/semantic_preference/__init__.py

@@ -1,12 +1,35 @@
 """Optional provider-neutral semantic preference recall."""
+from __future__ import annotations
 
-from .contract import application_receipt, maintenance_receipt, provider_doctor, recall
-from .reward_memory import run_semantic_preference_reward_memory
+from importlib import import_module
+from typing import TYPE_CHECKING, Any
 
-__all__ = [
-    "application_receipt",
-    "maintenance_receipt",
-    "provider_doctor",
-    "recall",
-    "run_semantic_preference_reward_memory",
-]
+if TYPE_CHECKING:
+    from .contract import application_receipt as application_receipt
+    from .contract import maintenance_receipt as maintenance_receipt
+    from .contract import provider_doctor as provider_doctor
+    from .contract import recall as recall
+    from .reward_memory import run_semantic_preference_reward_memory as run_semantic_preference_reward_memory
+
+# The Turn start hook reads local preferences without selecting a recall provider.
+_EXPORTS = {
+    "application_receipt": "contract",
+    "maintenance_receipt": "contract",
+    "provider_doctor": "contract",
+    "recall": "contract",
+    "run_semantic_preference_reward_memory": "reward_memory",
+}
+__all__ = list(_EXPORTS)
+
+
+def __getattr__(name: str) -> Any:
+    module = _EXPORTS.get(name)
+    if module is None:
+        raise AttributeError(f"module {__name__!r} has no attribute {name!r}")
+    value = getattr(import_module(f".{module}", __name__), name)
+    globals()[name] = value
+    return value
+
+
+def __dir__() -> list[str]:
+    return sorted(set(globals()) | set(__all__))

tests/test_loopx_turn_driver.py

@@ -1821,35 +1821,42 @@ def recording_collect_status(*args: Any, **kwargs: Any) -> dict[str, Any]:
     )
 
 
+@pytest.mark.parametrize("provider", ("file", "sqlite"))
 def test_turn_cli_projects_explicit_fresh_iteration_context(
-    tmp_path: Path,
+    tmp_path: Path, provider: str,
 ) -> None:
-    project, runtime, registry = _write_live_fixture(tmp_path)
-    output = io.StringIO()
-
-    with contextlib.redirect_stdout(output):
-        exit_code = cli_main(
-            [
-                "--registry",
-                str(registry),
-                "--runtime-root",
-                str(runtime),
-                "--format",
-                "json",
-                "turn",
-                "plan",
-                "--goal-id",
-                "loopx-turn-fixture",
-                "--agent-id",
-                "codex-fixture",
-                "--iteration-context",
-                "fresh",
-                "--include-transaction-detail",
-            ]
-        )
+    _, runtime, registry = _write_live_fixture(tmp_path, provider=provider)
+    # Use the CLI entrypoint in a fresh process. In-process
+    # callers and test collection may already have selected these providers.
+    process = subprocess.run(
+        [
+            sys.executable, "-c", "\n".join((
+                "import sys",
+                "from loopx.entrypoint import main",
+                "status = main(sys.argv[1:])",
+                "unselected = {'loopx.capabilities.semantic_preference.contract',",
+                "              'loopx.capabilities.semantic_preference.reward_memory'}",
+                "assert not unselected & sys.modules.keys(), unselected & sys.modules.keys()",
+                "raise SystemExit(status)",
+            )),
+            "--registry", str(registry),
+            "--runtime-root", str(runtime),
+            "--format", "json",
+            "turn", "plan",
+            "--goal-id", "loopx-turn-fixture",
+            "--agent-id", "codex-fixture",
+            "--iteration-context", "fresh",
+            "--include-transaction-detail",
+        ],
+        cwd=Path(__file__).resolve().parents[1],
+        env={**os.environ, "LOOPX_USAGE_PING": "0"},
+        capture_output=True,
+        text=True,
+        timeout=60,
+    )
 
-    payload = json.loads(output.getvalue())
-    assert exit_code == 0, payload
+    assert process.returncode == 0, process.stdout + process.stderr
+    payload = json.loads(process.stdout)
     assert payload["session"]["action"] == "start_new"
     assert payload["session"]["context_policy"] == {
         "schema_version": "loopx_iteration_context_policy_v0",