Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 23 additions & 2 deletions src/conductor/engine/context.py
Original file line number Diff line number Diff line change
Expand Up @@ -86,6 +86,18 @@ class WorkflowContext:
workflow_inputs: dict[str, Any] = field(default_factory=dict)
"""Inputs provided at workflow start."""

workflow_dir: str = ""
"""Directory containing the workflow YAML file (resolved absolute path).
Available in templates as ``{{ workflow.dir }}``."""

workflow_file: str = ""
"""Absolute path to the workflow YAML file.
Available in templates as ``{{ workflow.file }}``."""

workflow_name: str = ""
"""Name of the workflow from the YAML config.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Resume drops these fields.

The new workflow_dir/file/name fields aren't included in to_dict() (line 486) or restored by from_dict() (line 502). On resume, cli/run.py:1538 calls engine.set_context(restored_context), which replaces the engine's context (built with the metadata in __init__) with one that has the metadata wiped to empty strings. After resume, {{ workflow.dir/file/name }} will silently disappear from templates — exactly the registry-based script-path scenario this PR exists for.

Recommended fix: repopulate metadata inside WorkflowEngine.set_context() from self.workflow_path / self.config. Keeps absolute paths out of checkpoint files (which would otherwise become stale if the workflow moves) and keeps the source of truth co-located with the engine that knows the path.

Please also add a regression test in tests/test_engine/test_resume.py (or similar) asserting that workflow.dir survives a checkpoint round-trip + resume.

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, you're right, the round trip drops them and set_context was happily clobbering the engine's view. Fixed in eaf6c9f: set_context now repopulates workflow_dir/file/name from self.workflow_path and self.config after the replace, so the engine stays the source of truth and we keep absolute paths out of the checkpoint. Added test_set_context_repopulates_workflow_metadata that round-trips through to_dict/from_dict, calls set_context, and verifies via build_for_agent that templates actually resolve. Also covered the no-workflow_path case in test_set_context_without_workflow_path_still_sets_name.

Available in templates as ``{{ workflow.name }}``."""

agent_outputs: dict[str, dict[str, Any]] = field(default_factory=dict)
"""Outputs from executed agents, keyed by agent name."""

