Fix benchmark worker plugin defaults before config validation

deploy/benchmark_docker/worker.py

@@ -102,6 +102,7 @@ def _prepare_runtime(request: dict[str, Any]) -> tuple[Path, dict[str, Any]]:
     # Import after setting TMPDIR so ReMe and its subprocesses cannot leave
     # case-specific temporary files outside the disposable case root.
     from reme.config import resolve_app_config
+    from reme.plugin import resolve_plugin_runtime
     from reme.schema import ApplicationConfig
 
     plugins = request.get("plugins") or []
@@ -115,7 +116,10 @@ def _prepare_runtime(request: dict[str, Any]) -> tuple[Path, dict[str, Any]]:
         log_to_file=True,
     )
     app_config["environment"] = dict(os.environ)
-    resolved_config = ApplicationConfig.model_validate(app_config)
+    # Apply plugin defaults before validation materializes schema defaults,
+    # otherwise those defaults would override plugin paths and partial jobs.
+    runtime = resolve_plugin_runtime(app_config)
+    resolved_config = ApplicationConfig.model_validate(runtime.config)
     _write_runtime_layout(case_root, resolved_config)
     # The build checkpoint reads workspace paths from this mapping. Return the
     # validated model's dump so optional config fields (including ``daily_dir``)

tests/unit/test_benchmark_docker.py

@@ -421,6 +421,43 @@ def test_source_snapshot_preserves_custom_path_rules_and_symlink_checks(tmp_path
         SourceSnapshot.from_directory(tmp_path, excludes=patterns)
 
 
+def test_worker_merges_plugin_defaults_before_validating_config(tmp_path, monkeypatch):
+    from reme.plugin import Plugin, PluginManager
+
+    defaults = {
+        "daily_dir": "notes",
+        "digest_dir": "summaries",
+        "jobs": {"plugin_version": {"backend": "base", "steps": [{"backend": "version_step"}]}},
+    }
+    manager = PluginManager([Plugin(name="example", config=defaults)])
+    monkeypatch.setattr(PluginManager, "discover", classmethod(lambda cls, specs: manager))
+    # _prepare_runtime changes cwd and TMPDIR in the disposable worker process.
+    monkeypatch.chdir(tmp_path)
+    monkeypatch.setenv("TMPDIR", str(tmp_path))
+    config = tmp_path / "benchmark.yaml"
+    config.write_text(
+        "digest_dir: custom_digest\njobs:\n  plugin_version:\n    enable_serve: false\n",
+        encoding="utf-8",
+    )
+    case_root, prepared = sandbox_worker._prepare_runtime(
+        {
+            "case_root": str(tmp_path),
+            "workspace_dir": str(tmp_path / "reme_workspace"),
+            "config": str(config),
+            "plugins": ["example"],
+        },
+    )
+
+    assert prepared["daily_dir"] == "notes"
+    assert prepared["digest_dir"] == "custom_digest"
+    job = prepared["jobs"]["plugin_version"]
+    assert job["backend"] == "base"
+    assert job["steps"] == [{"backend": "version_step"}]
+    assert job["enable_serve"] is False
+    layout = json.loads((case_root / "runtime-layout.json").read_text(encoding="utf-8"))
+    assert layout["configured_paths"]["daily_dir"] == "reme_workspace/notes"
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("mode", ["job", "queries"])
 async def test_worker_keeps_startup_error_or_cancellation_when_close_fails(tmp_path, monkeypatch, mode):