Guard target selector overrides with action and shape evidence

graph_tool_call/graphify/catalog.py

@@ -25,7 +25,7 @@
 _READ_TERMS = frozenset(
     {"get", "read", "detail", "view", "show", "check", "retrieve", "조회", "상세", "보기", "확인"}
 )
-_DETAIL_READ_TERMS = frozenset({"detail", "details", "view", "show", "상세", "보기", "보여"})
+_DETAIL_READ_TERMS = frozenset({"detail", "details", "상세", "단건"})
 _SINGLE_TERMS = frozenset({"detail", "details", "info", "view", "single", "상세", "정보", "단건"})
 _LIST_TERMS = frozenset({"list", "lists", "search", "find", "query", "목록", "리스트", "검색"})
 _COUNT_TERMS = frozenset({"count", "total", "cnt", "건수", "개수", "카운트"})
@@ -184,8 +184,33 @@ def target_action_priority_for_query(query: str) -> dict[str, int]:
     has_search = _has_action_term(terms, _SEARCH_TERMS)
     has_read = _has_action_term(terms, _READ_TERMS)
 
+    last_action = _last_explicit_action(query)
+    directive_query = _query_has_directive_cue(query)
+
     if has_read and _has_action_term(terms, _AUDIT_READ_TERMS):
         return dict(_ACTION_PRIORITY_READ)
