Spaces:
Running on Zero
Running on Zero
| import json | |
| from collections import Counter | |
| def _accepted_v6_row(): | |
| from figment.retrieval import load_protocol_cards | |
| from scripts.generate_finetune_data import assemble_teacher_navigator_output | |
| from scripts.generate_finetune_data import build_sft_row | |
| from scripts.generate_finetune_data import case_spec_record | |
| from scripts.generate_finetune_data import generate_case_spec | |
| from scripts.generate_finetune_data import prepare_case | |
| from scripts.generate_finetune_data import score_candidate | |
| cards_by_id = {str(card["card_id"]): card for card in load_protocol_cards()} | |
| spec = generate_case_spec(0, cards_by_id, dataset_version="figment_sft_v6_delta") | |
| prepared = prepare_case(spec, cards_by_id) | |
| candidate = assemble_teacher_navigator_output( | |
| prepared, | |
| { | |
| "facts": ["confirmed field concern"], | |
| "missing": ["confirm current mental status", "record available vital signs"], | |
| "observe": ["confirm current mental status", "record available vital signs"], | |
| "checklist": ["cite deterministic rule cards"], | |
| "uncertain": ["some vitals remain incomplete"], | |
| "sbar": { | |
| "situation": "confirmed handoff concern", | |
| "background": "field workflow setting", | |
| "assessment_observations_only": "observations only from confirmed intake", | |
| "handoff_request": "request protocol review", | |
| }, | |
| "script": "I am checking protocol observations.", | |
| }, | |
| ) | |
| result = score_candidate(candidate, prepared) | |
| assert result.passed is True, result.reward_components | |
| row = build_sft_row( | |
| prepared=prepared, | |
| result=result, | |
| teacher_model_id="teacher-test", | |
| candidate_total=1, | |
| candidate_passed=1, | |
| ) | |
| return row, case_spec_record(prepared) | |
| def test_v6_failure_cycle_matches_updated_plan_and_interleaves_smoke_cases(): | |
| from scripts.generate_finetune_data import V6_NAVIGATOR_COUNTS | |
| from scripts.generate_finetune_data import _failure_class_for_index | |
| categories = Counter( | |
| _failure_class_for_index(index, dataset_version="figment_sft_v6_delta") | |
| for index in range(sum(V6_NAVIGATOR_COUNTS.values())) | |
| ) | |
| first_twelve = { | |
| _failure_class_for_index(index, dataset_version="figment_sft_v6_delta") | |
| for index in range(12) | |
| } | |
| assert categories == V6_NAVIGATOR_COUNTS | |
| assert {"required_observation_ownership", "observation_correction", "v6_preservation"} <= first_twelve | |
| def test_v6_full_corpus_wrapper_pins_delta_defaults(): | |
| from scripts.generate_v6_full_corpus import DEFAULT_OUTPUT_VERSION | |
| from scripts.generate_v6_full_corpus import DEFAULT_REPAIR_COUNT | |
| from scripts.generate_v6_full_corpus import DEFAULT_TEACHER_MODEL_ID | |
| from scripts.generate_v6_full_corpus import build_corpus_args | |
| assert DEFAULT_OUTPUT_VERSION == "figment_sft_v6_delta" | |
| assert DEFAULT_TEACHER_MODEL_ID == "nvidia/nemotron-3-ultra-550b-a55b:free" | |
| assert DEFAULT_REPAIR_COUNT == 250 | |
| args = build_corpus_args(["--new-delta-count", "10", "--correction-count", "2", "--repair-count", "3", "--dry-run"]) | |
| assert args[args.index("--navigator-count") + 1] == "12" | |
| assert args[args.index("--repair-count") + 1] == "3" | |
| assert args[-1] == "--dry-run" | |
| def test_v6_sft_row_records_observation_policy_metadata(): | |
| row, spec_record = _accepted_v6_row() | |
| output = json.loads(row["messages"][1]["content"]) | |
| metadata = row["metadata"] | |
| assert row["version"] == "figment_sft_v6_delta" | |
| assert row["category"] == "required_observation_ownership" | |
| assert metadata["training_focus"] == "required_observation_ownership" | |
| assert metadata["v6_training_policy_version"] == 1 | |
| assert metadata["required_observation_targets"] | |
| assert output["selected_required_observation_ids"] | |
| assert set(metadata["must_include_selected_required_observation_ids"]) <= set( | |
| output["selected_required_observation_ids"] | |
| ) | |
| assert output["missing_info_to_collect"] != output["next_observations_to_collect"] | |
| observation_text = json.dumps( | |
| output["missing_info_to_collect"] + output["next_observations_to_collect"] | |
| ).lower() | |
| assert "source card ids" not in observation_text | |
| assert spec_record["dataset_version"] == "figment_sft_v6_delta" | |
| assert spec_record["must_include_selected_required_observation_ids"] | |
| def test_v6_policy_rejects_duplicate_metadata_and_invisible_selected_ids(): | |
| from scripts.generate_finetune_data import v6_policy_issues | |
| output = { | |
| "source_cards": ["STROKE-SIGNS-v1"], | |
| "selected_required_observation_ids": ["STROKE-SIGNS-v1::required_observation::1"], | |
| "missing_info_to_collect": [ | |
| "source card IDs", | |
| "deterministic rule results", | |
| "ask about something else", | |
| "monitor closely", | |
| ], | |
| "next_observations_to_collect": [ | |
| "source card IDs", | |
| "deterministic rule results", | |
| "ask about something else", | |
| "monitor closely", | |
| ], | |
| "handoff_note_sbar": { | |
| "situation": "stroke signs", | |
| "background": "field setting", | |
| "assessment_observations_only": "observations pending", | |
| "handoff_request": "request protocol review", | |
| }, | |
| } | |
| retrieved_cards = [ | |
| { | |
| "card_id": "STROKE-SIGNS-v1", | |
| "card": { | |
| "card_id": "STROKE-SIGNS-v1", | |
| "required_observations": ["face droop observation"], | |
| }, | |
| } | |
| ] | |
| issues = v6_policy_issues( | |
| output, | |
| failure_class="required_observation_ownership", | |
| expected_red_flag_rule_ids=[], | |
| expected_candidate_pathway_card_ids=["STROKE-SIGNS-v1"], | |
| structured_intake={}, | |
| rule_results=[], | |
| retrieved_cards=retrieved_cards, | |
| target_protocol_card_id="STROKE-SIGNS-v1", | |
| ) | |
| assert "duplicate_long_missing_and_next_observations" in issues | |
| assert any(issue.startswith("harness_metadata_observation:") for issue in issues) | |
| assert "selected_required_observation_id_not_visible:STROKE-SIGNS-v1::required_observation::1" in issues | |
| def test_v6_repair_scope_schedule_targets_observation_repairs(): | |
| from scripts.augment_finetune_repair_rows import _scope_schedule | |
| assert Counter(_scope_schedule(250, dataset_version="figment_sft_v6_delta")) == { | |
| "missing_observations": 250 | |
| } | |
| def test_verify_v6_rejects_rows_with_harness_metadata_observations(tmp_path): | |
| from scripts.verify_finetune_harness_alignment import verify_rows | |
| row, spec_record = _accepted_v6_row() | |
| output = json.loads(row["messages"][1]["content"]) | |
| output["missing_info_to_collect"] = [ | |
| "source card IDs", | |
| "deterministic rule results", | |
| "navigator validation result", | |
| "confirmed intake status", | |
| ] | |
| output["next_observations_to_collect"] = list(output["missing_info_to_collect"]) | |
| row["messages"][1]["content"] = json.dumps(output, sort_keys=True) | |
| dataset = tmp_path / "rows.jsonl" | |
| case_specs = tmp_path / "specs.jsonl" | |
| dataset.write_text(json.dumps(row, sort_keys=True) + "\n", encoding="utf-8") | |
| case_specs.write_text(json.dumps(spec_record, sort_keys=True) + "\n", encoding="utf-8") | |
| summary = verify_rows(dataset_path=dataset, case_specs_path=case_specs) | |
| assert summary["passed"] is False | |
| assert summary["issue_types"]["v6_duplicate_long_missing_and_next_observations"] >= 1 | |
| assert any(key.startswith("v6_harness_metadata_observation") for key in summary["issue_types"]) | |