feat(runtime): add event_log_dir to configure event log output directory

src/conductor/cli/run.py

@@ -113,6 +113,20 @@ def generate_log_path(workflow_name: str) -> Path:
     return path
 
 
+def _resolve_event_log_dir(
+    event_log_dir: str | None,
+    workflow_path: Path,
+) -> Path | None:
+    """Resolve an event-log directory relative to its workflow file."""
+    if not event_log_dir:
+        return None
+
+    path = Path(event_log_dir)
+    if not path.is_absolute():
+        path = workflow_path.resolve().parent / path
+    return path.resolve()
+
+
 def init_file_logging(log_path: Path) -> None:
     """Initialize file logging to the given path.
 
@@ -2054,7 +2068,15 @@ async def run_workflow_async(
         # Start JSONL event log subscriber (always-on structured diagnostics)
         from conductor.engine.event_log import EventLogSubscriber
 
-        event_log_subscriber = EventLogSubscriber(config.workflow.name)
+        _event_log_dir = _resolve_event_log_dir(
+            config.workflow.runtime.event_log_dir,
+            workflow_path,
+        )
+
+        event_log_subscriber = EventLogSubscriber(
+            config.workflow.name,
+            event_log_dir=_event_log_dir,
+        )
         emitter.subscribe(event_log_subscriber.on_event)
 
         # Write the Fleet Manager run record (E2): this is the first point
@@ -2765,10 +2787,16 @@ async def resume_workflow_async(
             # appends to it and reuses run_id; otherwise it generates fresh.
             from conductor.engine.event_log import EventLogSubscriber
 
+            _event_log_dir = _resolve_event_log_dir(
+                config.workflow.runtime.event_log_dir,
+                resolved_workflow_path,
+            )
+
             event_log_subscriber = EventLogSubscriber(
                 config.workflow.name,
                 existing_path=existing_log_path,
                 existing_run_id=cp.run_id or None,
+                event_log_dir=_event_log_dir,
             )
             emitter.subscribe(event_log_subscriber.on_event)
 

src/conductor/config/schema.py

@@ -3452,6 +3452,16 @@ def _coerce_provider(cls, value: Any) -> Any:
     provider declares ``capabilities.working_dir=False``.
     """
 
+    event_log_dir: str | None = None
+    """Directory for event log output.
+
+    When set, event logs are written to this directory instead of
+    ``$TMPDIR/conductor/``. Relative paths are resolved against the
+    workflow file's directory.
+
+    When omitted, behavior is unchanged (writes to ``$TMPDIR/conductor/``).
+    """
+
     skills: list[str] = Field(default_factory=list)
     """Workflow-wide default skills for every provider-backed agent.
 

src/conductor/engine/event_log.py

@@ -79,6 +79,7 @@ def __init__(
         *,
         existing_path: Path | None = None,
         existing_run_id: str | None = None,
+        event_log_dir: Path | None = None,
     ) -> None:
         """Initialise the subscriber.
 
@@ -150,11 +151,12 @@ def __init__(
                 )
             self._run_id = new_run_id()
         ts = time.strftime("%Y%m%d-%H%M%S")
