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
50 changes: 48 additions & 2 deletions src/envs/fleet_env/task_env.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,31 @@
clear_task_context,
)

# Synthetic tool injected by the harness (not from MCP).
# Mirrors orchestrator/temporal/workflows/constants.py → ANSWER_SUBMISSION_TOOL.
SUBMIT_FINAL_ANSWER_TOOL = {
"type": "function",
"function": {
"name": "submit_final_answer",
"description": (
"Submit your final answer to complete the task. Use this when you "
"have finished the task and want to provide your answer for "
"verification. If the requested answer asks for json, then write "
"your response in the answer field using json brackets."
),
"parameters": {
"type": "object",
"properties": {
"answer": {
"type": "string",
"description": "Your final answer",
}
},
"required": ["answer"],
},
},
}


def _is_tool_error(result: Any) -> Tuple[bool, Optional[str]]:
"""Check if a tool result indicates an error.
Expand Down Expand Up @@ -132,6 +157,7 @@ def __init__(
self._tools_cache: Optional[List[Dict]] = None
self._reward_computed = False
self.final_reward: Optional[float] = None
self._submitted_answer: Optional[str] = None

# Set telemetry context so init failures are tracked with full context
set_task_context(
Expand Down Expand Up @@ -288,6 +314,7 @@ async def reset_async(self, seed: Optional[int] = None) -> Dict[str, Any]:
self._done = False
self._reward_computed = False
self.final_reward = None
self._submitted_answer = None

# Reset the environment (use short timeout to avoid blocking on broken manager APIs)
# reset() failure is non-fatal — env is up, just the manager API timed out
Expand Down Expand Up @@ -385,6 +412,13 @@ async def reset_async(self, seed: Optional[int] = None) -> Dict[str, Any]:
self._rollout_completed_emitted = True
raise RuntimeError("No tools available after filtering")

# Inject submit_final_answer synthetic tool for tool_use tasks whose
# prompt references it. This mirrors the harness's ANSWER_SUBMISSION_TOOL
# so that models can submit answers during SkyRL training exactly as
# they would in a Fleet harness session.
if self.modality == "tool_use" and "submit_final_answer" in self.prompt:
self._tools_cache.append(SUBMIT_FINAL_ANSWER_TOOL)

Copy link
Copy Markdown

Choose a reason for hiding this comment

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

Shared mutable dict reference risks cross-instance corruption

Low Severity

SUBMIT_FINAL_ANSWER_TOOL is a mutable module-level dict that gets appended by reference to _tools_cache. The tools list is then exposed in observations via obs["tools"]. If any downstream consumer (e.g., training framework, logging, serialization) mutates a tool dict in-place, it would corrupt the shared constant for all future FleetTaskEnv instances. A shallow copy (e.g., copy.deepcopy(SUBMIT_FINAL_ANSWER_TOOL)) at append time would prevent this.

Additional Locations (1)
Fix in Cursor Fix in Web


# Build observation with cached tools
obs = {
"prompt": self.prompt,
Expand Down Expand Up @@ -467,7 +501,14 @@ async def step_async(
tool_params = action.get("params", {})
tool_result = None

if tool_name:
if tool_name == "submit_final_answer":
# Synthetic tool — handled locally, not routed to MCP.
self._submitted_answer = tool_params.get("answer", "")
tool_result = {"status": "submitted", "message": "Answer recorded. Ending session."}
info["tool_result"] = tool_result
info["submitted_answer"] = self._submitted_answer
agent_done = True # Force episode end, same as harness behaviour
elif tool_name:
try:
tool_result = await self._tools.call_tool(tool_name, tool_params)
info["tool_result"] = tool_result
Expand Down Expand Up @@ -603,7 +644,12 @@ async def _compute_reward(self) -> float:

# Execute verifier in a thread to avoid blocking the event loop.
# verify_detailed() does sync HTTP calls internally.
response = await asyncio.to_thread(fleet_task.verify_detailed, fleet_env)
# Pass final_answer when model used submit_final_answer,
# mirroring how the harness routes the answer to the verifier.
verify_kwargs = {}
if self._submitted_answer is not None:
verify_kwargs["final_answer"] = self._submitted_answer
response = await asyncio.to_thread(fleet_task.verify_detailed, fleet_env, **verify_kwargs)

# Extract result from response
# response.success is bool, response.result is the verifier's return value (0.0 or 1.0)
Expand Down
64 changes: 64 additions & 0 deletions tests/envs/test_fleet_task_env.py
Original file line number Diff line number Diff line change
Expand Up @@ -520,3 +520,67 @@ async def mock_list_tools():
# Should only have computer tool
assert len(env._tools_cache) == 1
assert env._tools_cache[0]["function"]["name"] == "computer"


class TestSubmitFinalAnswer:
"""Tests for synthetic submit_final_answer tool injection."""

def test_submit_final_answer_tool_definition(self, mock_fleet_env_client):
"""SUBMIT_FINAL_ANSWER_TOOL has correct schema."""
from envs.fleet_env.task_env import SUBMIT_FINAL_ANSWER_TOOL

func = SUBMIT_FINAL_ANSWER_TOOL["function"]
assert func["name"] == "submit_final_answer"
assert "answer" in func["parameters"]["properties"]
assert func["parameters"]["required"] == ["answer"]

def test_submitted_answer_init(self, sample_task_config, mock_fleet_env_client):
"""_submitted_answer should be None on init."""
from envs.fleet_env.task_env import FleetTaskEnv

env = FleetTaskEnv(sample_task_config, api_key="test")
assert env._submitted_answer is None

@pytest.mark.anyio
async def test_step_submit_final_answer_stores_answer(
self, sample_task_config, mock_fleet_env_client
):
"""Calling submit_final_answer should store the answer and mark done."""
from envs.fleet_env.task_env import FleetTaskEnv

mock_orch, _ = mock_fleet_env_client
env = FleetTaskEnv(sample_task_config, api_key="test")
env._orch = mock_orch
env._tools = MagicMock()
env._tools_cache = [{"type": "function", "function": {"name": "bash"}}]
env._done = False
env._rollout_started = True

action = {"tool": "submit_final_answer", "params": {"answer": '["row1", "row2"]'}}
obs, reward, done, info = await env.step_async(action)

assert env._submitted_answer == '["row1", "row2"]'
assert done is True
assert info["submitted_answer"] == '["row1", "row2"]'
assert info["tool_result"]["status"] == "submitted"

@pytest.mark.anyio
async def test_step_submit_final_answer_not_routed_to_mcp(
self, sample_task_config, mock_fleet_env_client
):
"""submit_final_answer should NOT call MCP tools.call_tool."""
from envs.fleet_env.task_env import FleetTaskEnv

mock_orch, _ = mock_fleet_env_client
mock_tools = AsyncMock()
env = FleetTaskEnv(sample_task_config, api_key="test")
env._orch = mock_orch
env._tools = mock_tools
env._tools_cache = [{"type": "function", "function": {"name": "bash"}}]
env._done = False
env._rollout_started = True

action = {"tool": "submit_final_answer", "params": {"answer": "42"}}
await env.step_async(action)

mock_tools.call_tool.assert_not_called()