import json import pytest from scripts.generate_sft_environment_expansion import validate_case from scripts.run_sft_environment_expansion import has_unrecovered_tool_failure, score_turn def tool_start(action: str) -> dict: return { "type": "tool_start", "tool": "manage_calendar", "command": json.dumps({"action": action}), } def test_calendar_create_requires_create_action_not_any_calendar_call(): turn = { "prompt": "Add a one-hour prep event tomorrow at 9:00 AM.", "expected_tools": ["manage_calendar"], } failures = score_turn(turn, [tool_start("list_calendars")], "Which calendar?") assert any(item.startswith("missing_tool_action") for item in failures) def test_calendar_create_accepts_create_event_action(): turn = { "prompt": "Add a one-hour prep event tomorrow at 9:00 AM.", "expected_tools": ["manage_calendar"], } assert score_turn(turn, [tool_start("create_event")], "Done.") == [] def test_explicit_expected_actions_override_inference(): turn = { "prompt": "Handle the prep entry.", "expected_tools": ["manage_calendar"], "expected_actions": {"manage_calendar": ["delete_event"]}, } failures = score_turn(turn, [tool_start("list_events")], "Found it.") assert any(item.startswith("missing_tool_action") for item in failures) def test_compact_ui_control_command_extracts_action_name(): turn = { "prompt": "Open my calendar.", "expected_tools": ["ui_control"], "expected_actions": {"ui_control": ["open_panel"]}, } events = [{"type": "tool_start", "tool": "ui_control", "command": "open_panel calendar"}] assert score_turn(turn, events, "Opened.") == [] def test_mcp_expected_tool_name_matches_runtime_short_name(): turn = {"prompt": "Search my email.", "expected_tools": ["mcp__email__search_emails"]} events = [{"type": "tool_start", "tool": "search_emails", "command": "{}"}] assert score_turn(turn, events, "Found it.") == [] @pytest.mark.parametrize("expected,observed", [ ("edit_document", "update_document"), ("update_document", "edit_document"), ]) def test_active_document_writers_are_scored_by_function_not_name(expected, observed): turn = {"prompt": "Expand this active draft.", "expected_tools": [expected]} events = [{"type": "tool_start", "tool": observed, "command": "{}"}] assert score_turn(turn, events, "Updated.") == [] def test_corrected_tool_retry_does_not_remain_a_functional_failure(): events = [ {"type": "tool_output", "tool": "edit_document", "exit_code": 1, "error": True, "output": "Error: malformed arguments"}, {"type": "tool_output", "tool": "edit_document", "exit_code": 0, "error": False, "output": '{"action":"edit"}'}, ] assert not has_unrecovered_tool_failure(events) assert has_unrecovered_tool_failure(list(reversed(events))) def test_isolated_replay_materializes_inventory_email_rows(tmp_path, monkeypatch): from scripts import run_sft_environment_expansion as runner from core.database import Base from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker inventory = tmp_path / "environments.json" inventory.write_text(json.dumps({"environments": [{ "owner": "sft_maya_ops", "profile": { "primary": "maya@example.test", "secondary": "maya-research@example.test", "primary_account": "Primary Inbox", "secondary_account": "Ops Research", }, "emails": [ {"uid": "1001", "account": "Primary Inbox", "subject": "Review"}, {"uid": "1010", "account": "Ops Research", "subject": "Research"}, ], }]}), encoding="utf-8") monkeypatch.setattr(runner, "DATA_DIR", tmp_path / "isolated-data") engine = create_engine(f"sqlite:///{tmp_path / 'fixture.db'}") Base.metadata.create_all(engine) monkeypatch.setattr(runner, "SessionLocal", sessionmaker(bind=engine)) installed = runner.install_fixture_environments(inventory) assert installed["emails"] == 2 rows = json.loads( (runner.DATA_DIR / "fixture_email_messages.json").read_text(encoding="utf-8") )["messages"] assert rows[0]["owner"] == "sft_maya_ops" assert rows[0]["account_id"] == "primary-inbox" assert rows[0]["account_email"] == "maya@example.test" assert rows[1]["account_id"] == "secondary-inbox" assert rows[1]["account_email"] == "maya-research@example.test" assert rows[0]["body"].startswith("Fixture message for: Review") def test_generated_case_rejects_multiple_tool_families_in_one_turn(): seed = { "seed_family_id": "seed-1", "source_session_id": "session-1", "split": "train", "tools": ["bash", "manage_memory", "web_search"], } raw = { "owner": "sft_maya_ops", "turns": [ {"prompt": "Check the host and list my memories.", "expected_tools": ["bash", "manage_memory"]}, {"prompt": "Search for current operations news.", "expected_tools": ["web_search"]}, {"prompt": "Check the host name.", "expected_tools": ["bash"]}, ], } with pytest.raises(ValueError, match="invalid expected tools"): validate_case(seed, "sft_maya_ops", {"events": []}, raw, 1) def test_generated_case_normalizes_single_expected_tool_string(): seed = { "seed_family_id": "seed-1", "source_session_id": "session-1", "split": "train", "tools": ["manage_calendar"], } raw = { "owner": "sft_maya_ops", "turns": [ {"prompt": "Open my calendar for September 2026.", "expected_tools": "ui_control"}, {"prompt": "Show my events in September 2026.", "expected_tools": "manage_calendar"}, {"prompt": "Open the month view again.", "expected_tools": "ui_control"}, ], } case = validate_case(seed, "sft_maya_ops", {"events": []}, raw, 1) assert case["turns"][0]["expected_tools"] == ["ui_control"] def test_generated_case_rejects_action_contract_for_another_tool(): seed = { "seed_family_id": "seed-1", "source_session_id": "session-1", "split": "train", "tools": ["mcp__email__search_emails", "mcp__email__read_email"], } raw = { "owner": "sft_maya_ops", "turns": [ { "prompt": "Find Iris's email and open the thread.", "expected_tools": ["mcp__email__search_emails"], "expected_actions": {"mcp__email__read_email": ["read"]}, }, {"prompt": "Search again.", "expected_tools": ["mcp__email__search_emails"]}, {"prompt": "Read UID 1.", "expected_tools": ["mcp__email__read_email"]}, ], } with pytest.raises(ValueError, match="outside expected_tools"): validate_case(seed, "sft_maya_ops", {"events": []}, raw, 1) def test_generated_case_rejects_visible_fixture_marker_language(): seed = { "seed_family_id": "seed-1", "source_session_id": "session-1", "split": "train", "tools": ["manage_memory"], } raw = { "owner": "sft_maya_ops", "turns": [ {"prompt": "Save a temporary marker-scoped memory named {marker}.", "expected_tools": ["manage_memory"]}, {"prompt": "List my saved memories.", "expected_tools": ["manage_memory"]}, {"prompt": "Delete that memory.", "expected_tools": ["manage_memory"]}, ], } with pytest.raises(ValueError, match="empty or meta prompt"): validate_case(seed, "sft_maya_ops", {"events": []}, raw, 1) def test_generated_case_rejects_two_mutations_in_one_turn(): seed = { "seed_family_id": "seed-1", "source_session_id": "session-1", "split": "train", "tools": ["manage_memory"], } raw = { "owner": "sft_maya_ops", "turns": [ {"prompt": "Create a memory, then delete it.", "expected_tools": ["manage_memory"]}, {"prompt": "List my memories.", "expected_tools": ["manage_memory"]}, {"prompt": "Search for my preference.", "expected_tools": ["manage_memory"]}, ], } with pytest.raises(ValueError, match="multiple mutations"): validate_case(seed, "sft_maya_ops", {"events": []}, raw, 1)