Skip to content
Closed
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
128 changes: 103 additions & 25 deletions src/nooa/strategies/codeact.py
Original file line number Diff line number Diff line change
Expand Up @@ -1803,8 +1803,8 @@ def _handle_return_result(
normalized_args["result"] = parsed
else:
# Constructor-call coercion: if the string looks like a Python
# constructor call (e.g. "Answer(answer=1, reason='...')"), eval it
# in the session namespace where the type is available.
# constructor call (e.g. "Answer(answer=1, reason='...')"),
# parse its arguments without evaluating against session state.
coerced = self._maybe_eval_constructor_string(
result_str, return_type, session
)
Expand Down Expand Up @@ -2009,13 +2009,87 @@ def _corrected_return_args(self, validated: Any, original_args: dict) -> dict:

return {"result": corrected_result}

def _json_detached_session_value(self, name: str, session: Any) -> Any:
"""Return a JSON-round-tripped session value if it is a plain value."""
session_locals = getattr(session, "session_locals", {})
if name not in session_locals:
raise ValueError(f"Unknown constructor argument reference: {name}")

value = session_locals[name]
if not self._is_plain_json_value(value):
raise TypeError(f"Session value {name!r} is not a plain JSON value")

return json.loads(json.dumps(value, allow_nan=False))

def _is_plain_json_value(self, value: Any, seen: set[int] | None = None) -> bool:
"""True for exact JSON-compatible builtin values, excluding subclasses."""
if value is None or type(value) in {bool, int, float, str}:
return True

if seen is None:
seen = set()

if type(value) in {list, tuple}:
value_id = id(value)
if value_id in seen:
return False
seen.add(value_id)
try:
return all(self._is_plain_json_value(item, seen) for item in value)
finally:
seen.remove(value_id)

if type(value) is dict:
value_id = id(value)
if value_id in seen:
return False
seen.add(value_id)
try:
return all(
type(key) is str and self._is_plain_json_value(item, seen)
for key, item in value.items()
)
finally:
seen.remove(value_id)

return False

def _safe_constructor_arg_value(self, node: ast.AST, session: Any) -> Any:
"""Parse one constructor argument without executing Python code."""
if isinstance(node, ast.Call):
raise ValueError("Constructor argument calls are not allowed")

if isinstance(node, ast.Name):
return self._json_detached_session_value(node.id, session)

if isinstance(node, ast.List):
return [self._safe_constructor_arg_value(item, session) for item in node.elts]

if isinstance(node, ast.Tuple):
return tuple(self._safe_constructor_arg_value(item, session) for item in node.elts)

if isinstance(node, ast.Set):
return {self._safe_constructor_arg_value(item, session) for item in node.elts}

if isinstance(node, ast.Dict):
result: dict[Any, Any] = {}
for key_node, value_node in zip(node.keys, node.values, strict=True):
if key_node is None:
raise ValueError("Dictionary unpacking is not allowed")
key = self._safe_constructor_arg_value(key_node, session)
result[key] = self._safe_constructor_arg_value(value_node, session)
return result

return ast.literal_eval(node)

def _maybe_eval_constructor_string(self, value: str, return_type: Any, session: Any) -> Any:
"""Eval a string that looks like a Python constructor call.
"""Safely coerce a string that looks like a constructor call.

Detects patterns like 'ClassName(field=value, ...)' where ClassName
matches the expected return type. Evaluates in the session namespace
so the type is available. Returns the constructed object on success,
or the original string on failure.
matches the expected return type. Constructor arguments may contain
Python literals or references to plain JSON-compatible session values.
Session values are copied through JSON before use, and executable
expressions such as function calls or attribute access are rejected.

This handles a common LLM failure mode where the model calls
return_result as a tool with the constructor as a string argument
Expand All @@ -2024,41 +2098,45 @@ def _maybe_eval_constructor_string(self, value: str, return_type: Any, session:
"""
stripped = value.strip()

# Must look like a constructor call: Identifier(...)
# Quick check before parsing
paren_idx = stripped.find("(")
if paren_idx <= 0 or not stripped.endswith(")"):
try:
expression = ast.parse(stripped, mode="eval")
except SyntaxError:
return value

if not isinstance(expression.body, ast.Call):
return value

candidate_name = stripped[:paren_idx].strip()
if not candidate_name.isidentifier():
call = expression.body
if not isinstance(call.func, ast.Name):
return value

# Check that the candidate name matches the expected return type
# or is available in session locals
type_name = getattr(return_type, "__name__", None)
if candidate_name != type_name and candidate_name not in session.session_locals:
base_return_type, _ = self._extract_annotated_description(return_type)
type_name = getattr(base_return_type, "__name__", None)
candidate_name = call.func.id
if candidate_name != type_name or not isinstance(base_return_type, type):
return value

# Build eval namespace: session locals + the return type itself (which may
# live in module globals rather than session_locals).
eval_ns = dict(session.session_locals)
if type_name and type_name not in eval_ns and isinstance(return_type, type):
eval_ns[type_name] = return_type
if any(keyword.arg is None for keyword in call.keywords):
return value

# Try to eval in the combined namespace
try:
result = eval(stripped, {"__builtins__": {}}, eval_ns) # noqa: S307
args = [self._safe_constructor_arg_value(arg, session) for arg in call.args]
kwargs = {
keyword.arg: self._safe_constructor_arg_value(keyword.value, session)
for keyword in call.keywords
if keyword.arg is not None
}
result = base_return_type(*args, **kwargs)
get_harness_metrics().constructor_string_coerced(candidate_name)
logger.debug(
"[CODEACT] Coerced constructor-call string %r into %s instance",
"[CODEACT] Safely coerced constructor-call string %r into %s instance",
stripped[:80],
type(result).__name__,
)
return result
except Exception:
logger.debug(
"[CODEACT] Failed to eval constructor string %r, returning as-is",
"[CODEACT] Failed to safely coerce constructor string %r, returning as-is",
stripped[:80],
)
return value
Expand Down
86 changes: 73 additions & 13 deletions tests/strategies/test_codeact_pure_python_coverage.py
Original file line number Diff line number Diff line change
Expand Up @@ -4177,7 +4177,7 @@ def test_generation_id_none_check_is_documented(self):


class TestMaybeEvalConstructorString:
"""Tests for _maybe_eval_constructor_string constructor-call coercion."""
"""Tests for safe _maybe_eval_constructor_string constructor-call coercion."""

def _make_session(self, **locals_):
"""Create a mock session with given session_locals."""
Expand All @@ -4186,7 +4186,7 @@ def _make_session(self, **locals_):
return session

def test_basic_constructor_with_type_in_locals(self):
"""Should eval 'Answer(answer=1, reason="test")' when Answer is in locals."""
"""Should coerce 'Answer(answer=1, reason="test")' when Answer is return type."""
from pydantic import BaseModel

class Answer(BaseModel):
Expand All @@ -4203,7 +4203,7 @@ class Answer(BaseModel):
assert result.reason == "the minimum"

def test_constructor_with_type_injected_from_return_type(self):
"""Should inject return_type into eval ns when not in session_locals."""
"""Should use return_type when not in session_locals."""
from pydantic import BaseModel

class MyResult(BaseModel):
Expand All @@ -4220,7 +4220,7 @@ class MyResult(BaseModel):
assert result.note == "computed"

def test_constructor_with_variable_reference_in_args(self):
"""Should resolve variables from session_locals inside constructor args."""
"""Should resolve plain JSON values from session_locals inside constructor args."""
from pydantic import BaseModel

class Answer(BaseModel):
Expand All @@ -4236,6 +4236,21 @@ class Answer(BaseModel):
assert result.answer == 7
assert result.reason == "found it"

def test_constructor_with_session_value_is_json_detached(self):
"""Should copy plain session values through JSON before constructor use."""
from pydantic import BaseModel

class Answer(BaseModel):
payload: dict[str, list[int]]

payload = {"numbers": [1, 2, 3]}
strat = CodeActStrategy()
session = self._make_session(Answer=Answer, payload=payload)
result = strat._maybe_eval_constructor_string("Answer(payload=payload)", Answer, session)
assert isinstance(result, Answer)
assert result.payload == {"numbers": [1, 2, 3]}
assert result.payload is not payload

def test_non_constructor_string_returns_as_is(self):
"""Plain string should be returned unchanged."""
strat = CodeActStrategy()
Expand All @@ -4253,12 +4268,12 @@ def test_no_parens_returns_as_is(self):
def test_unknown_type_returns_as_is(self):
"""Constructor with unknown type name should return as-is."""
strat = CodeActStrategy()
session = self._make_session()
session = self._make_session(Unknown=lambda **kwargs: "should not run")
result = strat._maybe_eval_constructor_string("Unknown(x=1)", object, session)
assert result == "Unknown(x=1)"

def test_eval_failure_returns_as_is(self):
"""If eval raises, return original string."""
def test_constructor_coercion_failure_returns_as_is(self):
"""If safe argument parsing fails, return original string."""
from pydantic import BaseModel

class Answer(BaseModel):
Expand All @@ -4267,15 +4282,15 @@ class Answer(BaseModel):

strat = CodeActStrategy()
session = self._make_session(Answer=Answer)
# Missing required field should raise ValidationError in eval
# Missing variable should fail safe argument parsing.
result = strat._maybe_eval_constructor_string(
'Answer(answer="not_an_int_but_coerced", reason=missing_var)', Answer, session
)
# missing_var is not in session_locals → NameError → returns as-is
# missing_var is not in session_locals, so the original value is preserved.
assert result == 'Answer(answer="not_an_int_but_coerced", reason=missing_var)'

def test_nested_parens_work(self):
"""Nested parens in args should be handled by eval."""
def test_nested_calls_return_as_is(self):
"""Nested calls in args should be rejected instead of executed."""
from pydantic import BaseModel

class Answer(BaseModel):
Expand All @@ -4287,8 +4302,53 @@ class Answer(BaseModel):
result = strat._maybe_eval_constructor_string(
'Answer(answer=min(3, 1, 2), reason="picked smallest")', Answer, session
)
assert isinstance(result, Answer)
assert result.answer == 1
assert result == 'Answer(answer=min(3, 1, 2), reason="picked smallest")'

def test_constructor_does_not_call_session_object(self):
"""Pre-planted callable session objects must not be invoked."""
from pydantic import BaseModel

class Answer(BaseModel):
answer: str
reason: str

class Trap:
called = False

def __call__(self):
self.called = True
return "owned"

trap = Trap()
strat = CodeActStrategy()
session = self._make_session(Answer=Answer, payload=trap)
result = strat._maybe_eval_constructor_string(
'Answer(answer=payload(), reason="should not execute")', Answer, session
)
assert result == 'Answer(answer=payload(), reason="should not execute")'
assert trap.called is False

def test_constructor_rejects_non_plain_session_object(self):
"""Session references to arbitrary objects should not be serialized or used."""
from pydantic import BaseModel

class Answer(BaseModel):
answer: object
reason: str

class Trap:
def __iter__(self):
raise AssertionError("object serialization should not be attempted")

def __repr__(self):
raise AssertionError("object repr should not be needed")

strat = CodeActStrategy()
session = self._make_session(Answer=Answer, payload=Trap())
result = strat._maybe_eval_constructor_string(
'Answer(answer=payload, reason="should not serialize")', Answer, session
)
assert result == 'Answer(answer=payload, reason="should not serialize")'

def test_non_identifier_prefix_returns_as_is(self):
"""String starting with non-identifier before parens should return as-is."""
Expand Down