-        self._path = (
-            Path(tempfile.gettempdir())
-            / "conductor"
-            / f"conductor-{workflow_name}-{ts}-{self._run_id}.events.jsonl"
+        base_dir = (
+            event_log_dir
+            if event_log_dir is not None
+            else Path(tempfile.gettempdir()) / "conductor"
         )
+        self._path = base_dir / f"conductor-{workflow_name}-{ts}-{self._run_id}.events.jsonl"
         self._path.parent.mkdir(parents=True, exist_ok=True)
         self._handle = open(self._path, "w", encoding="utf-8")  # noqa: SIM115
 

tests/test_cli/test_event_log_dir.py

@@ -0,0 +1,77 @@
+"""Tests for event log directory path resolution."""
+
+from pathlib import Path
+
+import pytest
+
+from conductor.cli.run import _resolve_event_log_dir
+
+
+def test_relative_event_log_dir_is_resolved_from_workflow_directory(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Relative paths use the workflow file directory, not process CWD."""
+    workflow_dir = tmp_path / "workflow"
+    workflow_dir.mkdir()
+    workflow_path = workflow_dir / "workflow.yaml"
+    workflow_path.write_text("", encoding="utf-8")
+
+    other_cwd = tmp_path / "elsewhere"
+    other_cwd.mkdir()
+    monkeypatch.chdir(other_cwd)
+
+    result = _resolve_event_log_dir("./logs", workflow_path)
+
+    assert result == (workflow_dir / "logs").resolve()
+    assert result != (other_cwd / "logs").resolve()
+
+
+def test_parent_relative_event_log_dir_is_workflow_relative(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Parent components are evaluated from the workflow directory."""
+    workflow_dir = tmp_path / "project" / "workflows"
+    workflow_dir.mkdir(parents=True)
+    workflow_path = workflow_dir / "workflow.yaml"
+    workflow_path.write_text("", encoding="utf-8")
+
+    other_cwd = tmp_path / "elsewhere"
+    other_cwd.mkdir()
+    monkeypatch.chdir(other_cwd)
+
+    result = _resolve_event_log_dir("../logs", workflow_path)
+
+    assert result == (workflow_dir.parent / "logs").resolve()
+
+
+def test_absolute_event_log_dir_is_unchanged(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Absolute paths do not depend on workflow location or process CWD."""
+    workflow_dir = tmp_path / "workflow"
+    workflow_dir.mkdir()
+    workflow_path = workflow_dir / "workflow.yaml"
+    workflow_path.write_text("", encoding="utf-8")
+
+    other_cwd = tmp_path / "elsewhere"
+    other_cwd.mkdir()
+    monkeypatch.chdir(other_cwd)
+
+    absolute_dir = tmp_path / "absolute-logs"
+
+    assert _resolve_event_log_dir(str(absolute_dir), workflow_path) == absolute_dir.resolve()
+
+
+@pytest.mark.parametrize("value", [None, ""])
+def test_missing_event_log_dir_remains_unset(
+    tmp_path: Path,
+    value: str | None,
+) -> None:
+    """Omitted and empty values retain the default TMPDIR behavior."""
+    workflow_path = tmp_path / "workflow.yaml"
+    workflow_path.write_text("", encoding="utf-8")
+
+    assert _resolve_event_log_dir(value, workflow_path) is None

tests/test_engine/test_event_log_dir.py

@@ -0,0 +1,69 @@
+"""Tests for configurable event log output directory."""
+
+import tempfile
+import time
+from pathlib import Path
+
+from conductor.engine.event_log import EventLogSubscriber
+from conductor.events import WorkflowEvent
+
+
+def test_runtime_config_event_log_dir():
+    """RuntimeConfig accepts event_log_dir field, defaults to None."""
+    from conductor.config.schema import RuntimeConfig
+
+    assert RuntimeConfig().event_log_dir is None
+    assert RuntimeConfig(event_log_dir="./logs").event_log_dir == "./logs"
+    assert RuntimeConfig(event_log_dir="/var/log/conductor").event_log_dir == ("/var/log/conductor")
+
+
+def test_default_writes_to_tmpdir():
+    """Without event_log_dir, writes to $TMPDIR/conductor/ (existing behavior)."""
+    sub = EventLogSubscriber("test_wf")
+    try:
+        assert sub.path.parent == Path(tempfile.gettempdir()) / "conductor"
+    finally:
+        sub.close()
+
+
+def test_custom_event_log_dir(tmp_path):
+    """With event_log_dir, writes to the specified directory."""
+    sub = EventLogSubscriber(
+        "test_wf",
+        event_log_dir=tmp_path / "logs",
+    )
+    try:
+        assert sub.path.parent == tmp_path / "logs"
+        assert sub.path.exists()
+
+        sub.on_event(
+            WorkflowEvent(
+                type="test",
+                timestamp=time.time(),
+                data={},
+            )
+        )
+        sub.close()
+
+        assert sub.path.read_text().strip()
+    finally:
+        if not sub._handle.closed:
+            sub.close()
+
+
+def test_existing_path_overrides_event_log_dir(tmp_path):
+    """On resume, existing_path takes precedence over event_log_dir."""
+    existing = tmp_path / "existing.events.jsonl"
+    existing.write_text("")
+
+    sub = EventLogSubscriber(
+        "test_wf",
+        existing_path=existing,
+        existing_run_id="abcd1234",
+        event_log_dir=tmp_path / "custom",
+    )
+
+    try:
+        assert sub.path == existing
+    finally:
+        sub.close()

tests/test_fleet/test_run_record_wiring.py

@@ -105,6 +105,7 @@ def _mock_config(name: str = "wiring-test", agent_names: list[str] | None = None
     mock_config.parallel = []
     mock_config.for_each = []
     mock_config.workflow.runtime.provider = ProviderSettings(name="copilot")
+    mock_config.workflow.runtime.event_log_dir = None
     mock_config.workflow.limits.max_iterations = 50
     mock_config.workflow.limits.timeout_seconds = None
     mock_config.workflow.limits.budget_usd = None
@@ -153,12 +154,16 @@ class TestRunWorkflowAsyncRunRecordWiring:
     """``run_workflow_async`` writes a discoverable record on every path."""
 
     async def test_fg_record_has_no_port_and_is_removed_on_completion(
-        self, tmp_path: Path, fleet_env: Path
+        self, tmp_path: Path, fleet_env: Path, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         from conductor.cli.run import run_workflow_async
 
-        wf_path = _write_workflow(tmp_path)
+        workflow_dir = tmp_path / "workflow"
+        workflow_dir.mkdir()
+        monkeypatch.chdir(tmp_path)
+        wf_path = _write_workflow(workflow_dir)
         mock_config = _mock_config()
+        mock_config.workflow.runtime.event_log_dir = "./event-logs"
         seen: dict[str, Any] = {}
 
         async def _fake_run(inputs: dict[str, Any]) -> dict[str, Any]:
@@ -200,6 +205,7 @@ async def _fake_run(inputs: dict[str, Any]) -> dict[str, Any]:
         # so this assertion actually exercises the distinction.
         assert record.workflow_name == wf_path.stem == "wf"
         assert Path(record.event_log_path).exists()
+        assert Path(record.event_log_path).parent == (workflow_dir / "event-logs").resolve()
         assert record.checkpoint_dir is not None
 
         # Removed once the run finishes.
@@ -535,12 +541,18 @@ async def _fake_resume(current_agent: str) -> dict[str, Any]:
         assert read_run_records() == []
 
     async def test_resume_record_removed_on_workflow_terminated(
-        self, tmp_path: Path, fleet_env: Path
+        self, tmp_path: Path, fleet_env: Path, monkeypatch: pytest.MonkeyPatch
     ) -> None:
         from conductor.cli.run import resume_workflow_async
 
-        wf_path, cp_path = _write_checkpoint_and_config(tmp_path)
+        workflow_dir = tmp_path / "workflow"
+        workflow_dir.mkdir()
+        monkeypatch.chdir(tmp_path)
+        wf_path, cp_path = _write_checkpoint_and_config(workflow_dir)
+        (workflow_dir / "original.events.jsonl").unlink()
         mock_config = _mock_config(agent_names=["a"])
+        mock_config.workflow.runtime.event_log_dir = "./event-logs"
+        expected_dir = (workflow_dir / "event-logs").resolve()
 
         terminate_exc = WorkflowTerminated(
             "bye",
@@ -550,7 +562,9 @@ async def test_resume_record_removed_on_workflow_terminated(
         )
 
         async def _fake_resume(current_agent: str) -> dict[str, Any]:
-            assert len(read_run_records()) == 1
+            records = read_run_records()
+            assert len(records) == 1
+            assert Path(records[0].event_log_path).parent == expected_dir
             raise terminate_exc
 
         mock_engine = MagicMock()