feat: enhance job info handling and normalize file paths

- Added a new method to normalize job file paths based on the current execution directory.
- Updated the `emit` method to handle `job_info` events, ensuring paths for input, output, log, and benchmark are correctly prefixed.
- Refactored the effective work directory handling to improve path resolution and context management.

Author: zhanghaomiao

src/snakemake_logger_plugin_flowo/plugin/client/log_handler.py

@@ -188,7 +188,8 @@ def emit(self, record: LogRecord) -> None:
                 configfiles, config = self._get_configfiles()
                 self.context["configfiles"] = configfiles
                 self._merge_workflow_config_context(config)
-                self._set_effective_workdir_from_snakemake()
+            elif event_name == "job_info":
+                self._normalize_job_file_paths(data)
 
             self._send_to_api(event_name, data)
         except Exception as e:
@@ -286,24 +287,41 @@ def _merge_workflow_config_context(self, config: dict) -> None:
         if tags:
             self.context["flowo_tags"] = tags
 
-    def _set_effective_workdir_from_snakemake(self) -> None:
-        """Use Snakemake's effective workdir, including ``--directory``."""
+    def _current_execution_prefix(self) -> Path | None:
+        """Relative cwd prefix for paths emitted from Snakemake's effective workdir."""
         try:
-            import snakemake.workflow
+            workflow_root = Path(str(self.context.get("workdir") or "")).resolve()
+            current = Path.cwd().resolve()
+            if current == workflow_root:
+                return None
+            return current.relative_to(workflow_root)
+        except ValueError:
+            return None
+        except Exception as e:
+            logger.debug(f"Failed to derive snakemake execution prefix: {e}")
+            return None
 
-            wf = getattr(snakemake.workflow, "workflow", None)
-            if not wf:
-                return
+    def _normalize_job_file_paths(self, data: dict) -> None:
+        prefix = self._current_execution_prefix()
+        if prefix is None or str(prefix) == ".":
+            return
 
-            workdir = (
-                getattr(wf, "overwrite_workdir", None)
-                or getattr(wf, "workdir_init", None)
-                or getattr(wf, "_workdir_init", None)
-            )
-            if workdir:
-                self.context["workdir"] = str(Path(workdir).resolve())
-        except Exception as e:
-            logger.debug(f"Failed to access snakemake workflow workdir: {e}")
+        def normalize_one(raw: object) -> str:
+            path = Path(str(raw))
+            if path.is_absolute():
+                return str(path)
+            if path.parts[: len(prefix.parts)] == prefix.parts:
+                return path.as_posix()
+            return (prefix / path).as_posix()
+
+        for key in ("input", "output", "log", "benchmark"):
+            value = data.get(key)
+            if value is None:
+                continue
+            if isinstance(value, list):
+                data[key] = [normalize_one(v) for v in value]
+            else:
+                data[key] = [normalize_one(value)]
 
     def close(self) -> None:
         self.file_handler.close()

tests/snakemake_compat/test_log_handler_compat.py

@@ -7,9 +7,7 @@
 import importlib.util
 import inspect
 import logging
-import sys
-from pathlib import Path
-from types import ModuleType, SimpleNamespace
+from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -542,19 +540,12 @@ def test_emit_workflow_started_preserves_cli_name_tags_when_config_is_empty():
     assert handler.context["flowo_tags"] == ["rna-seq", "demo"]
 
 
-def test_emit_workflow_started_records_snakemake_effective_workdir(
-    tmp_path, monkeypatch
-):
-    import snakemake
-
+def test_emit_job_info_prefixes_paths_from_current_execution_dir(tmp_path, monkeypatch):
     from snakemake_logger_plugin_flowo.plugin.client.log_handler import FlowoLogHandler
 
-    effective_workdir = tmp_path / ".test"
-    effective_workdir.mkdir()
-    fake_workflow_module = ModuleType("snakemake.workflow")
-    fake_workflow_module.workflow = SimpleNamespace(overwrite_workdir=effective_workdir)
-    monkeypatch.setitem(sys.modules, "snakemake.workflow", fake_workflow_module)
-    monkeypatch.setattr(snakemake, "workflow", fake_workflow_module, raising=False)
+    workflow_root = tmp_path / "rna-seq-star-deseq2"
+    execution_dir = workflow_root / ".test"
+    execution_dir.mkdir(parents=True)
 
     with (
         patch.object(FlowoLogHandler, "_init_file_handler", return_value=MagicMock()),
@@ -564,24 +555,42 @@ def test_emit_workflow_started_records_snakemake_effective_workdir(
     ):
         mock_client_cls.return_value = MagicMock(is_closed=False)
         handler = FlowoLogHandler(_make_common_settings())
+        handler.context["workdir"] = str(workflow_root)
         handler.file_handler.emit = MagicMock()
         handler._parsers = {
-            "workflow_started": lambda record: SimpleNamespace(
+            "job_info": lambda record: SimpleNamespace(
                 model_dump=lambda mode="json": {
-                    "workflow_id": "wf-1",
-                    "snakefile": "Snakefile",
-                    "rules": [],
+                    "job_id": 1,
+                    "rule_name": "deseq2",
+                    "threads": 1,
+                    "input": ["data/in.tsv"],
+                    "output": [".test/results/out.tsv"],
+                    "log": ["logs/deseq2/init.log"],
+                    "benchmark": None,
                 }
             )
         }
 
         with (
-            patch.object(handler, "_get_configfiles", return_value=([], {})),
-            patch.object(handler, "_send_to_api"),
+            monkeypatch.context() as m,
+            patch.object(handler, "_send_to_api") as send_to_api,
         ):
-            handler.emit(SimpleNamespace(event="workflow_started"))
+            m.chdir(execution_dir)
+            handler.emit(SimpleNamespace(event="job_info"))
 
-    assert handler.context["workdir"] == str(Path(effective_workdir).resolve())
+    send_to_api.assert_called_once_with(
+        "job_info",
+        {
+            "job_id": 1,
+            "rule_name": "deseq2",
+            "threads": 1,
+            "input": [".test/data/in.tsv"],
+            "output": [".test/results/out.tsv"],
+            "log": [".test/logs/deseq2/init.log"],
+            "benchmark": None,
+        },
+    )
+    assert handler.context["workdir"] == str(workflow_root)
 
 
 def test_emit_workflow_started_uses_non_empty_config_name_tags():