Expand Down Expand Up @@ -176,6 +188,15 @@ def build_for_agent(
Raises:
KeyError: If explicit mode is used and a required (non-optional) input is missing.
"""
# Build workflow metadata available in all modes
workflow_meta: dict[str, Any] = {}
if self.workflow_dir:
workflow_meta["dir"] = self.workflow_dir
if self.workflow_file:
workflow_meta["file"] = self.workflow_file
if self.workflow_name:
workflow_meta["name"] = self.workflow_name

# For explicit mode, start with empty workflow inputs
# For other modes, include all workflow inputs
if mode == "explicit":
Expand All @@ -185,7 +206,7 @@ def build_for_agent(
self.workflow_inputs.copy() if agent_type in _LOCAL_RENDER_AGENT_TYPES else {}
)
ctx: dict[str, Any] = {
"workflow": {"input": initial_workflow_inputs},
"workflow": {"input": initial_workflow_inputs, **workflow_meta},
"context": {
"iteration": self.current_iteration,
"history": self.execution_history.copy(),
Expand All @@ -196,7 +217,7 @@ def build_for_agent(
self._add_explicit_input(ctx, input_ref)
else:
ctx = {
"workflow": {"input": self.workflow_inputs.copy()},
"workflow": {"input": self.workflow_inputs.copy(), **workflow_meta},
"context": {
"iteration": self.current_iteration,
"history": self.execution_history.copy(),
Expand Down
18 changes: 17 additions & 1 deletion src/conductor/engine/workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -326,7 +326,11 @@ def __init__(
self._run_context = run_context or RunContext()
self._run_id = self._run_context.run_id
self._log_file = self._run_context.log_file
self.context = WorkflowContext()
self.context = WorkflowContext(
workflow_dir=str(Path(workflow_path).resolve().parent) if workflow_path else "",
workflow_file=str(Path(workflow_path).resolve()) if workflow_path else "",
workflow_name=config.workflow.name,
)
self.renderer = TemplateRenderer()
self.router = Router()
self.limits = LimitEnforcer(
Expand Down Expand Up @@ -893,10 +897,22 @@ def set_context(self, context: WorkflowContext) -> None:
Used by the CLI resume path to inject context reconstructed from
a checkpoint file.

Workflow metadata (``workflow_dir``, ``workflow_file``, ``workflow_name``)
is repopulated from the engine's ``workflow_path`` and ``config`` rather
than the restored context. Restored contexts come from
``WorkflowContext.from_dict()``, which intentionally omits absolute path
metadata to keep checkpoint files portable across machines and
relocatable when workflows move. The engine, which knows the current
path, is the source of truth.

Args:
context: A WorkflowContext restored via ``WorkflowContext.from_dict()``.
"""
self.context = context
if self.workflow_path is not None:
self.context.workflow_dir = str(Path(self.workflow_path).resolve().parent)
self.context.workflow_file = str(Path(self.workflow_path).resolve())
self.context.workflow_name = self.config.workflow.name

def set_limits(self, limits: LimitEnforcer) -> None:
"""Replace the engine's limit enforcer with a restored one.
Expand Down
47 changes: 47 additions & 0 deletions tests/test_engine/test_context.py
Original file line number Diff line number Diff line change
Expand Up @@ -30,6 +30,9 @@ def test_init_default_values(self) -> None:
assert ctx.agent_outputs == {}
assert ctx.current_iteration == 0
assert ctx.execution_history == []
assert ctx.workflow_dir == ""
assert ctx.workflow_file == ""
assert ctx.workflow_name == ""

def test_set_workflow_inputs(self) -> None:
"""Test setting workflow inputs."""
Expand Down Expand Up @@ -151,6 +154,50 @@ def test_last_only_mode_empty_history(self) -> None:
assert "context" in agent_ctx


class TestWorkflowContextMetadata:
"""Tests for workflow metadata (dir, file, name) in context."""

def test_workflow_dir_file_name_in_accumulate_context(self) -> None:
"""Test workflow.dir, workflow.file, workflow.name available in accumulate mode."""
ctx = WorkflowContext(
workflow_dir="/home/user/workflows",
workflow_file="/home/user/workflows/main.yaml",
workflow_name="my-workflow",
)
ctx.set_workflow_inputs({"key": "val"})

agent_ctx = ctx.build_for_agent("agent", [], mode="accumulate")

assert agent_ctx["workflow"]["dir"] == "/home/user/workflows"
assert agent_ctx["workflow"]["file"] == "/home/user/workflows/main.yaml"
assert agent_ctx["workflow"]["name"] == "my-workflow"
assert agent_ctx["workflow"]["input"] == {"key": "val"}

def test_workflow_metadata_in_explicit_mode(self) -> None:
"""Test workflow.dir/file/name available in explicit mode (not filtered)."""
ctx = WorkflowContext(
workflow_dir="/registry/twig",
workflow_file="/registry/twig/sdlc.yaml",
workflow_name="twig-sdlc",
)

agent_ctx = ctx.build_for_agent("agent", [], mode="explicit")

assert agent_ctx["workflow"]["dir"] == "/registry/twig"
assert agent_ctx["workflow"]["file"] == "/registry/twig/sdlc.yaml"
assert agent_ctx["workflow"]["name"] == "twig-sdlc"

def test_empty_metadata_omitted(self) -> None:
"""Test that empty workflow metadata fields are not included."""
ctx = WorkflowContext()

agent_ctx = ctx.build_for_agent("agent", [], mode="accumulate")

assert "dir" not in agent_ctx["workflow"]
assert "file" not in agent_ctx["workflow"]
assert "name" not in agent_ctx["workflow"]

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestion: add an engine-level wiring test.

These tests verify WorkflowContext.build_for_agent() formatting given pre-set fields, but nothing asserts that WorkflowEngine.__init__(config, workflow_path=...) actually wires the path through into those fields (the change at workflow.py:311-315). A small test in tests/test_engine/test_workflow.py like:

def test_engine_populates_workflow_metadata(tmp_path):
    wf_file = tmp_path / "wf.yaml"
    wf_file.write_text("...")  # or use a fixture config
    engine = WorkflowEngine(config, workflow_path=wf_file)
    assert engine.context.workflow_dir == str(tmp_path.resolve())
    assert engine.context.workflow_file == str(wf_file.resolve())
    assert engine.context.workflow_name == config.workflow.name

would catch any future regression in the wiring (e.g., if someone refactors __init__ and reverts to the old WorkflowContext()).

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fair, there was no guard that __init__ was actually wiring the path into the context. Added test_engine_populates_workflow_metadata and test_engine_workflow_metadata_empty_without_path in tests/test_engine/test_workflow.py to lock that down. Same commit, eaf6c9f.



class TestWorkflowContextExplicitMode:
"""Tests for explicit context mode."""

Expand Down
52 changes: 52 additions & 0 deletions tests/test_engine/test_resume.py
Original file line number Diff line number Diff line change
Expand Up @@ -525,3 +525,55 @@ def test_set_limits_replaces_limits(self) -> None:
assert engine.limits.current_iteration == 5
assert engine.limits.max_iterations == 20
assert engine.limits.timeout_seconds == 120

def test_set_context_repopulates_workflow_metadata(self, tmp_path: Path) -> None:
"""Resume must not drop workflow_dir/file/name from the context.

``WorkflowContext.from_dict()`` intentionally omits absolute path
metadata so checkpoint files stay portable. The engine, which knows
the current ``workflow_path`` and ``config``, must repopulate those
fields when ``set_context()`` swaps in the restored context.

Regression test for the resume path: without this, ``{{ workflow.dir }}``
silently disappears from templates after resume — exactly the
registry-based script-path scenario this feature exists for.
"""
wf_path = _write_workflow(tmp_path)
config = _multi_agent_config()
engine = WorkflowEngine(config, workflow_path=wf_path)

# Simulate a context restored from checkpoint: round-trip through
# to_dict/from_dict, which strips the metadata.
restored = WorkflowContext.from_dict(engine.context.to_dict())
assert restored.workflow_dir == ""
assert restored.workflow_file == ""
assert restored.workflow_name == ""

engine.set_context(restored)

assert engine.context.workflow_dir == str(tmp_path.resolve())
assert engine.context.workflow_file == str(wf_path.resolve())
assert engine.context.workflow_name == config.workflow.name

# End-to-end: the restored context must render workflow metadata
# in templates via build_for_agent.
agent_ctx = engine.context.build_for_agent("synthesizer", [], mode="accumulate")
assert agent_ctx["workflow"]["dir"] == str(tmp_path.resolve())
assert agent_ctx["workflow"]["file"] == str(wf_path.resolve())
assert agent_ctx["workflow"]["name"] == config.workflow.name

def test_set_context_without_workflow_path_still_sets_name(self) -> None:
"""When the engine has no workflow_path, only name is repopulated.

Path-derived fields stay empty (and are omitted from rendered context
per ``build_for_agent`` semantics).
"""
config = _multi_agent_config()
engine = WorkflowEngine(config) # no workflow_path

restored = WorkflowContext()
engine.set_context(restored)

assert engine.context.workflow_dir == ""
assert engine.context.workflow_file == ""
assert engine.context.workflow_name == config.workflow.name
33 changes: 33 additions & 0 deletions tests/test_engine/test_workflow.py
Original file line number Diff line number Diff line change
Expand Up @@ -199,6 +199,39 @@ def mock_handler(agent, prompt, context):
assert received_contexts[1][0] == "executor"
assert received_contexts[1][1]["planner"]["output"]["plan"] == "the plan"

def test_engine_populates_workflow_metadata(
self, tmp_path, simple_workflow_config: WorkflowConfig
) -> None:
"""``WorkflowEngine.__init__`` wires ``workflow_path`` into context fields.

Guards against a regression where someone refactors ``__init__`` and
reverts to a bare ``WorkflowContext()``, silently dropping
``workflow.dir``/``workflow.file``/``workflow.name`` from templates.
"""
wf_file = tmp_path / "wf.yaml"
wf_file.write_text("name: test\n")

engine = WorkflowEngine(simple_workflow_config, workflow_path=wf_file)

assert engine.context.workflow_dir == str(tmp_path.resolve())
assert engine.context.workflow_file == str(wf_file.resolve())
assert engine.context.workflow_name == simple_workflow_config.workflow.name

def test_engine_workflow_metadata_empty_without_path(
self, simple_workflow_config: WorkflowConfig
) -> None:
"""Without ``workflow_path``, path-derived fields stay empty.

Empty strings are omitted from the rendered context (see
``WorkflowContext.build_for_agent``), so this preserves the existing
no-pollution behaviour for path-less engines (e.g., test fixtures).
"""
engine = WorkflowEngine(simple_workflow_config)

assert engine.context.workflow_dir == ""
assert engine.context.workflow_file == ""
assert engine.context.workflow_name == simple_workflow_config.workflow.name


class TestWorkflowEngineContextModes:
"""Tests for different context accumulation modes."""
Expand Down
Loading