+    if (
+        directive_query
+        and last_action == "action"
+        and _has_action_term(terms, _NOTIFICATION_TERMS)
+        and _has_action_term(terms, _NOTIFICATION_SEND_TERMS)
+    ):
+        return dict(_ACTION_PRIORITY_NOTIFICATION)
+    if directive_query:
+        if last_action == "delete":
+            return dict(_ACTION_PRIORITY_DELETE)
+        if last_action == "action":
+            return dict(_ACTION_PRIORITY_ACTION)
+        if last_action == "update":
+            return dict(_ACTION_PRIORITY_UPDATE)
+        if last_action == "create":
+            return dict(_ACTION_PRIORITY_CREATE)
+        if last_action == "search":
+            return dict(_ACTION_PRIORITY_SEARCH)
+        if last_action == "read":
+            if has_search and not _query_has_strict_detail(terms):
+                return dict(_ACTION_PRIORITY_SEARCH)
+            return dict(_ACTION_PRIORITY_READ)
     if _has_action_term(terms, _NOTIFICATION_TERMS) and _has_action_term(
         terms,
         _NOTIFICATION_SEND_TERMS,
@@ -418,17 +443,36 @@ def select_target_candidate(
     overrode = False
     ambiguous = False
     reason_codes: list[str] = []
+    override_block_reason = ""
 
     if llm_name and llm_row and llm_row["name"] != winner["name"]:
+        winner_action_compatible = _action_priority_compatible(
+            action_priority,
+            str(winner.get("canonical_action") or ""),
+        )
+        llm_action_compatible = _action_priority_compatible(
+            action_priority,
+            str(llm_row.get("canonical_action") or ""),
+        )
         strong = (
             bool(winner["strong_evidence"])
             and llm_margin is not None
             and llm_margin >= _SELECTOR_OVERRIDE_MARGIN
         )
+        if strong and llm_action_compatible and not winner_action_compatible:
+            strong = False
+            override_block_reason = "action_incompatible_override_blocked"
+        winner_rank = int(winner.get("original_rank") or 9999)
+        llm_rank = int(llm_row.get("original_rank") or 9999)
+        if strong and winner_rank > llm_rank and not _has_decisive_override_evidence(winner):
+            strong = False
+            override_block_reason = "lower_rank_surface_override_blocked"
         if policy == "strong_evidence" and strong:
             selected = winner
             overrode = True
             reason_codes.append("llm_target_overridden")
+        elif override_block_reason:
+            reason_codes.append(override_block_reason)
         else:
             ambiguous = True
             reason_codes.append("ambiguous_target")
@@ -751,19 +795,27 @@ def _score_target_candidate(
     effective_result_shape = result_shape or _infer_result_shape_from_surface(surface_terms, action)
 
     action_score = _normalized_priority(action_priority, action)
-    if action_score:
+    if action_score >= 0.5:
         value = 0.2 * action_score
         score += value
         evidence.append({"source": "canonical_action", "value": action, "score": round(value, 6)})
+    elif action_priority and action:
+        score -= 0.18
+        evidence.append({"source": "action_mismatch", "value": action, "score": -0.18})
 
     shape_score = _normalized_priority(shape_priority, effective_result_shape)
-    if shape_score:
+    if shape_score >= 0.5:
         source = "result_shape" if result_shape else "inferred_result_shape"
         value = 0.18 * shape_score
         score += value
         evidence.append(
             {"source": source, "value": effective_result_shape, "score": round(value, 6)}
         )
+    elif shape_priority and effective_result_shape:
+        score -= 0.14
+        evidence.append(
+            {"source": "result_shape_mismatch", "value": effective_result_shape, "score": -0.14}
+        )
 
     overlap = query_terms & surface_terms
     if overlap:
@@ -845,14 +897,14 @@ def _score_target_candidate(
 
 
 def _result_shape_priority_for_query(query: str) -> dict[str, int]:
-    terms = _query_terms(query)
-    if _has_action_term(terms, _COUNT_TERMS):
+    last_shape = _last_explicit_shape(query)
+    if last_shape == "count":
         return {"count": 6, "list": 3, "single": 1}
-    if _query_has_detail(terms):
+    if last_shape == "single":
         return {"single": 6, "list": 2, "count": 1}
-    if _has_action_term(terms, _LIST_TERMS):
+    if last_shape == "list":
         return {"list": 6, "count": 3, "single": 1}
-    if _has_action_term(terms, _CREATE_TERMS | _UPDATE_TERMS | _DELETE_TERMS | _ACTION_TERMS):
+    if _last_explicit_action(query) in {"create", "update", "delete", "action"}:
         return {"mutation": 5, "single": 1}
     return {}
 
@@ -866,6 +918,95 @@ def _normalized_priority(priority: dict[str, int], key: str) -> float:
     return max(0.0, float(priority.get(key, 0)) / max_value)
 
 
+def _action_priority_compatible(priority: dict[str, int], action: str) -> bool:
+    if not priority or not action:
+        return True
+    return _normalized_priority(priority, action) >= 0.5
+
+
+def _has_decisive_override_evidence(row: dict[str, Any]) -> bool:
+    decisive_sources = {
+        "api_contract",
+        "detail_surface",
+        "identifier_detail_contract",
+        "identifier_detail_surface",
+    }
+    return any(
+        evidence.get("source") in decisive_sources
+        for evidence in row.get("evidence") or []
+        if isinstance(evidence, dict)
+    )
+
+
+def _last_explicit_action(query: str) -> str:
+    text = str(query or "").strip().lower()
+    if not text:
+        return ""
+    groups = {
+        "delete": _DELETE_TERMS,
+        "action": _ACTION_TERMS | _NOTIFICATION_SEND_TERMS,
+        "update": _UPDATE_TERMS,
+        "create": _CREATE_TERMS,
+        "search": _SEARCH_TERMS,
+        "read": _READ_TERMS,
+    }
+    matches: list[tuple[int, int, str]] = []
+    for action, action_terms in groups.items():
+        for term in action_terms:
+            index = text.rfind(term)
+            if index >= 0:
+                matches.append((index, len(term), action))
+    if not matches:
+        return ""
+    return max(matches)[2]
+
+
+def _last_explicit_shape(query: str) -> str:
+    text = str(query or "").strip().lower()
+    if not text:
+        return ""
+    groups = {
+        "count": _COUNT_TERMS,
+        "single": _SINGLE_TERMS,
+        "list": _LIST_TERMS,
+    }
+    matches: list[tuple[int, int, str]] = []
+    for shape, shape_terms in groups.items():
+        for term in shape_terms:
+            index = text.rfind(term)
+            if index >= 0:
+                matches.append((index, len(term), shape))
+    if not matches:
+        return ""
+    return max(matches)[2]
+
+
+def _query_has_directive_cue(query: str) -> bool:
+    text = str(query or "").strip().lower()
+    if not text:
+        return False
+    if any(
+        cue in text
+        for cue in (
+            "해줘",
+            "해주세요",
+            "해 줘",
+            "해 주",
+            "줘",
+            "주세요",
+            "줄래",
+            "please",
+        )
+    ):
+        return True
+    return bool(
+        re.match(
+            r"^(show|get|read|list|search|find|create|add|update|edit|delete|remove|send|run)\b",
+            text,
+        )
+    )
+
+
 def _selector_terms(text: str) -> set[str]:
     spaced = re.sub(r"(?<=[a-z0-9])(?=[A-Z])", " ", str(text or ""))
     raw_terms = [

tests/test_graphify_contract_025.py

@@ -1977,6 +1977,155 @@ def test_select_target_candidate_preserves_llm_target_when_margin_is_weak():
     assert "ambiguous_target" in result["reason_codes"]
 
 
+def test_target_action_priority_uses_final_requested_action():
+    assert target_action_priority_for_query("이미지 생성 작업의 현재 상태를 확인해줘") == {
+        "read": 6,
+        "search": 4,
+        "action": 2,
+    }
+    assert target_action_priority_for_query(
+        "처리내역을 등록할 때 필요한 기초 데이터를 조회해줘"
+    ) == {"read": 6, "search": 4, "action": 2}
+    assert target_action_priority_for_query("현재 목록을 조회한 후 선택한 항목을 수정해줘") == {
+        "update": 6,
+        "action": 5,
+        "create": 3,
+        "read": 2,
+        "search": 1,
+    }
+    assert target_action_priority_for_query("발송된 메시지 정보를 목록으로 조회해줘") == {
+        "search": 6,
+        "read": 4,
+        "action": 2,
+    }
+
+
+def test_select_target_candidate_does_not_override_read_with_mutation():
+    tools = {
+        "getImageJob": {
+            "description": "이미지 생성 작업 상태 조회",
+            "metadata": {
+                "ai_metadata": {
+                    "canonical_action": "read",
+                    "primary_resource": "image_job",
+                    "result_shape": "single",
+                },
+                "openapi": {"summary": "이미지 생성 잡 상태 조회"},
+            },
+        },
+        "startImageJob": {
+            "description": "이미지 생성 작업 시작",
+            "metadata": {
+                "ai_metadata": {
+                    "canonical_action": "create",
+                    "primary_resource": "image_job",
+                    "result_shape": "mutation",
+                },
+                "openapi": {"summary": "이미지 생성 잡 시작"},
+            },
+        },
+    }
+
+    result = select_target_candidate(
+        "이미지 생성 작업의 현재 상태를 확인해줘",
+        ["startImageJob", "getImageJob"],
+        tools,
+        retrieval_results=[
+            {"name": "startImageJob", "score": 0.9},
+            {"name": "getImageJob", "score": 0.01},
+        ],
+        llm_target="getImageJob",
+    )
+
+    assert result["selected_target"] == "getImageJob"
+    assert result["overrode_llm"] is False
+    assert "llm_target_overridden" not in result["reason_codes"]
+    mutation = next(row for row in result["rank_signals"] if row["name"] == "startImageJob")
+    assert any(row["source"] == "action_mismatch" for row in mutation["evidence"])
+
+
+def test_select_target_candidate_blocks_lower_rank_surface_only_override():
+    tools = {
+        "getBusinessProcessingStatus": {
+            "description": "CS processing status by business type",
+            "metadata": {
+                "ai_metadata": {
+                    "canonical_action": "search",
+                    "primary_resource": "customer_service_statistics",
+                    "result_shape": "list",
+                }
+            },
+        },
+        "getAssigneeAllocationStatus": {
+            "description": "업무유형별 고객상담 처리 현황 목록",
+            "metadata": {
+                "ai_metadata": {
+                    "canonical_action": "search",
+                    "primary_resource": "counsel_assignment",
+                    "result_shape": "list",
+                }
+            },
+        },
+    }
+
+    result = select_target_candidate(
+        "업무유형별 고객상담 처리 현황 목록을 보여줘",
+        ["getBusinessProcessingStatus", "getAssigneeAllocationStatus"],
+        tools,
+        retrieval_results=[
+            {"name": "getBusinessProcessingStatus", "score": 0.03},
+            {"name": "getAssigneeAllocationStatus", "score": 0.02},
+        ],
+        llm_target="getBusinessProcessingStatus",
+    )
+
+    assert result["selected_target"] == "getBusinessProcessingStatus"
+    assert result["overrode_llm"] is False
+    assert "lower_rank_surface_override_blocked" in result["reason_codes"]
+    assert "llm_target_preserved" in result["reason_codes"]
+
+
+def test_select_target_candidate_prefers_explicit_list_shape_over_info_term():
+    tools = {
+        "getMessageBaseInfo": {
+            "description": "메시지 기본 정보 조회",
+            "metadata": {
+                "ai_metadata": {
+                    "canonical_action": "read",
+                    "primary_resource": "message",
+                    "result_shape": "single",
+                }
+            },
+        },
+        "getMessageList": {
+            "description": "메시지 정보 목록 조회",
+            "metadata": {
+                "ai_metadata": {
+                    "canonical_action": "search",
+                    "primary_resource": "message",
+                    "result_shape": "list",
+                }
+            },
+        },
+    }
+
+    result = select_target_candidate(
+        "발송된 메시지 정보를 목록으로 조회해줘",
+        ["getMessageBaseInfo", "getMessageList"],
+        tools,
+        retrieval_results=[
+            {"name": "getMessageBaseInfo", "score": 0.03},
+            {"name": "getMessageList", "score": 0.02},
+        ],
+        llm_target="getMessageList",
+    )
+
+    assert result["selected_target"] == "getMessageList"
+    assert result["overrode_llm"] is False
+    single = next(row for row in result["rank_signals"] if row["name"] == "getMessageBaseInfo")
+    assert any(row["source"] == "result_shape_mismatch" for row in single["evidence"])
+
+
 def test_edge_normalize_merge_and_trace_derivation_contract():
     structural = normalize_graph_edge(
         {