Skip to content
Merged
Show file tree
Hide file tree
Changes from 1 commit
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
67 changes: 67 additions & 0 deletions python/tests/codegen/test_get_flow.py
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import json
import textwrap
from pathlib import Path

Expand Down Expand Up @@ -144,6 +145,72 @@ def test_system_prompt_none(self, workspace):
config = _single_node(_flow(ws))["data"]["config"]
assert config["system_prompt"]["value"] is None

def test_voice_config_dict(self, workspace):
"""voice_config set in the source round-trips through get-flow."""
ws = workspace("""\
from timbal.core import Agent

agent = Agent(
name="a",
model="openai/gpt-4o-mini",
max_tokens=128,
voice_config={"voice": "CURRENT_VOICE", "tts_extra": {"auto_mode": True}},
)
""")
config = _single_node(_flow(ws))["data"]["config"]
assert config["voice_config"]["value"] == {
"voice": "CURRENT_VOICE",
"tts_extra": {"auto_mode": True},
}

def test_voice_config_absent(self, workspace):
"""Agents without voice_config still expose the key with a None value."""
ws = workspace("""\
from timbal.core import Agent

agent = Agent(name="a", model="openai/gpt-4o-mini", max_tokens=128)
""")
config = _single_node(_flow(ws))["data"]["config"]
assert "voice_config" in config
assert config["voice_config"]["value"] is None

def test_voice_config_callable(self, workspace):
"""A callable voice_config is rendered as an opaque placeholder."""
ws = workspace("""\
from timbal.core import Agent

def make_voice_config():
return {"voice": "CURRENT_VOICE"}

agent = Agent(
name="a",
model="openai/gpt-4o-mini",
max_tokens=128,
voice_config=make_voice_config,
)
""")
config = _single_node(_flow(ws))["data"]["config"]
assert config["voice_config"]["value"] == "<make_voice_config>"

def test_voice_config_instance(self, workspace):
"""A VoiceConfig instance is dumped to a JSON-safe dict."""
ws = workspace("""\
from timbal.core import Agent
from timbal.voice.config import VoiceConfig

agent = Agent(
name="a",
model="openai/gpt-4o-mini",
max_tokens=128,
voice_config=VoiceConfig(voice="my-voice"),
)
""")
config = _single_node(_flow(ws))["data"]["config"]
value = config["voice_config"]["value"]
assert isinstance(value, dict)
assert value["voice"] == "my-voice"
json.dumps(value) # must stay JSON-serialisable

