From 13cba6c8e29f56c5edeb3d1e285aa543311ee717 Mon Sep 17 00:00:00 2001 From: nskun <8086.nk@gmail.com> Date: Fri, 21 Aug 2026 11:48:53 +0000 Subject: [PATCH 1/3] feat(runtime): add event_log_dir to configure event log output directory --- src/conductor/cli/run.py | 12 ++++- src/conductor/config/schema.py | 10 ++++ src/conductor/engine/event_log.py | 10 ++-- tests/test_engine/test_event_log_dir.py | 69 +++++++++++++++++++++++++ 4 files changed, 96 insertions(+), 5 deletions(-) create mode 100644 tests/test_engine/test_event_log_dir.py diff --git a/src/conductor/cli/run.py b/src/conductor/cli/run.py index d6dc70bb..a480717b 100644 --- a/src/conductor/cli/run.py +++ b/src/conductor/cli/run.py @@ -2054,7 +2054,13 @@ 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) + _eld = config.workflow.runtime.event_log_dir + _event_log_dir = Path(_eld).resolve() if _eld else None + + 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 +2771,14 @@ async def resume_workflow_async( # appends to it and reuses run_id; otherwise it generates fresh. from conductor.engine.event_log import EventLogSubscriber + _eld = config.workflow.runtime.event_log_dir + _event_log_dir = Path(_eld).resolve() if _eld else None + 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) diff --git a/src/conductor/config/schema.py b/src/conductor/config/schema.py index 889b51c4..0b2de746 100644 --- a/src/conductor/config/schema.py +++ b/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. diff --git a/src/conductor/engine/event_log.py b/src/conductor/engine/event_log.py index 1fb56a51..d3b2e6a5 100644 --- a/src/conductor/engine/event_log.py +++ b/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 diff --git a/tests/test_engine/test_event_log_dir.py b/tests/test_engine/test_event_log_dir.py new file mode 100644 index 00000000..2020b23a --- /dev/null +++ b/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() From 85b4672666a0cd6980317da35a4b0dd118be466e Mon Sep 17 00:00:00 2001 From: nskun <8086.nk@gmail.com> Date: Mon, 24 Aug 2026 01:16:48 +0000 Subject: [PATCH 2/3] chore: retrigger CI From a807f0bed5c919e0a5c07b5ed4d3e2ad69928f39 Mon Sep 17 00:00:00 2001 From: nskun <8086.nk@gmail.com> Date: Mon, 24 Aug 2026 10:53:01 +0000 Subject: [PATCH 3/3] fix(runtime): resolve event_log_dir relative to workflow file --- src/conductor/cli/run.py | 26 ++++++-- tests/test_cli/test_event_log_dir.py | 77 ++++++++++++++++++++++ tests/test_fleet/test_run_record_wiring.py | 24 +++++-- 3 files changed, 118 insertions(+), 9 deletions(-) create mode 100644 tests/test_cli/test_event_log_dir.py diff --git a/src/conductor/cli/run.py b/src/conductor/cli/run.py index a480717b..11b5fa04 100644 --- a/src/conductor/cli/run.py +++ b/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,8 +2068,10 @@ async def run_workflow_async( # Start JSONL event log subscriber (always-on structured diagnostics) from conductor.engine.event_log import EventLogSubscriber - _eld = config.workflow.runtime.event_log_dir - _event_log_dir = Path(_eld).resolve() if _eld else None + _event_log_dir = _resolve_event_log_dir( + config.workflow.runtime.event_log_dir, + workflow_path, + ) event_log_subscriber = EventLogSubscriber( config.workflow.name, @@ -2771,8 +2787,10 @@ async def resume_workflow_async( # appends to it and reuses run_id; otherwise it generates fresh. from conductor.engine.event_log import EventLogSubscriber - _eld = config.workflow.runtime.event_log_dir - _event_log_dir = Path(_eld).resolve() if _eld else None + _event_log_dir = _resolve_event_log_dir( + config.workflow.runtime.event_log_dir, + resolved_workflow_path, + ) event_log_subscriber = EventLogSubscriber( config.workflow.name, diff --git a/tests/test_cli/test_event_log_dir.py b/tests/test_cli/test_event_log_dir.py new file mode 100644 index 00000000..70e330f1 --- /dev/null +++ b/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 diff --git a/tests/test_fleet/test_run_record_wiring.py b/tests/test_fleet/test_run_record_wiring.py index 3ace7b55..66475740 100644 --- a/tests/test_fleet/test_run_record_wiring.py +++ b/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()