"""Phase 5F + 5G + 5H tests — graduation, conflicts, MCP tools.""" from __future__ import annotations import pytest from atocore.engineering.conflicts import ( detect_conflicts_for_entity, list_open_conflicts, resolve_conflict, ) from atocore.engineering._graduation_prompt import ( build_user_message, parse_graduation_output, ) from atocore.engineering.service import ( create_entity, create_relationship, get_entity, init_engineering_schema, promote_entity, ) from atocore.memory.service import create_memory from atocore.models.database import get_connection, init_db # --- 5F Memory graduation --- def test_graduation_prompt_parses_positive_decision(): raw = """ {"graduate": true, "entity_type": "component", "name": "Primary Mirror", "description": "The 1.2m primary mirror for p04", "confidence": 0.85, "relationships": [{"rel_type": "part_of", "target_hint": "Optics Subsystem"}]} """ decision = parse_graduation_output(raw) assert decision is not None assert decision["graduate"] is True assert decision["entity_type"] == "component" assert decision["name"] == "Primary Mirror" assert decision["confidence"] == 0.85 assert decision["relationships"] == [ {"rel_type": "part_of", "target_hint": "Optics Subsystem"} ] def test_graduation_prompt_parses_negative_decision(): raw = '{"graduate": false, "reason": "conversational filler, no typed entity"}' decision = parse_graduation_output(raw) assert decision is not None assert decision["graduate"] is False assert "filler" in decision["reason"] def test_graduation_prompt_rejects_unknown_entity_type(): raw = '{"graduate": true, "entity_type": "quantum_thing", "name": "x"}' assert parse_graduation_output(raw) is None def test_graduation_prompt_tolerates_markdown_fences(): raw = '```json\n{"graduate": false, "reason": "ok"}\n```' d = parse_graduation_output(raw) assert d is not None assert d["graduate"] is False def test_promote_entity_marks_source_memory_graduated(tmp_data_dir): init_db() init_engineering_schema() mem = create_memory("knowledge", "The Primary Mirror is 1.2m Zerodur", project="p-test", status="active") # Create entity candidate pointing back to the memory ent = create_entity( "component", "Primary Mirror", project="p-test", status="candidate", source_refs=[f"memory:{mem.id}"], ) # Promote assert promote_entity(ent.id, actor="test-triage") # Memory should now be graduated with forward pointer with get_connection() as conn: row = conn.execute( "SELECT status, graduated_to_entity_id FROM memories WHERE id = ?", (mem.id,), ).fetchone() assert row["status"] == "graduated" assert row["graduated_to_entity_id"] == ent.id def test_promote_entity_without_memory_refs_no_graduation(tmp_data_dir): """Entity not backed by any memory — promote still works, no graduation.""" init_db() init_engineering_schema() ent = create_entity("component", "Orphan", project="p-test", status="candidate") assert promote_entity(ent.id) assert get_entity(ent.id).status == "active" # --- 5G Conflict detection --- def test_component_material_conflict_detected(tmp_data_dir): init_db() init_engineering_schema() c = create_entity("component", "Mirror", project="p-test") m1 = create_entity("material", "Zerodur", project="p-test") m2 = create_entity("material", "ULE", project="p-test") create_relationship(c.id, m1.id, "uses_material") create_relationship(c.id, m2.id, "uses_material") detected = detect_conflicts_for_entity(c.id) assert len(detected) == 1 conflicts = list_open_conflicts(project="p-test") assert any(c["slot_kind"] == "component.material" for c in conflicts) conflict = next(c for c in conflicts if c["slot_kind"] == "component.material") assert len(conflict["members"]) == 2 def test_component_part_of_conflict_detected(tmp_data_dir): init_db() init_engineering_schema() c = create_entity("component", "MultiPart", project="p-test") s1 = create_entity("subsystem", "Mechanical", project="p-test") s2 = create_entity("subsystem", "Optical", project="p-test") create_relationship(c.id, s1.id, "part_of") create_relationship(c.id, s2.id, "part_of") detected = detect_conflicts_for_entity(c.id) assert len(detected) == 1 conflicts = list_open_conflicts(project="p-test") assert any(c["slot_kind"] == "component.part_of" for c in conflicts) def test_requirement_name_conflict_detected(tmp_data_dir): init_db() init_engineering_schema() r1 = create_entity("requirement", "Surface figure < 25nm", project="p-test", description="Primary mirror spec") r2 = create_entity("requirement", "Surface figure < 25nm", project="p-test", description="Different interpretation") detected = detect_conflicts_for_entity(r2.id) assert len(detected) == 1 conflicts = list_open_conflicts(project="p-test") assert any(c["slot_kind"] == "requirement.name" for c in conflicts) def test_conflict_not_detected_for_clean_component(tmp_data_dir): init_db() init_engineering_schema() c = create_entity("component", "Clean", project="p-test") m = create_entity("material", "Zerodur", project="p-test") create_relationship(c.id, m.id, "uses_material") detected = detect_conflicts_for_entity(c.id) assert detected == [] def test_conflict_resolution_supersedes_losers(tmp_data_dir): init_db() init_engineering_schema() c = create_entity("component", "Mirror2", project="p-test") m1 = create_entity("material", "Zerodur2", project="p-test") m2 = create_entity("material", "ULE2", project="p-test") create_relationship(c.id, m1.id, "uses_material") create_relationship(c.id, m2.id, "uses_material") detected = detect_conflicts_for_entity(c.id) conflict_id = detected[0] # Resolve by picking m1 as the winner assert resolve_conflict(conflict_id, "supersede_others", winner_id=m1.id) # m2 should now be superseded; m1 stays active assert get_entity(m1.id).status == "active" assert get_entity(m2.id).status == "superseded" # Conflict should be marked resolved open_conflicts = list_open_conflicts(project="p-test") assert not any(c["id"] == conflict_id for c in open_conflicts) def test_conflict_resolution_dismiss_leaves_entities_alone(tmp_data_dir): init_db() init_engineering_schema() r1 = create_entity("requirement", "Dup req", project="p-test", description="first meaning") r2 = create_entity("requirement", "Dup req", project="p-test", description="second meaning") detected = detect_conflicts_for_entity(r2.id) conflict_id = detected[0] assert resolve_conflict(conflict_id, "dismiss") # Both still active — dismiss just clears the conflict marker assert get_entity(r1.id).status == "active" assert get_entity(r2.id).status == "active" def test_deduplicate_conflicts_for_same_slot(tmp_data_dir): """Running detection twice on the same entity shouldn't dup the conflict row.""" init_db() init_engineering_schema() c = create_entity("component", "Dup", project="p-test") m1 = create_entity("material", "A", project="p-test") m2 = create_entity("material", "B", project="p-test") create_relationship(c.id, m1.id, "uses_material") create_relationship(c.id, m2.id, "uses_material") detect_conflicts_for_entity(c.id) detect_conflicts_for_entity(c.id) # should be a no-op conflicts = list_open_conflicts(project="p-test") mat_conflicts = [c for c in conflicts if c["slot_kind"] == "component.material"] assert len(mat_conflicts) == 1 def test_promote_triggers_conflict_detection(tmp_data_dir): """End-to-end: promoting a candidate component with 2 active material edges triggers conflict detection.""" init_db() init_engineering_schema() c = create_entity("component", "AutoFlag", project="p-test", status="candidate") m1 = create_entity("material", "X1", project="p-test") m2 = create_entity("material", "X2", project="p-test") create_relationship(c.id, m1.id, "uses_material") create_relationship(c.id, m2.id, "uses_material") promote_entity(c.id, actor="test") conflicts = list_open_conflicts(project="p-test") assert any(c["slot_kind"] == "component.material" for c in conflicts) # --- 5H MCP tool shape checks (via build_user_message) --- def test_graduation_user_message_includes_project_and_type(): msg = build_user_message("some content", "p04-gigabit", "project") assert "p04-gigabit" in msg assert "project" in msg assert "some content" in msg