def test_has_params_and_return(self, workspace):
ws = workspace("""\
from timbal.core import Agent
Expand Down
89 changes: 89 additions & 0 deletions python/tests/codegen/test_set_config.py
Original file line number Diff line number Diff line change
Expand Up @@ -484,6 +484,95 @@ def my_func():
assert tool.description == "Does something useful"


# ---------------------------------------------------------------------------
# Entry point shapes (annotated / aliased / factory assignments)
# ---------------------------------------------------------------------------


class TestEntryPointShapes:
def test_annotated_assignment(self, workspace):
"""``agent: Agent = Agent(...)`` is transformed like a plain assignment."""
ws = workspace("""\
from timbal.core import Agent

agent: Agent = Agent(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].voice_config == {"voice": "NEW_VOICE"}

def test_annotated_assignment_set_model(self, workspace):
ws = workspace("""\
from timbal.core import Agent

agent: Agent = Agent(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"model": "openai/gpt-4o"})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].model == "openai/gpt-4o"

def test_aliased_assignment_fails_loudly(self, workspace):
"""``agent = _agent`` cannot be transformed — must exit non-zero, not silently no-op."""
ws = workspace("""\
from timbal.core import Agent

_agent = Agent(name="a", model="openai/gpt-4o-mini")
agent = _agent
""")
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
result = _run_dry_fail(ws, "--config", config)
assert result.returncode != 0
assert "produced no changes" in result.stderr

def test_factory_call_fails_loudly(self, workspace):
"""``agent = build()`` must error instead of appending kwargs to build()."""
ws = workspace("""\
from timbal.core import Agent

def build():
return Agent(name="a", model="openai/gpt-4o-mini")

agent = build()
""")
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
result = _run_dry_fail(ws, "--config", config)
assert result.returncode != 0
assert "factory" in result.stderr

def test_noop_same_value_is_idempotent_success(self, workspace):
"""Setting a field to its current value is a success: the transformer
matched the entry point, the source was already in the desired state."""
ws = workspace("""\
from timbal.core import Agent

agent = Agent(name="a", model="openai/gpt-4o-mini")
""")
config = json.dumps({"model": "openai/gpt-4o-mini"})
output = _run_dry(ws, "--config", config)
ns = _exec_agent(output)
assert ns["agent"].model == "openai/gpt-4o-mini"

def test_write_mode_noop_leaves_file_untouched(self, workspace):
"""A failed no-op without --dry-run must not rewrite the source file."""
ws = workspace("""\
from timbal.core import Agent

_agent = Agent(name="a", model="openai/gpt-4o-mini")
agent = _agent
""")
original = (ws / "agent.py").read_text()
config = json.dumps({"voice_config": {"voice": "NEW_VOICE"}})
result = subprocess.run(
codegen_cmd("--path", str(ws), "set-config", "--config", config),
capture_output=True,
text=True,
)
assert result.returncode != 0
assert (ws / "agent.py").read_text() == original


# ---------------------------------------------------------------------------
# Workflow step config
# ---------------------------------------------------------------------------
Expand Down
53 changes: 41 additions & 12 deletions python/timbal/codegen/cst_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,27 +12,46 @@
ENTRY_POINT_TYPES = {"Agent", "Workflow"}


def _root_constructor_name(value: cst.BaseExpression) -> str | None:
"""Return the root constructor's class name for an assignment value.

Walks down chained method calls (e.g. ``Workflow(...).step(...).step(...)``)
to find the root constructor.
"""
call = value
while isinstance(call, cst.Call) and isinstance(call.func, cst.Attribute):
call = call.func.value
if isinstance(call, cst.Call) and isinstance(call.func, cst.Name):
return call.func.value
return None


def resolve_entry_point_type(tree: cst.Module, entry_point: str) -> str | None:
"""Return the constructor class name ('Agent' or 'Workflow') for the entry point variable.

Inspects top-level assignments to find `entry_point = ClassName(...)` and returns
the class name if it's a known entry point type. Returns None if not found.
Inspects top-level assignments (plain and annotated, e.g.
``agent: Agent = Agent(...)``) to find `entry_point = ClassName(...)` and
returns the class name if it's a known entry point type. Returns None if
not found.
"""
for stmt in tree.body:
if isinstance(stmt, cst.SimpleStatementLine):
for item in stmt.body:
if isinstance(item, cst.Assign):
for target in item.targets:
if isinstance(target.target, cst.Name) and target.target.value == entry_point:
# Walk down chained method calls (e.g. Workflow(...).step(...).step(...))
# to find the root constructor.
call = item.value
while isinstance(call, cst.Call) and isinstance(call.func, cst.Attribute):
call = call.func.value
if isinstance(call, cst.Call) and isinstance(call.func, cst.Name):
cls_name = call.func.value
if cls_name in ENTRY_POINT_TYPES:
return cls_name
cls_name = _root_constructor_name(item.value)
if cls_name in ENTRY_POINT_TYPES:
return cls_name
elif isinstance(item, cst.AnnAssign):
if (
isinstance(item.target, cst.Name)
and item.target.value == entry_point
and item.value is not None
):
cls_name = _root_constructor_name(item.value)
if cls_name in ENTRY_POINT_TYPES:
return cls_name
return None


Expand Down Expand Up @@ -141,7 +160,11 @@ def build_cst_value(value: object) -> cst.BaseExpression:


def collect_assignments(tree: cst.Module) -> dict[str, cst.Call]:
"""Build a map of variable_name -> Call node for all top-level assignments."""
"""Build a map of variable_name -> Call node for all top-level assignments.

Covers plain assignments (``x = Call(...)``) and annotated assignments
(``x: T = Call(...)``).
"""
result = {}
for stmt in tree.body:
if isinstance(stmt, cst.SimpleStatementLine):
Expand All @@ -150,6 +173,12 @@ def collect_assignments(tree: cst.Module) -> dict[str, cst.Call]:
for target in item.targets:
if isinstance(target.target, cst.Name):
result[target.target.value] = item.value
elif (
isinstance(item, cst.AnnAssign)
and isinstance(item.target, cst.Name)
and isinstance(item.value, cst.Call)
):
result[item.target.value] = item.value
return result


Expand Down
24 changes: 23 additions & 1 deletion python/timbal/codegen/transformers/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -307,4 +307,26 @@ def apply_operation(workspace_path: str | Path, operation: str, **kwargs) -> str
if getattr(transformer, "needs_reorder", False):
code = reorder_step_calls(code, spec.target)

return format_code(code, spec.path)
formatted = format_code(code, spec.path)

# Every transformer operation is a mutation, so identical output is
# suspicious — it usually means the transformer never matched its target
# (e.g. an aliased entry point like ``agent = _agent``). Fail loudly
# instead of reporting a phantom success, except when:
# - the transformer has intentional no-op semantics (``allow_noop = True``,
# e.g. add-edge deduplication, removing an already-absent tool), or
# - the transformer affirmatively found its target (``matched = True``),
# in which case an unchanged file just means the source was already in
# the desired state (idempotent save).
if (
formatted == source
and not getattr(transformer, "allow_noop", False)
and getattr(transformer, "matched", None) is not True
):
raise ValueError(
f"{operation.replace('_', '-')} produced no changes to {spec.path.name}. "
f"The source may use a shape the transformer does not recognize "
f"(e.g. an aliased entry point assignment)."
)
Comment thread
cursor[bot] marked this conversation as resolved.
Comment thread
cursor[bot] marked this conversation as resolved.

return formatted
3 changes: 3 additions & 0 deletions python/timbal/codegen/transformers/add_edge.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,9 @@ class EdgeAdder(cst.CSTTransformer):
"""Add an ordering or conditional edge between two workflow steps."""

needs_reorder = True
# Adding an edge that already exists deduplicates to an unchanged file —
# that is a legitimate success, not a silent failure.
allow_noop = True

def __init__(
self,
Expand Down
4 changes: 4 additions & 0 deletions python/timbal/codegen/transformers/add_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -320,6 +320,10 @@ def _name_collision_error(runtime_name: str) -> ValueError:


class MCPAdder(cst.CSTTransformer):
# Re-adding an identical server is an idempotent success (same-named
# assignment replaced in place, file unchanged), not a silent failure.
allow_noop = True

def __init__(self, assignments: dict[str, cst.Call], *, target: str, servers: list[tuple[str, dict]]):
self.assignments = assignments
self.target = target
Expand Down
4 changes: 4 additions & 0 deletions python/timbal/codegen/transformers/add_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -161,6 +161,10 @@ def run(entry_point: str, args: argparse.Namespace, *, tree: cst.Module | None =


class ToolAdder(cst.CSTTransformer):
# Re-adding an existing tool is an idempotent success (no duplicate entry,
# file unchanged), not a silent failure.
allow_noop = True

Comment thread
cursor[bot] marked this conversation as resolved.
def __init__(
self,
entry_point: str,
Expand Down
3 changes: 3 additions & 0 deletions python/timbal/codegen/transformers/remove_step.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,9 @@ def run(entry_point: str, args: argparse.Namespace, *, tree: cst.Module | None =


class StepRemover(cst.CSTTransformer):
# Removing a step that is already absent is an idempotent success.
allow_noop = True

def __init__(self, entry_point: str, step_name: str, assignments: dict[str, cst.Call]):
self.entry_point = entry_point
self.step_name = step_name
Expand Down
3 changes: 3 additions & 0 deletions python/timbal/codegen/transformers/remove_tool.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,6 +35,9 @@ def run(entry_point: str, args: argparse.Namespace, *, tree: cst.Module | None =


class ToolRemover(cst.CSTTransformer):
# Removing a tool that is already absent is an idempotent success.
allow_noop = True

Comment thread
cursor[bot] marked this conversation as resolved.
def __init__(self, entry_point: str, tool_name: str, assignments: dict[str, cst.Call]):
self.entry_point = entry_point
self.tool_name = tool_name
Expand Down
Loading
Loading