diff --git a/src/backend/services/text_value_service.py b/src/backend/services/text_value_service.py index fac1c8bc5..3bf5f0a65 100644 --- a/src/backend/services/text_value_service.py +++ b/src/backend/services/text_value_service.py @@ -95,11 +95,23 @@ def _invalidate_workflow_routing_projection_for_text_relation_change( return try: - from .workflow_capability_service import invalidate_workflow_capability_index + from .workflow_capability_service import ( + invalidate_workflow_capability_index, + is_authoritative_workflow_concept_id, + ) from .workflow_discovery_service import ( invalidate_workflow_discovery_executability_caches, ) + # Scope invalidation to concepts the capability index actually tracks. + # Background workflows continually rewrite descriptions on unrelated + # (non-workflow) concepts; without this guard every such write churns + # the routing index and starves live workflow discovery of a ready + # index. New workflows are indexed via the registration path, so this + # only suppresses no-op invalidations from non-workflow subjects. + if not is_authoritative_workflow_concept_id(subject_concept_id): + return + invalidate_workflow_capability_index( reason=( "workflow_routing_text_relation_changed:" diff --git a/src/backend/services/workflow_capability_service.py b/src/backend/services/workflow_capability_service.py index 014b2cf4b..890e52f45 100644 --- a/src/backend/services/workflow_capability_service.py +++ b/src/backend/services/workflow_capability_service.py @@ -313,6 +313,34 @@ def _authoritative_registry_workflow_ids(registry: Any) -> tuple[str, ...]: ) +def is_authoritative_workflow_concept_id(concept_id: str) -> bool: + """Return True when ``concept_id`` is an authoritative workflow concept. + + Used to scope routing-projection invalidation: only writes to concepts the + capability index actually tracks should churn it. Description/content writes + on unrelated concepts (e.g. task or episode-critique concepts produced by + background workflows) must not invalidate the workflow routing index. + + Fails open (returns True) when the registry cannot be resolved, so a lookup + failure degrades to the prior always-invalidate behaviour rather than + silently skipping a legitimate workflow update. + """ + + cleaned = str(concept_id or "").strip() + if not cleaned: + return False + try: + from ..workflows.durable.registry_factory import ( + get_shared_workflow_registry_read_only, + ) + + registry = get_shared_workflow_registry_read_only(defer_parity_work=True) + authoritative_ids = set(_authoritative_registry_workflow_ids(registry)) + except Exception: + return True + return cleaned in authoritative_ids + + def _authoritative_registry_fingerprint_rows( registry: Any, ) -> tuple[dict[str, str], ...]: diff --git a/tests/backend/test_workflow_capability_service.py b/tests/backend/test_workflow_capability_service.py index b5e73126c..fe5ff67b6 100644 --- a/tests/backend/test_workflow_capability_service.py +++ b/tests/backend/test_workflow_capability_service.py @@ -1282,6 +1282,10 @@ def test_workflow_routing_text_relation_change_invalidates_projection( "src.backend.services.workflow_discovery_service.invalidate_workflow_discovery_executability_caches", lambda: discovery_cache_clears.append(True), ) + monkeypatch.setattr( + "src.backend.services.workflow_capability_service.is_authoritative_workflow_concept_id", + lambda concept_id: concept_id == "#V#workflow_repair_or_create_workflow", + ) text_value_service._invalidate_workflow_routing_projection_for_text_relation_change( subject_concept_id="#V#workflow_repair_or_create_workflow", @@ -1297,6 +1301,85 @@ def test_workflow_routing_text_relation_change_invalidates_projection( assert discovery_cache_clears == [True] +def test_workflow_routing_text_relation_change_skips_non_workflow_subject( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A routing-relevant predicate on a non-workflow concept must not churn + the capability index. Background workflows continually rewrite descriptions + on task/episode concepts; those writes previously invalidated the routing + index and starved live workflow discovery of a ready index.""" + + from src.backend.services import text_value_service + + invalidations: list[str | None] = [] + discovery_cache_clears: list[bool] = [] + + monkeypatch.setattr( + "src.backend.services.workflow_capability_service.invalidate_workflow_capability_index", + lambda **kwargs: ( + invalidations.append(kwargs.get("reason")) or {"success": True} + ), + ) + monkeypatch.setattr( + "src.backend.services.workflow_discovery_service.invalidate_workflow_discovery_executability_caches", + lambda: discovery_cache_clears.append(True), + ) + monkeypatch.setattr( + "src.backend.services.workflow_capability_service.is_authoritative_workflow_concept_id", + lambda concept_id: False, + ) + + text_value_service._invalidate_workflow_routing_projection_for_text_relation_change( + subject_concept_id=( + "#V#task_episode_critique_remediation_" + "vpaperrecommendationevaluationworkflow_metadatavalidationfailure_c64dffb4" + ), + predicate="#V#hasDescription", + ) + + assert invalidations == [] + assert discovery_cache_clears == [] + + +def test_is_authoritative_workflow_concept_id( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import src.backend.services.workflow_capability_service as capability_service + + monkeypatch.setattr( + "src.backend.workflows.durable.registry_factory.get_shared_workflow_registry_read_only", + lambda **kwargs: object(), + ) + monkeypatch.setattr( + capability_service, + "_authoritative_registry_workflow_ids", + lambda registry: ("#V#a_workflow", "#V#b_workflow"), + ) + + assert capability_service.is_authoritative_workflow_concept_id("#V#a_workflow") + assert not capability_service.is_authoritative_workflow_concept_id("#V#task_x") + assert not capability_service.is_authoritative_workflow_concept_id("") + assert not capability_service.is_authoritative_workflow_concept_id(None) # type: ignore[arg-type] + + +def test_is_authoritative_workflow_concept_id_fails_open( + monkeypatch: pytest.MonkeyPatch, +) -> None: + import src.backend.services.workflow_capability_service as capability_service + + def _raise(**kwargs: object) -> object: + raise RuntimeError("registry unavailable") + + monkeypatch.setattr( + "src.backend.workflows.durable.registry_factory.get_shared_workflow_registry_read_only", + _raise, + ) + + # Registry lookup failure must degrade to the prior always-invalidate + # behaviour rather than silently skipping a legitimate workflow update. + assert capability_service.is_authoritative_workflow_concept_id("#V#task_x") + + def test_workflow_graph_relationship_change_invalidates_projection( monkeypatch: pytest.MonkeyPatch, ) -> None: