diff --git a/docs/engineering/von_workflow_language_manual.md b/docs/engineering/von_workflow_language_manual.md
index a2e2a307a..075c6f60e 100644
--- a/docs/engineering/von_workflow_language_manual.md
+++ b/docs/engineering/von_workflow_language_manual.md
@@ -222,12 +222,23 @@ Required input:
 Optional input:
 
 - `tool_arguments`: object payload passed to the internal MCP gateway.
+- `suppress_event_workflow_launches`: static boolean control for a
+  workflow-owned mutation whose event-driven fan-out would duplicate the
+  enclosing workflow's represented work. The control is not forwarded to the
+  MCP tool, does not grant mutation authority or bypass access/write
+  guardrails, and applies only for the request-local gateway invocation. When
+  true, event-workflow launch and workflow-discovery cache invalidation are
+  suppressed for that invocation. Workflows MUST use it only when the affected
+  mutation cannot change workflow routing authority and the enclosing workflow
+  performs explicit canonical read-back.
 
 Runtime semantics:
 
 - workflow context bindings inside `tool_arguments` are resolved by the normal action-input resolver before invocation;
 - authenticated namespace context is propagated into the MCP payload as `namespace` when available;
 - the action returns the workflow-visible MCP payload under `result` and `mcp_result`, with `mcp_tool`, `mcp_requested_tool`, `mcp_resolved_tool`, and `mcp_duration_ms` diagnostics;
+- the action reports `event_workflow_launch_suppression_requested` so traces
+  distinguish an authored suppression request from ordinary tool execution;
 - internal MCP advisory budgets are telemetry only: a handler that completes
   before its hard deadline remains successful even when the advisory budget was
   exceeded;
diff --git a/scripts/generate_spreadsheet_programme_workflow_seed.py b/scripts/generate_spreadsheet_programme_workflow_seed.py
index 0daeee005..156c698b2 100644
--- a/scripts/generate_spreadsheet_programme_workflow_seed.py
+++ b/scripts/generate_spreadsheet_programme_workflow_seed.py
@@ -207,18 +207,14 @@ def stable_json(value: object) -> str:
 def kr_materialisation_dependency_identity() -> dict[str, str]:
     """Return the stable identity of the exact invoked KR seed authority."""
 
-    payload = json.loads(
-        KR_MATERIALISATION_SEED_BUNDLE.read_text(encoding="utf-8")
-    )
+    payload = json.loads(KR_MATERIALISATION_SEED_BUNDLE.read_text(encoding="utf-8"))
     if not isinstance(payload, dict):
         raise ValueError("kr_materialisation_seed_bundle_invalid")
     seed_version = str(payload.get("seed_version") or "").strip()
     family_id = str(payload.get("family_id") or "").strip()
     if not seed_version or not family_id:
         raise ValueError("kr_materialisation_seed_bundle_identity_missing")
-    canonical_sha256 = hashlib.sha256(
-        stable_json(payload).encode("utf-8")
-    ).hexdigest()
+    canonical_sha256 = hashlib.sha256(stable_json(payload).encode("utf-8")).hexdigest()
     return {
         "family_id": family_id,
         "seed_version": seed_version,
@@ -937,7 +933,9 @@ def main_workflow() -> dict[str, object]:
             ],
             "tool_output_mapping_specs": [
                 mapping(prefix, "result.spreadsheet", "spreadsheet_evidence"),
-                mapping(prefix, "result.original_filename", "spreadsheet_source_filename"),
+                mapping(
+                    prefix, "result.original_filename", "spreadsheet_source_filename"
+                ),
                 mapping(
                     prefix,
                     "result.spreadsheet.planning_view",
@@ -1585,9 +1583,10 @@ def build_bundle() -> dict[str, object]:
         "workflows": workflows,
         "authority_dependencies": authority_dependencies,
     }
-    authority_fingerprint = "sha256:" + hashlib.sha256(
-        stable_json(authority_payload).encode("utf-8")
-    ).hexdigest()
+    authority_fingerprint = (
+        "sha256:"
+        + hashlib.sha256(stable_json(authority_payload).encode("utf-8")).hexdigest()
+    )
     resolved_workflows = replace_authority_fingerprint(
         workflows,
         authority_fingerprint,
@@ -1595,7 +1594,7 @@ def build_bundle() -> dict[str, object]:
     return {
         "family_id": "spreadsheet_programme_representation_workflow_seed_bundle",
         "schema_version": "repo_seed_workflow_bundle.v1",
-        "seed_version": "15",
+        "seed_version": "17",
         "source_tag": "JVNAUTOSCI-2592",
         "managed_by": "spreadsheet_programme_workflow_vontology_service",
         "processing_authority_fingerprint": authority_fingerprint,
diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py
index affe7623a..f9adca5e6 100644
--- a/src/backend/integrations/internal_mcp/catalogue.py
+++ b/src/backend/integrations/internal_mcp/catalogue.py
@@ -839,6 +839,7 @@ def _coerce_scholarly_materialisation_metadata(
 _CREATE_CONCEPTS_SCOPE_DEFAULT = "user_org_default"
 _CREATE_CONCEPTS_SCOPE_ORGANISATION_GENERAL = "organisation_general"
 _CREATE_CONCEPTS_SCOPE_GLOBAL_GENERAL = "global_general"
+_CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY = "canonical_id_only"
 _CREATE_CONCEPTS_SCOPE_MODE_ALIASES: dict[str, str] = {
     "default": _CREATE_CONCEPTS_SCOPE_DEFAULT,
     "user_org_default": _CREATE_CONCEPTS_SCOPE_DEFAULT,
@@ -863,6 +864,12 @@ def _normalise_create_concepts_scope_mode(raw_value: Any) -> str | None:
 
 
 def _create_concepts(**kwargs):
+    from .transport import raise_if_internal_mcp_cancelled
+
+    # A handler may sit in the bounded executor queue until after its caller's
+    # deadline. Never begin a write-side preflight in that state.
+    raise_if_internal_mcp_cancelled()
+
     from ...vontology.utils_vontology import create_vontology_concept
     from ...vontology.code_concepts_registry import PREDICATE_TYPE_ID
     from ...services.create_concepts_duplicate_guard_service import (
@@ -888,6 +895,36 @@ def _create_concepts(**kwargs):
         else str(allow_duplicate_instances_raw).strip().lower()
         in {"1", "true", "yes", "on"}
     )
+    raw_duplicate_resolution_mode = kwargs.get("duplicate_resolution_mode")
+    canonical_id_only = False
+    if raw_duplicate_resolution_mode is not None:
+        canonical_id_only = (
+            isinstance(raw_duplicate_resolution_mode, str)
+            and raw_duplicate_resolution_mode
+            == _CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY
+        )
+        if not canonical_id_only:
+            return make_error_response(
+                "invalid_parameter",
+                (
+                    "Invalid duplicate_resolution_mode "
+                    f"'{raw_duplicate_resolution_mode}'."
+                ),
+                details={
+                    "duplicate_resolution_mode": raw_duplicate_resolution_mode,
+                    "supported_duplicate_resolution_modes": [
+                        _CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY
+                    ],
+                    "default_behaviour": "canonical identity then semantic name resolution",
+                },
+                suggestions=[
+                    "Omit duplicate_resolution_mode to preserve semantic duplicate resolution",
+                    (
+                        "Use duplicate_resolution_mode='canonical_id_only' only "
+                        "for deterministic stable identities"
+                    ),
+                ],
+            )
     raw_scope_mode = kwargs.get("scope_mode")
     if raw_scope_mode is None:
         raw_scope_mode = kwargs.get("visibility_scope_mode")
@@ -1009,6 +1046,9 @@ def _create_concepts(**kwargs):
         )
 
     parent_resolution = resolve_parent_for_create_concepts(parent_id)
+    # Parent resolution can involve bounded fallback reads. Honour a transport
+    # timeout before progressing to any item-level work.
+    raise_if_internal_mcp_cancelled()
     if not parent_resolution.success:
         canonical_parent = parent_resolution.canonical_parent_id
         related_ids = [
@@ -1055,6 +1095,9 @@ def _create_concepts(**kwargs):
 
     results = []
     for concept_data in concepts:
+        # This is both the batch boundary and the safe cancellation point after
+        # the preceding concept's complete logical write bundle.
+        raise_if_internal_mcp_cancelled()
         if not isinstance(concept_data, dict):
             results.append({"error": "Concept must be an object", "data": concept_data})
             continue
@@ -1085,6 +1128,11 @@ def _create_concepts(**kwargs):
             parent_id_for_concept=parent_id_for_concept,
             preferred_language="en-NZ",
             allow_duplicate_instances=allow_duplicate_instances,
+            duplicate_resolution_mode=(
+                _CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY
+                if canonical_id_only
+                else None
+            ),
         )
         if duplicate_match is not None:
             result = build_duplicate_prevented_create_concepts_result(
@@ -1098,6 +1146,9 @@ def _create_concepts(**kwargs):
             results.append(result)
             continue
 
+        # Duplicate preflights are read-only and may consume the entire
+        # transport budget. Never start an insert after cancellation.
+        raise_if_internal_mcp_cancelled()
         result = create_vontology_concept(
             parent_id=parent_id_for_concept,
             new_concept_name=name,
@@ -6971,18 +7022,27 @@ def _concepts_create_input_schema() -> Schema:
         },
         optional={
             "allow_duplicate_instances": (bool,),
+            "duplicate_resolution_mode": (str, type(None)),
             "namespace": (str, type(None)),
             "scope_mode": (str, type(None)),
             "visibility_scope_mode": (str, type(None)),
             "created_by_concept_id": (str, type(None)),
             "organisation_concept_id": (str, type(None)),
         },
+        enum_values={
+            "duplicate_resolution_mode": [
+                _CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY,
+                None,
+            ],
+        },
         allow_unknown=True,
         description=(
             "create_concepts input: parent_id (str, parent concept_id), concepts (list of {name, kind?, description?, notes?}). "
             "kind: 'instance' for individuals, 'type' for subtypes (default), 'predicate' for relationships. "
             "By default, deterministic pre-create lookup blocks duplicate instances/types/predicates; "
             "set allow_duplicate_instances=true to opt into legacy instance suffixing. "
+            "For deterministic stable identities, set duplicate_resolution_mode='canonical_id_only' "
+            "to skip semantic name resolution after an exact concept-id miss; omitting it preserves the default semantic fallback. "
             "Scope defaults to authenticated user+organisation visibility when context is available. "
             "Use scope_mode='organisation_general' for organisation-shared concepts, or scope_mode='global_general' "
             "for broadly visible concepts. organisation_general requires organisation context (namespace #V#user@org or organisation_concept_id). "
@@ -16610,8 +16670,7 @@ def _workflow_create_instance(**kwargs):
     if required_worker_build_raw is None:
         required_worker_build = None
     elif (
-        isinstance(required_worker_build_raw, str)
-        and required_worker_build_raw.strip()
+        isinstance(required_worker_build_raw, str) and required_worker_build_raw.strip()
     ):
         required_worker_build = required_worker_build_raw.strip()
     else:
@@ -16757,8 +16816,7 @@ def _workflow_execute(**kwargs):
     if required_worker_build_raw is None:
         required_worker_build = None
     elif (
-        isinstance(required_worker_build_raw, str)
-        and required_worker_build_raw.strip()
+        isinstance(required_worker_build_raw, str) and required_worker_build_raw.strip()
     ):
         required_worker_build = required_worker_build_raw.strip()
     else:
@@ -17338,9 +17396,7 @@ def _workflow_instance_control_status(instance: Any) -> dict[str, Any]:
         "status": raw_status.get("status"),
         "current_state": raw_status.get("current_state"),
         "step_index": raw_status.get("step_index"),
-        "manual_resume_required": bool(
-            raw_status.get("manual_resume_required", False)
-        ),
+        "manual_resume_required": bool(raw_status.get("manual_resume_required", False)),
         "checkpoint_pause_receipt": _workflow_control_safe_value(
             raw_status.get("checkpoint_pause_receipt")
         ),
@@ -30425,7 +30481,7 @@ def _build_default_catalogue_core_definitions() -> List[MethodDefinition]:
             input_schema=_concepts_create_input_schema(),
             output_schema=_concepts_create_output_schema(),
             category="write",
-            description="Create one or more concepts (instances, types, or predicates). Each concept needs name and kind ('instance' for individuals, 'type' for subtypes/default, 'predicate' for relationships). Accepts array of {name, kind?, description?, notes?}. Default visibility is user+organisation scoped when authenticated context exists. Override with scope_mode='organisation_general' for organisation-shared concepts, or scope_mode='global_general' for broadly visible concepts when the concept is clearly general. Supports singleton arrays. Use add_names afterward for alternative names/translations.",
+            description="Create one or more concepts (instances, types, or predicates). Each concept needs name and kind ('instance' for individuals, 'type' for subtypes/default, 'predicate' for relationships). Accepts array of {name, kind?, description?, notes?}. For deterministic stable identities, duplicate_resolution_mode='canonical_id_only' skips semantic name resolution after an exact concept-id miss; omitting it preserves the default semantic fallback. Default visibility is user+organisation scoped when authenticated context exists. Override with scope_mode='organisation_general' for organisation-shared concepts, or scope_mode='global_general' for broadly visible concepts when the concept is clearly general. Supports singleton arrays. Use add_names afterward for alternative names/translations.",
         ),
         MethodDefinition(
             name="get_source_processing_marker",
@@ -31540,9 +31596,9 @@ def _build_default_catalogue_knowledge_io_definitions() -> List[MethodDefinition
     return definitions
 
 
-def _build_default_catalogue_external_integration_definitions() -> (
-    List[MethodDefinition]
-):
+def _build_default_catalogue_external_integration_definitions() -> List[
+    MethodDefinition
+]:
     jira_search_output_schema = _jira_generic_output_schema("search")
     jira_get_issue_output_schema = _jira_generic_output_schema("get_issue")
     jira_get_project_issue_types_output_schema = (
@@ -32043,9 +32099,9 @@ def _build_default_catalogue_external_integration_definitions() -> (
     return definitions
 
 
-def _build_default_catalogue_diagnostics_and_research_definitions() -> (
-    List[MethodDefinition]
-):
+def _build_default_catalogue_diagnostics_and_research_definitions() -> List[
+    MethodDefinition
+]:
     definitions: List[MethodDefinition] = [
         MethodDefinition(
             name="rag_get_status",
diff --git a/src/backend/mcp_server/vontology_mcp.json b/src/backend/mcp_server/vontology_mcp.json
index b97bcd758..5e4093c48 100644
--- a/src/backend/mcp_server/vontology_mcp.json
+++ b/src/backend/mcp_server/vontology_mcp.json
@@ -766,7 +766,7 @@
         },
         {
             "name": "create_concepts",
-            "description": "Create one or more concepts (instances, types, or predicates). Each concept needs name and kind ('instance' for individuals, 'type' for subtypes/default, 'predicate' for relationships). Accepts array of {name, kind?, description?, notes?}. Default visibility is user+organisation scoped when authenticated context exists. Override with scope_mode='organisation_general' for organisation-shared concepts, or scope_mode='global_general' for broadly visible concepts when the concept is clearly general. Supports singleton arrays. Use add_names afterward for alternative names/translations.",
+            "description": "Create one or more concepts (instances, types, or predicates). Each concept needs name and kind ('instance' for individuals, 'type' for subtypes/default, 'predicate' for relationships). Accepts array of {name, kind?, description?, notes?}. For deterministic stable identities, duplicate_resolution_mode='canonical_id_only' skips semantic name resolution after an exact concept-id miss; omitting it preserves the default semantic fallback. Default visibility is user+organisation scoped when authenticated context exists. Override with scope_mode='organisation_general' for organisation-shared concepts, or scope_mode='global_general' for broadly visible concepts when the concept is clearly general. Supports singleton arrays. Use add_names afterward for alternative names/translations.",
             "inputSchema": {
                 "type": "object",
                 "properties": {
@@ -780,6 +780,16 @@
                     "allow_duplicate_instances": {
                         "type": "boolean"
                     },
+                    "duplicate_resolution_mode": {
+                        "type": [
+                            "string",
+                            "null"
+                        ],
+                        "enum": [
+                            "canonical_id_only",
+                            null
+                        ]
+                    },
                     "namespace": {
                         "type": [
                             "string",
@@ -816,7 +826,7 @@
                     "concepts"
                 ],
                 "additionalProperties": true,
-                "description": "create_concepts input: parent_id (str, parent concept_id), concepts (list of {name, kind?, description?, notes?}). kind: 'instance' for individuals, 'type' for subtypes (default), 'predicate' for relationships. By default, deterministic pre-create lookup blocks duplicate instances/types/predicates; set allow_duplicate_instances=true to opt into legacy instance suffixing. Scope defaults to authenticated user+organisation visibility when context is available. Use scope_mode='organisation_general' for organisation-shared concepts, or scope_mode='global_general' for broadly visible concepts. organisation_general requires organisation context (namespace #V#user@org or organisation_concept_id). Optional namespace is accepted and propagated into event-triggered workflow launches for tenancy attribution. Unknown top-level fields are still tolerated for orchestrator-added context."
+                "description": "create_concepts input: parent_id (str, parent concept_id), concepts (list of {name, kind?, description?, notes?}). kind: 'instance' for individuals, 'type' for subtypes (default), 'predicate' for relationships. By default, deterministic pre-create lookup blocks duplicate instances/types/predicates; set allow_duplicate_instances=true to opt into legacy instance suffixing. For deterministic stable identities, set duplicate_resolution_mode='canonical_id_only' to skip semantic name resolution after an exact concept-id miss; omitting it preserves the default semantic fallback. Scope defaults to authenticated user+organisation visibility when context is available. Use scope_mode='organisation_general' for organisation-shared concepts, or scope_mode='global_general' for broadly visible concepts. organisation_general requires organisation context (namespace #V#user@org or organisation_concept_id). Optional namespace is accepted and propagated into event-triggered workflow launches for tenancy attribution. Unknown top-level fields are still tolerated for orchestrator-added context."
             }
         },
         {
@@ -2488,6 +2498,18 @@
                             "string",
                             "null"
                         ]
+                    },
+                    "source_fingerprint": {
+                        "type": [
+                            "string",
+                            "null"
+                        ]
+                    },
+                    "processing_authority_fingerprint": {
+                        "type": [
+                            "string",
+                            "null"
+                        ]
                     }
                 },
                 "required": [
@@ -2685,7 +2707,7 @@
         },
         {
             "name": "gmail_get_message",
-            "description": "Fetch a Gmail message for a profile using message_id from gmail_list_messages. The 'profile' parameter accepts either the configured profile alias or the authorised Gmail address. Supports Gmail API formats metadata|full|raw|minimal and returns normalised sender, subject, date, and snippet fields when available. Read-only; profile token required.",
+            "description": "Fetch a Gmail message for a profile using message_id from gmail_list_messages. The 'profile' parameter accepts either the configured profile alias or the authorised Gmail address. Supports Gmail API formats metadata|full|raw|minimal and returns normalised sender, subject, date, and snippet fields when available. Set include_body=true only for an explicit user request for body text; that returns a bounded plain-text body and truncation flag. Read-only; profile token required.",
             "inputSchema": {
                 "type": "object",
                 "properties": {
@@ -2697,6 +2719,9 @@
                     },
                     "format": {
                         "type": "string"
+                    },
+                    "include_body": {
+                        "type": "boolean"
                     }
                 },
                 "required": [
@@ -2704,7 +2729,7 @@
                     "message_id"
                 ],
                 "additionalProperties": false,
-                "description": "Fetch a Gmail message for a profile (formats: metadata|full|raw|minimal).",
+                "description": "Fetch a Gmail message for a profile (formats: metadata|full|raw|minimal). Set include_body=true only when the user explicitly requests message body text; this forces format=full and returns bounded plain-text body evidence.",
                 "x-von-argument-aliases": {
                     "profile_id": "profile",
                     "identity": "profile",
@@ -3147,7 +3172,11 @@
                     "issue_key"
                 ],
                 "additionalProperties": true,
-                "description": "jira_get_issue input: issue_key (str, required), optional fields (list of field names), and optional expand (list of Jira expand tokens such as ['changelog'])."
+                "description": "jira_get_issue input: issue_key (str, required), optional fields (list of individual field names), and optional expand (list of individual Jira expand tokens such as ['changelog']).",
+                "x-von-comma-separated-list-fields": [
+                    "fields",
+                    "expand"
+                ]
             }
         },
         {
@@ -3398,7 +3427,10 @@
                     "jql"
                 ],
                 "additionalProperties": true,
-                "description": "jira_search input: jql (str, required) plus optional max_results, next_page_token, start_at (deprecated), and fields (list of field names)."
+                "description": "jira_search input: jql (str, required) plus optional max_results, next_page_token, start_at (deprecated), and fields (list of individual field names; do not combine names into one item).",
+                "x-von-comma-separated-list-fields": [
+                    "fields"
+                ]
             }
         },
         {
@@ -3952,6 +3984,18 @@
                             "null"
                         ]
                     },
+                    "source_fingerprint": {
+                        "type": [
+                            "string",
+                            "null"
+                        ]
+                    },
+                    "processing_authority_fingerprint": {
+                        "type": [
+                            "string",
+                            "null"
+                        ]
+                    },
                     "processing_status": {
                         "type": [
                             "string",
@@ -4002,6 +4046,13 @@
                         ],
                         "items": {},
                         "additionalProperties": true
+                    },
+                    "processing_evidence": {
+                        "type": [
+                            "object",
+                            "null"
+                        ],
+                        "additionalProperties": true
                     }
                 },
                 "required": [
@@ -4519,7 +4570,10 @@
                     "query"
                 ],
                 "additionalProperties": false,
-                "description": "search_concepts input: query (str, optional - defaults to empty), match_type ('exact'|'substring'|'similarity'|'all'), min_similarity (float 0.0-1.0), filter_kind (list[str]), scope_root (str), instance_of (str concept_id - finds instances of this type), include_description (bool), include_hierarchy_path (bool), limit (int)"
+                "description": "search_concepts input: query (str, optional - defaults to empty), match_type ('exact'|'substring'|'similarity'|'all'), min_similarity (float 0.0-1.0), filter_kind (list[str]), scope_root (str), instance_of (str concept_id - finds instances of this type), include_description (bool), include_hierarchy_path (bool), limit (int)",
+                "x-von-comma-separated-list-fields": [
+                    "filter_kind"
+                ]
             }
         },
         {
@@ -6355,7 +6409,10 @@
                     "query"
                 ],
                 "additionalProperties": false,
-                "description": "search_concepts input: query (str, optional - defaults to empty), match_type ('exact'|'substring'|'similarity'|'all'), min_similarity (float 0.0-1.0), filter_kind (list[str]), scope_root (str), instance_of (str concept_id - finds instances of this type), include_description (bool), include_hierarchy_path (bool), limit (int)"
+                "description": "search_concepts input: query (str, optional - defaults to empty), match_type ('exact'|'substring'|'similarity'|'all'), min_similarity (float 0.0-1.0), filter_kind (list[str]), scope_root (str), instance_of (str concept_id - finds instances of this type), include_description (bool), include_hierarchy_path (bool), limit (int)",
+                "x-von-comma-separated-list-fields": [
+                    "filter_kind"
+                ]
             }
         },
         {
diff --git a/src/backend/services/create_concepts_duplicate_guard_service.py b/src/backend/services/create_concepts_duplicate_guard_service.py
index fa9ba6172..710dc0aa1 100644
--- a/src/backend/services/create_concepts_duplicate_guard_service.py
+++ b/src/backend/services/create_concepts_duplicate_guard_service.py
@@ -10,6 +10,9 @@
   a lookup was skipped.
 - Callers can explicitly opt-in to legacy duplicate-instance behaviour via
   ``allow_duplicate_instances=True`` when true homonyms are intended.
+- Callers with deterministic stable identities can select
+  ``duplicate_resolution_mode="canonical_id_only"`` to retain exact identity
+  reuse while skipping semantic name resolution after an exact miss.
 """
 
 from __future__ import annotations
@@ -63,7 +66,11 @@ def _extract_str_list(value: Any) -> list[str]:
         cleaned = value.strip()
         return [cleaned] if cleaned else []
     if isinstance(value, list):
-        return [str(item).strip() for item in value if isinstance(item, str) and item.strip()]
+        return [
+            str(item).strip()
+            for item in value
+            if isinstance(item, str) and item.strip()
+        ]
     return []
 
 
@@ -159,8 +166,13 @@ def find_existing_concept_for_create_concepts(
     parent_id_for_concept: str | None,
     preferred_language: str | None = None,
     allow_duplicate_instances: bool = False,
+    duplicate_resolution_mode: str | None = None,
 ) -> CreateConceptDuplicateGuardMatch | None:
-    """Return an existing concept match when duplicate creation should be blocked."""
+    """Return an existing concept match when duplicate creation should be blocked.
+
+    ``canonical_id_only`` retains the normal exact match and scope checks, but
+    deliberately returns after an exact miss instead of searching text values.
+    """
 
     normalised_kind = _normalise_kind(kind)
     scope = _guard_scope_for_request(
@@ -185,6 +197,9 @@ def find_existing_concept_for_create_concepts(
                 guard_scope=scope,
             )
 
+    if duplicate_resolution_mode == "canonical_id_only":
+        return None
+
     # Secondary bounded lookup: deterministic resolver over hasName/code aliases.
     try:
         from .concept_resolution_service import resolve_concept_by_name
@@ -222,7 +237,9 @@ def find_existing_concept_for_create_concepts(
     if not isinstance(resolved_id, str) or not resolved_id.strip():
         return None
 
-    canonical_resolved = canonicalise_vontology_concept_id(resolved_id) or resolved_id.strip()
+    canonical_resolved = (
+        canonicalise_vontology_concept_id(resolved_id) or resolved_id.strip()
+    )
     if canonical_requested_id and canonical_resolved == canonical_requested_id:
         return None
 
diff --git a/src/backend/services/feature_flags.py b/src/backend/services/feature_flags.py
index af82b0dfa..72fcc0650 100644
--- a/src/backend/services/feature_flags.py
+++ b/src/backend/services/feature_flags.py
@@ -47,7 +47,22 @@ def get_workflow_discovery_cache_invalidation_enabled(
     *,
     default: bool = True,
 ) -> bool:
-    """Return whether mutation paths should invalidate workflow discovery caches."""
+    """Return whether mutation paths should invalidate workflow discovery caches.
+
+    Event-workflow-owned mutations may suppress their own recursive launch and
+    routing-cache fan-out through a request-local context. Import lazily to
+    avoid coupling ordinary feature-flag reads to workflow service start-up.
+    """
+
+    try:
+        from .workflow_event_integration_service import (
+            event_workflow_launches_suppressed,
+        )
+
+        if event_workflow_launches_suppressed():
+            return False
+    except ImportError:
+        pass
 
     return _read_env_flag(
         "VON_WORKFLOW_DISCOVERY_CACHE_INVALIDATION_ENABLE",
diff --git a/src/backend/services/kr_materialisation_guard_service.py b/src/backend/services/kr_materialisation_guard_service.py
index f7ebbc51a..7ea5e9e1b 100644
--- a/src/backend/services/kr_materialisation_guard_service.py
+++ b/src/backend/services/kr_materialisation_guard_service.py
@@ -13,6 +13,7 @@
 
 
 KR_MATERIALISATION_GUARD_SCHEMA_VERSION = "kr_materialisation_guard.v1"
+GUARDED_CREATE_DUPLICATE_RESOLUTION_MODE = "canonical_id_only"
 _MAX_GUARD_CONCEPT_SLOTS = 256
 _MAX_GUARD_RELATIONSHIP_RULES = 512
 _MAX_DESCRIPTION_CHARS = 100_000
@@ -23,9 +24,7 @@ def _text(value: Any) -> str:
 
 
 def _sequence(value: Any) -> list[Any]:
-    if not isinstance(value, Sequence) or isinstance(
-        value, (str, bytes, bytearray)
-    ):
+    if not isinstance(value, Sequence) or isinstance(value, (str, bytes, bytearray)):
         return []
     return list(value)
 
@@ -36,6 +35,7 @@ def _reject(code: str, *, phase: str) -> dict[str, Any]:
         "guard_applied": True,
         "guard_passed": False,
         "guard_phase": phase,
+        "duplicate_resolution_mode": None,
         "error_code": code,
         "blocking_reason": code,
     }
@@ -53,6 +53,9 @@ def _pass(
         "guard_applied": guard_applied,
         "guard_passed": True,
         "guard_phase": phase,
+        "duplicate_resolution_mode": (
+            GUARDED_CREATE_DUPLICATE_RESOLUTION_MODE if guard_applied else None
+        ),
         "concept_spec_count": concept_count,
         "relationship_spec_count": relationship_count,
         "blocking_reason": None,
@@ -116,9 +119,7 @@ def _normalise_contract(
 
     fixed_ids = {
         _text(item)
-        for item in _sequence(
-            guard_contract.get("fixed_authorised_concept_ids")
-        )
+        for item in _sequence(guard_contract.get("fixed_authorised_concept_ids"))
         if _text(item)
     }
     rules: list[dict[str, Any]] = []
@@ -183,13 +184,11 @@ def _normalise_contract(
         )
 
     try:
-        max_concepts = int(
-            guard_contract.get("max_concept_specs", len(slots))
-        )
+        max_concepts = int(guard_contract.get("max_concept_specs", len(slots)))
         max_relationships = int(
-            guard_contract.get("max_relationship_specs", sum(
-                rule["maximum_count"] for rule in rules
-            ))
+            guard_contract.get(
+                "max_relationship_specs", sum(rule["maximum_count"] for rule in rules)
+            )
         )
     except (TypeError, ValueError):
         return None, "kr_materialisation_guard_contract_bounds_invalid"
@@ -210,15 +209,11 @@ def _normalise_contract(
         "max_relationships": max_relationships,
         "require_all_slots": guard_contract.get("require_all_concept_slots")
         is not False,
-        "reject_unreferenced": guard_contract.get(
-            "reject_unreferenced_concepts"
-        )
+        "reject_unreferenced": guard_contract.get("reject_unreferenced_concepts")
         is not False,
         "require_fixed_ids_referenced": {
             _text(item)
-            for item in _sequence(
-                guard_contract.get("require_fixed_ids_referenced")
-            )
+            for item in _sequence(guard_contract.get("require_fixed_ids_referenced"))
             if _text(item)
         },
     }, None
@@ -240,12 +235,8 @@ def _validate_concept_specs(
 ) -> tuple[dict[str, Mapping[str, Any]] | None, str | None]:
     raw_specs = _sequence(concept_specs)
     slots = contract["slots"]
-    if (
-        len(raw_specs) > int(contract["max_concepts"])
-        or (
-            bool(contract["require_all_slots"])
-            and len(raw_specs) != len(slots)
-        )
+    if len(raw_specs) > int(contract["max_concepts"]) or (
+        bool(contract["require_all_slots"]) and len(raw_specs) != len(slots)
     ):
         return None, "kr_materialisation_guard_concept_count_rejected"
 
@@ -286,11 +277,9 @@ def _validate_concept_specs(
             if (
                 _text(create_spec.get("name")) != slot["stable_name"]
                 or _text(create_spec.get("kind")).lower() != slot["target_kind"]
-                or len(_text(create_spec.get("description")))
-                > _MAX_DESCRIPTION_CHARS
+                or len(_text(create_spec.get("description"))) > _MAX_DESCRIPTION_CHARS
                 or _text(
-                    raw_spec.get("existing_concept_id")
-                    or raw_spec.get("concept_id")
+                    raw_spec.get("existing_concept_id") or raw_spec.get("concept_id")
                 )
             ):
                 return None, "kr_materialisation_guard_create_payload_rejected"
@@ -300,8 +289,7 @@ def _validate_concept_specs(
             )
             if (
                 existing_concept_id == ""
-                or existing_concept_id
-                not in slot["allowed_existing_concept_ids"]
+                or existing_concept_id not in slot["allowed_existing_concept_ids"]
                 or raw_create_specs not in (None, [])
             ):
                 return None, "kr_materialisation_guard_reuse_payload_rejected"
@@ -454,9 +442,7 @@ def _validate_relationship_specs(
         }
         if not required_slots.issubset(referenced_slots):
             return None, "kr_materialisation_guard_unreferenced_concept_rejected"
-    if not set(contract["require_fixed_ids_referenced"]).issubset(
-        referenced_fixed_ids
-    ):
+    if not set(contract["require_fixed_ids_referenced"]).issubset(referenced_fixed_ids):
         return None, "kr_materialisation_guard_fixed_endpoint_missing"
     for raw_spec in raw_specs:
         if not isinstance(raw_spec, Mapping):
@@ -544,8 +530,7 @@ def validate_kr_materialisation_guard(
     )
     if requested is None:
         return _reject(
-            relationship_error
-            or "kr_materialisation_guard_relationships_rejected",
+            relationship_error or "kr_materialisation_guard_relationships_rejected",
             phase=phase_text,
         )
     if phase_text == "plan":
@@ -596,9 +581,7 @@ def validate_kr_materialisation_guard(
         source_id = _text(raw_spec.get("source_id"))
         predicate, predicate_error = _normalise_relationship_predicate(
             raw_spec,
-            invalid_code=(
-                "kr_materialisation_guard_resolved_relationship_invalid"
-            ),
+            invalid_code=("kr_materialisation_guard_resolved_relationship_invalid"),
         )
         if predicate_error is not None or predicate is None:
             return _reject(
@@ -640,6 +623,7 @@ def validate_kr_materialisation_guard(
 
 
 __all__ = [
+    "GUARDED_CREATE_DUPLICATE_RESOLUTION_MODE",
     "KR_MATERIALISATION_GUARD_SCHEMA_VERSION",
     "validate_kr_materialisation_guard",
 ]
diff --git a/src/backend/services/relationship_write_service.py b/src/backend/services/relationship_write_service.py
index 0df82d7c4..3546097bf 100644
--- a/src/backend/services/relationship_write_service.py
+++ b/src/backend/services/relationship_write_service.py
@@ -31,7 +31,10 @@
     get_structural_predicate_aliases,
     get_structural_inverse_map,
 )
-from .feature_flags import get_event_workflow_integration_enabled
+from .feature_flags import (
+    get_event_workflow_integration_enabled,
+    get_workflow_discovery_cache_invalidation_enabled,
+)
 from ..vontology.utils_vontology import is_predicate, is_type
 from ..vontology.code_concepts_registry import (
     build_virtual_concept_doc,
@@ -128,6 +131,9 @@ def _invalidate_workflow_routing_projection_for_relationship_change(
 ) -> None:
     """Best-effort invalidation for workflow graph/type routing projections."""
 
+    if not get_workflow_discovery_cache_invalidation_enabled(default=True):
+        return
+
     predicate_text = str(predicate or "").strip()
     if not predicate_text:
         return
@@ -564,7 +570,9 @@ def add_structural_relationship(
             sync_sources.append(target_id)
         _sync_relationship_extent_index_for_sources(sync_sources)
         try:
-            from .workflow_event_integration_service import EVENT_TYPE_RELATIONSHIP_ADDED
+            from .workflow_event_integration_service import (
+                EVENT_TYPE_RELATIONSHIP_ADDED,
+            )
 
             _emit_relationship_mutation_event(
                 event_type=EVENT_TYPE_RELATIONSHIP_ADDED,
@@ -655,7 +663,9 @@ def add_dynamic_relationship(
     if bool(result.get("modified")):
         _sync_relationship_extent_index_for_sources([source_id])
         try:
-            from .workflow_event_integration_service import EVENT_TYPE_RELATIONSHIP_ADDED
+            from .workflow_event_integration_service import (
+                EVENT_TYPE_RELATIONSHIP_ADDED,
+            )
 
             _emit_relationship_mutation_event(
                 event_type=EVENT_TYPE_RELATIONSHIP_ADDED,
@@ -703,9 +713,11 @@ def add_relationship(
         predicate_for_invalidation = predicate
 
     if isinstance(result, Mapping) and bool(result.get("success")):
-        modified = bool(result.get("forward_modified")) or bool(
-            result.get("modified")
-        ) or bool(result.get("created"))
+        modified = (
+            bool(result.get("forward_modified"))
+            or bool(result.get("modified"))
+            or bool(result.get("created"))
+        )
         if modified:
             _invalidate_workflow_routing_projection_for_relationship_change(
                 source_id=source_id,
diff --git a/src/backend/services/text_value_service.py b/src/backend/services/text_value_service.py
index 064df697c..ace47e72c 100644
--- a/src/backend/services/text_value_service.py
+++ b/src/backend/services/text_value_service.py
@@ -22,7 +22,10 @@
     TextValueModel,
     TextRelationModel,
 )
-from .feature_flags import get_event_workflow_integration_enabled
+from .feature_flags import (
+    get_event_workflow_integration_enabled,
+    get_workflow_discovery_cache_invalidation_enabled,
+)
 from ..security.access_control import (
     can_access_concept,
     filter_accessible_concept_ids,
@@ -70,6 +73,9 @@ def _invalidate_workflow_routing_projection_for_text_relation_change(
 ) -> None:
     """Best-effort invalidation for workflow-routing support projections."""
 
+    if not get_workflow_discovery_cache_invalidation_enabled(default=True):
+        return
+
     predicate_text = str(predicate or "").strip()
     if not predicate_text:
         return
@@ -431,10 +437,7 @@ def get_texts_for_concepts(
         if not isinstance(raw_id, str):
             continue
         subject_concept_id = raw_id.strip()
-        if (
-            not subject_concept_id
-            or subject_concept_id in seen_subject_ids
-        ):
+        if not subject_concept_id or subject_concept_id in seen_subject_ids:
             continue
         seen_subject_ids.add(subject_concept_id)
         candidate_subject_ids.append(subject_concept_id)
@@ -702,7 +705,9 @@ def get_preferred_texts_for_concepts(
         relation_filter["predicate"] = {"$in": allowed_predicates}
 
     relation_limit = max(len(ordered_subject_ids) * max(limit_per_concept, 1), 1)
-    relations = list(TextRelationsRepository.find(relation_filter, limit=relation_limit))
+    relations = list(
+        TextRelationsRepository.find(relation_filter, limit=relation_limit)
+    )
     if not relations:
         return {}
 
@@ -982,7 +987,9 @@ def delete_text_relation(
     tv_id = rel.get("object_text_id")
     predicate = rel.get("predicate")
     TextRelationsRepository.delete_one({"_id": ObjectId(relation_id)})
-    _invalidate_stats_for_predicate_change(predicate if isinstance(predicate, str) else None)
+    _invalidate_stats_for_predicate_change(
+        predicate if isinstance(predicate, str) else None
+    )
     _invalidate_workflow_routing_projection_for_text_relation_change(
         subject_concept_id=subject_concept_id,
         predicate=predicate if isinstance(predicate, str) else None,
diff --git a/src/backend/services/workflow_event_integration_service.py b/src/backend/services/workflow_event_integration_service.py
index 33ef25260..33e045cc7 100644
--- a/src/backend/services/workflow_event_integration_service.py
+++ b/src/backend/services/workflow_event_integration_service.py
@@ -7,12 +7,14 @@
 
 from __future__ import annotations
 
+from contextlib import contextmanager
+from contextvars import ContextVar
 from datetime import datetime, timedelta, timezone
 import hashlib
 import logging
 import os
 from time import perf_counter
-from typing import Any, Mapping
+from typing import Any, Iterator, Mapping
 
 from .episode_evaluation_workflow_contracts import (
     EPISODE_EVALUATION_AUTOTRIGGER_ENV,
@@ -63,9 +65,48 @@
 EVENT_TYPE_FILE_COPY_UPLOADED = "file_copy.uploaded"
 
 FILE_COPY_UPLOAD_EVENT_BINDING_ENABLE_ENV = "VON_FILE_COPY_UPLOAD_EVENT_BINDING_ENABLE"
-FILE_COPY_UPLOAD_EVENT_BINDING_MANAGED_BY = "workflow_event_integration:file_copy_upload"
+FILE_COPY_UPLOAD_EVENT_BINDING_MANAGED_BY = (
+    "workflow_event_integration:file_copy_upload"
+)
 
 _DEFAULT_BINDINGS_ENSURED = False
+_EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REASONS: ContextVar[tuple[str, ...]] = ContextVar(
+    "event_workflow_launch_suppression_reasons",
+    default=(),
+)
+
+
+def current_event_workflow_launch_suppression_reason() -> str | None:
+    """Return the innermost request-local event-launch suppression reason."""
+
+    reasons = _EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REASONS.get()
+    return reasons[-1] if reasons else None
+
+
+def event_workflow_launches_suppressed() -> bool:
+    """Return whether event-driven workflow launch is suppressed in this context."""
+
+    return bool(_EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REASONS.get())
+
+
+@contextmanager
+def suppress_event_workflow_launches(reason: str) -> Iterator[None]:
+    """Suppress event-workflow fan-out within one nested execution context.
+
+    The context is copied into internal MCP handler workers by the bounded
+    transport, while independent threads retain their own default. Resetting
+    the token preserves an outer suppression reason when scopes are nested.
+    """
+
+    cleaned_reason = str(reason or "").strip() or "unspecified"
+    current_reasons = _EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REASONS.get()
+    token = _EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REASONS.set(
+        (*current_reasons, cleaned_reason)
+    )
+    try:
+        yield
+    finally:
+        _EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REASONS.reset(token)
 
 
 def _clean_text(value: Any) -> str | None:
@@ -1007,6 +1048,19 @@ def launch_event_workflow(
 ) -> dict[str, Any]:
     """Launch configured durable workflow binding(s) for an event."""
 
+    suppression_reason = current_event_workflow_launch_suppression_reason()
+    if suppression_reason is not None:
+        return {
+            "success": True,
+            "triggered": False,
+            "outcome": "suppressed",
+            "event_type": event_type,
+            "event_id": str(event_id or "").strip() or None,
+            "reason": "event_workflow_launch_suppressed",
+            "suppression_reason": suppression_reason,
+            "event_workflow_launch_suppressed": True,
+        }
+
     integration_enabled = get_event_workflow_integration_enabled(default=True)
     if not integration_enabled:
         return {
diff --git a/src/backend/workflows/repo_seed_bundles/kr_materialisation_workflow_seed_bundle.json b/src/backend/workflows/repo_seed_bundles/kr_materialisation_workflow_seed_bundle.json
index 39917860a..deb31ddbb 100644
--- a/src/backend/workflows/repo_seed_bundles/kr_materialisation_workflow_seed_bundle.json
+++ b/src/backend/workflows/repo_seed_bundles/kr_materialisation_workflow_seed_bundle.json
@@ -2,7 +2,7 @@
   "family_id": "#V#kr_design_materialisation_workflow_family",
   "managed_by": "kr_materialisation_workflow_vontology_service",
   "schema_version": "repo_seed_workflow_bundle.v1",
-  "seed_version": "2",
+  "seed_version": "4",
   "source_tag": "JVNAUTOSCI-2271",
   "supported_action_ids": [
     "llm.action",
@@ -267,6 +267,12 @@
                 "context_key": "kr_concept_parent_id",
                 "required": true,
                 "tool_param": "parent_id"
+              },
+              {
+                "concept_id": "#V#workflow_mapping_kr_design_concept_materialisation_item_workflow_create_concept_kr_concept_duplicate_resolution_mode_to_duplicate_resolution_mode_parameter",
+                "context_key": "kr_concept_duplicate_resolution_mode",
+                "required": false,
+                "tool_param": "duplicate_resolution_mode"
               }
             ],
             "execution_mode": "deterministic",
@@ -287,6 +293,10 @@
                 "scope_mode",
                 "user_org_default"
               ],
+              [
+                "suppress_event_workflow_launches",
+                true
+              ],
               [
                 "tool_name",
                 "create_concepts"
@@ -559,6 +569,10 @@
                 "predicate",
                 "type_of"
               ],
+              [
+                "suppress_event_workflow_launches",
+                true
+              ],
               [
                 "tool_name",
                 "add_relationship"
@@ -617,6 +631,10 @@
                 "predicate",
                 "instance_of"
               ],
+              [
+                "suppress_event_workflow_launches",
+                true
+              ],
               [
                 "tool_name",
                 "add_relationship"
@@ -687,6 +705,10 @@
                 "predicate",
                 "hasDescription"
               ],
+              [
+                "suppress_event_workflow_launches",
+                true
+              ],
               [
                 "tool_name",
                 "upsert_singleton_text_relation"
@@ -1120,6 +1142,10 @@
             "on_failure_state": "summarise_blocker",
             "state_id": "assert_relationship",
             "static_input_bindings": [
+              [
+                "suppress_event_workflow_launches",
+                true
+              ],
               [
                 "tool_name",
                 "add_relationship"
@@ -1609,11 +1635,17 @@
                 "concept_id": "#V#workflow_mapping_tool_field_kr_design_materialisation_workflow_validate_materialisation_plan_blocking_reason_to_kr_materialisation_blocking_reason",
                 "context_key": "kr_materialisation_blocking_reason",
                 "tool_output_field": "blocking_reason"
+              },
+              {
+                "concept_id": "#V#workflow_mapping_tool_field_kr_design_materialisation_workflow_validate_materialisation_plan_duplicate_resolution_mode_to_kr_concept_duplicate_resolution_mode",
+                "context_key": "kr_concept_duplicate_resolution_mode",
+                "tool_output_field": "duplicate_resolution_mode"
               }
             ],
             "writes_context_keys": [
               "kr_materialisation_guard_passed",
-              "kr_materialisation_blocking_reason"
+              "kr_materialisation_blocking_reason",
+              "kr_concept_duplicate_resolution_mode"
             ]
           },
           {
@@ -1682,6 +1714,10 @@
                 "max_concurrency",
                 1
               ],
+              [
+                "stop_on_error",
+                true
+              ],
               [
                 "max_items",
                 24
diff --git a/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json b/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json
index fe33f3f3c..c808fae1c 100644
--- a/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json
+++ b/src/backend/workflows/repo_seed_bundles/spreadsheet_programme_representation_workflow_seed_bundle.json
@@ -3,14 +3,14 @@
   "managed_by": "spreadsheet_programme_workflow_vontology_service",
   "processing_authority_dependencies": {
     "kr_materialisation_workflow_seed_bundle": {
-      "canonical_payload_sha256": "sha256:ce53d8c2ba252fb1b251f778444a797cb4c436fc2bbb94fa50391b5c7a1307f6",
+      "canonical_payload_sha256": "sha256:d01ea4c91f61802ebd4c8935d5e6d8c817f2f2559db495da965c38add43a0b33",
       "family_id": "#V#kr_design_materialisation_workflow_family",
-      "seed_version": "2"
+      "seed_version": "4"
     }
   },
-  "processing_authority_fingerprint": "sha256:62aaed4941a0e0dbb16d5c4bd6c2fc8b09ff34f1e905c561c40004c04d840c87",
+  "processing_authority_fingerprint": "sha256:4a23437988d6e745aac1a2515ed8b1231b8b3fb0f63abe078c107d176b74f79f",
   "schema_version": "repo_seed_workflow_bundle.v1",
-  "seed_version": "15",
+  "seed_version": "17",
   "source_tag": "JVNAUTOSCI-2592",
   "support_concepts": [
     {
@@ -306,7 +306,7 @@
               },
               {
                 "key": "processing_authority_fingerprint",
-                "value": "sha256:62aaed4941a0e0dbb16d5c4bd6c2fc8b09ff34f1e905c561c40004c04d840c87"
+                "value": "sha256:4a23437988d6e745aac1a2515ed8b1231b8b3fb0f63abe078c107d176b74f79f"
               }
             ],
             "tool_output_mapping_specs": [
@@ -702,7 +702,7 @@
               },
               {
                 "key": "processing_authority_fingerprint",
-                "value": "sha256:62aaed4941a0e0dbb16d5c4bd6c2fc8b09ff34f1e905c561c40004c04d840c87"
+                "value": "sha256:4a23437988d6e745aac1a2515ed8b1231b8b3fb0f63abe078c107d176b74f79f"
               }
             ],
             "tool_output_mapping_specs": [
@@ -776,7 +776,7 @@
               },
               {
                 "key": "processing_authority_fingerprint",
-                "value": "sha256:62aaed4941a0e0dbb16d5c4bd6c2fc8b09ff34f1e905c561c40004c04d840c87"
+                "value": "sha256:4a23437988d6e745aac1a2515ed8b1231b8b3fb0f63abe078c107d176b74f79f"
               }
             ],
             "tool_output_mapping_specs": [
@@ -1234,7 +1234,7 @@
               },
               {
                 "key": "processing_authority_fingerprint",
-                "value": "sha256:62aaed4941a0e0dbb16d5c4bd6c2fc8b09ff34f1e905c561c40004c04d840c87"
+                "value": "sha256:4a23437988d6e745aac1a2515ed8b1231b8b3fb0f63abe078c107d176b74f79f"
               }
             ],
             "tool_output_mapping_specs": [
@@ -1557,7 +1557,7 @@
               },
               {
                 "key": "processing_authority_fingerprint",
-                "value": "sha256:62aaed4941a0e0dbb16d5c4bd6c2fc8b09ff34f1e905c561c40004c04d840c87"
+                "value": "sha256:4a23437988d6e745aac1a2515ed8b1231b8b3fb0f63abe078c107d176b74f79f"
               }
             ],
             "tool_output_mapping_specs": [
@@ -1625,7 +1625,7 @@
               },
               {
                 "key": "processing_authority_fingerprint",
-                "value": "sha256:62aaed4941a0e0dbb16d5c4bd6c2fc8b09ff34f1e905c561c40004c04d840c87"
+                "value": "sha256:4a23437988d6e745aac1a2515ed8b1231b8b3fb0f63abe078c107d176b74f79f"
               }
             ],
             "tool_output_mapping_specs": [
diff --git a/src/backend/workflows/workflow_mcp_tool_actions.py b/src/backend/workflows/workflow_mcp_tool_actions.py
index 6a7f2ee86..c1c6ceb89 100644
--- a/src/backend/workflows/workflow_mcp_tool_actions.py
+++ b/src/backend/workflows/workflow_mcp_tool_actions.py
@@ -36,9 +36,14 @@
     normalise_workflow_execution_side_effect_policy,
     normalise_workflow_step_mutation_authority_spec,
 )
+
 logger = logging.getLogger(__name__)
 
 WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID = "workflow_mcp.invoke_tool"
+SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT = "suppress_event_workflow_launches"
+EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REQUESTED_OUTPUT = (
+    "event_workflow_launch_suppression_requested"
+)
 
 WORKFLOW_MCP_TOOL_NAME_INPUT_KEYS: tuple[str, ...] = (
     "tool_name",
@@ -56,7 +61,11 @@
     "mcp_payload",
 )
 _CONTROL_INPUT_KEYS: frozenset[str] = frozenset(
-    (*_TOOL_NAME_INPUT_KEYS, *_PAYLOAD_INPUT_KEYS)
+    (
+        *_TOOL_NAME_INPUT_KEYS,
+        *_PAYLOAD_INPUT_KEYS,
+        SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT,
+    )
 )
 _WRITE_POLICY_METADATA_KEYS: tuple[str, ...] = (
     "workflow_execution_side_effect_policy",
@@ -138,7 +147,9 @@ def _normalise_tool_payload(inputs: Mapping[str, Any]) -> dict[str, Any]:
             return {
                 str(payload_key): payload_value
                 for payload_key, payload_value in value.items()
-                if isinstance(payload_key, str) and str(payload_key).strip()
+                if isinstance(payload_key, str)
+                and str(payload_key).strip()
+                and str(payload_key).strip() != SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT
             }
 
     return {
@@ -150,14 +161,33 @@ def _normalise_tool_payload(inputs: Mapping[str, Any]) -> dict[str, Any]:
     }
 
 
+def _event_launch_suppression_requested(inputs: Mapping[str, Any]) -> bool:
+    return inputs.get(SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT) is True
+
+
+def _with_event_launch_suppression_telemetry(
+    result: WorkflowActionResult,
+    *,
+    requested: bool,
+) -> WorkflowActionResult:
+    result.outputs[EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REQUESTED_OUTPUT] = bool(requested)
+    return result
+
+
 def _has_explicit_write_policy(metadata: Mapping[str, Any]) -> bool:
     if metadata.get("mutation_authority") is not None:
-        return normalise_workflow_step_mutation_authority_spec(
-            metadata.get("mutation_authority")
-        ) is not None
+        return (
+            normalise_workflow_step_mutation_authority_spec(
+                metadata.get("mutation_authority")
+            )
+            is not None
+        )
 
     for key in _WRITE_POLICY_METADATA_KEYS:
-        if normalise_workflow_execution_side_effect_policy(metadata.get(key)) is not None:
+        if (
+            normalise_workflow_execution_side_effect_policy(metadata.get(key))
+            is not None
+        ):
             return True
     return False
 
@@ -341,14 +371,19 @@ def _handle_workflow_mcp_invoke_tool(
     request: WorkflowActionRequest,
 ) -> WorkflowActionResult:
     inputs = dict(request.inputs or {})
+    event_launch_suppression_requested = _event_launch_suppression_requested(inputs)
     requested_tool_name = extract_static_workflow_mcp_tool_name(inputs)
     if not requested_tool_name:
-        return WorkflowActionResult(
-            status="failed",
-            error="workflow_mcp_tool_name_missing",
+        return _with_event_launch_suppression_telemetry(
+            WorkflowActionResult(
+                status="failed",
+                error="workflow_mcp_tool_name_missing",
+            ),
+            requested=event_launch_suppression_requested,
         )
 
     payload = _normalise_tool_payload(inputs)
+    payload.pop(SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT, None)
     resolved_tool_name = requested_tool_name
 
     try:
@@ -363,13 +398,16 @@ def _handle_workflow_mcp_invoke_tool(
         )
         method_definition = gateway.get_method_definition(resolved_tool_name)
         if method_definition is None:
-            return WorkflowActionResult(
-                status="failed",
-                error=f"workflow_mcp_tool_not_registered:{requested_tool_name}",
-                outputs={
-                    "mcp_tool": requested_tool_name,
-                    "mcp_requested_tool": requested_tool_name,
-                },
+            return _with_event_launch_suppression_telemetry(
+                WorkflowActionResult(
+                    status="failed",
+                    error=f"workflow_mcp_tool_not_registered:{requested_tool_name}",
+                    outputs={
+                        "mcp_tool": requested_tool_name,
+                        "mcp_requested_tool": requested_tool_name,
+                    },
+                ),
+                requested=event_launch_suppression_requested,
             )
 
         apply_runtime_defaults_to_mcp_payload(
@@ -425,16 +463,19 @@ def _handle_workflow_mcp_invoke_tool(
             request=request,
             method_definition=method_definition,
         ):
-            return WorkflowActionResult(
-                status="failed",
-                error=f"workflow_mcp_write_policy_missing:{resolved_tool_name}",
-                outputs={
-                    "mcp_tool": resolved_tool_name,
-                    "mcp_requested_tool": requested_tool_name,
-                    "mcp_resolved_tool": resolved_tool_name,
-                    "mutation_guardrail_blocked": True,
-                    "write_policy_reason": "workflow_mcp_write_policy_missing",
-                },
+            return _with_event_launch_suppression_telemetry(
+                WorkflowActionResult(
+                    status="failed",
+                    error=f"workflow_mcp_write_policy_missing:{resolved_tool_name}",
+                    outputs={
+                        "mcp_tool": resolved_tool_name,
+                        "mcp_requested_tool": requested_tool_name,
+                        "mcp_resolved_tool": resolved_tool_name,
+                        "mutation_guardrail_blocked": True,
+                        "write_policy_reason": "workflow_mcp_write_policy_missing",
+                    },
+                ),
+                requested=event_launch_suppression_requested,
             )
 
         blocked_result = enforce_workflow_mcp_write_guardrails(
@@ -443,7 +484,10 @@ def _handle_workflow_mcp_invoke_tool(
             method_definition=method_definition,
         )
         if blocked_result is not None:
-            return blocked_result
+            return _with_event_launch_suppression_telemetry(
+                blocked_result,
+                requested=event_launch_suppression_requested,
+            )
 
         target_validation = validate_tool_target_contract(
             tool_name=resolved_tool_name,
@@ -460,22 +504,25 @@ def _handle_workflow_mcp_invoke_tool(
         )
         if not target_validation.ok:
             error_code = target_validation.first_error_code()
-            return WorkflowActionResult(
-                status="failed",
-                error=(
-                    f"workflow_mcp_target_contract_validation_failed:"
-                    f"{resolved_tool_name}:{error_code or 'invalid_target'}"
-                ),
-                outputs={
-                    "mcp_tool": resolved_tool_name,
-                    "mcp_requested_tool": requested_tool_name,
-                    "mcp_resolved_tool": resolved_tool_name,
-                    "target_contract_validation_failed": True,
-                    "target_contract_validation_error_code": error_code,
-                    "tool_call_validation_diagnostics": list(
-                        target_validation.diagnostics
+            return _with_event_launch_suppression_telemetry(
+                WorkflowActionResult(
+                    status="failed",
+                    error=(
+                        f"workflow_mcp_target_contract_validation_failed:"
+                        f"{resolved_tool_name}:{error_code or 'invalid_target'}"
                     ),
-                },
+                    outputs={
+                        "mcp_tool": resolved_tool_name,
+                        "mcp_requested_tool": requested_tool_name,
+                        "mcp_resolved_tool": resolved_tool_name,
+                        "target_contract_validation_failed": True,
+                        "target_contract_validation_error_code": error_code,
+                        "tool_call_validation_diagnostics": list(
+                            target_validation.diagnostics
+                        ),
+                    },
+                ),
+                requested=event_launch_suppression_requested,
             )
 
         from ..security.access_control import override_current_actor
@@ -484,7 +531,17 @@ def _handle_workflow_mcp_invoke_tool(
             getattr(request.environment, "user_concept_id", None),
             getattr(request.environment, "org_concept_id", None),
         ):
-            result = gateway.invoke(resolved_tool_name, payload)
+            if event_launch_suppression_requested:
+                from ..services.workflow_event_integration_service import (
+                    suppress_event_workflow_launches,
+                )
+
+                with suppress_event_workflow_launches(
+                    "workflow_mcp.invoke_tool_owned_mutation"
+                ):
+                    result = gateway.invoke(resolved_tool_name, payload)
+            else:
+                result = gateway.invoke(resolved_tool_name, payload)
         action_result = workflow_action_result_from_mcp_payload(
             tool_name=resolved_tool_name,
             payload=result.payload,
@@ -498,7 +555,10 @@ def _handle_workflow_mcp_invoke_tool(
         action_result.outputs["mcp_requested_tool"] = requested_tool_name
         action_result.outputs["mcp_resolved_tool"] = resolved_tool_name
         action_result.outputs["workflow_actor_scope_enforced"] = True
-        return action_result
+        return _with_event_launch_suppression_telemetry(
+            action_result,
+            requested=event_launch_suppression_requested,
+        )
     except SchemaValidationError as exc:
         # Schema validation is a hard interface boundary, but workflows still
         # need a typed, inspectable failure in order to choose a represented
@@ -533,16 +593,22 @@ def _handle_workflow_mcp_invoke_tool(
         action_result.outputs["mcp_requested_tool"] = requested_tool_name
         action_result.outputs["mcp_resolved_tool"] = resolved_tool_name
         action_result.outputs["workflow_actor_scope_enforced"] = True
-        return action_result
+        return _with_event_launch_suppression_telemetry(
+            action_result,
+            requested=event_launch_suppression_requested,
+        )
     except Exception as exc:
         logger.warning(
             "[workflow_mcp] invoke failed for %s: %s",
             requested_tool_name,
             exc,
         )
-        return WorkflowActionResult(
-            status="failed",
-            error=f"workflow_mcp_invoke_failed:{requested_tool_name}:{exc}",
+        return _with_event_launch_suppression_telemetry(
+            WorkflowActionResult(
+                status="failed",
+                error=f"workflow_mcp_invoke_failed:{requested_tool_name}:{exc}",
+            ),
+            requested=event_launch_suppression_requested,
         )
 
 
@@ -563,6 +629,7 @@ def register_workflow_mcp_tool_actions(registry: ActionRegistry) -> None:
                 "properties": {
                     "tool_name": {"type": "string"},
                     "tool_arguments": {"type": "object"},
+                    SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT: {"type": "boolean"},
                 },
             },
             output_schema={
@@ -574,6 +641,9 @@ def register_workflow_mcp_tool_actions(registry: ActionRegistry) -> None:
                     "mcp_duration_ms": {"type": "number"},
                     "mcp_result": {"type": "object"},
                     "result": {"type": "object"},
+                    EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REQUESTED_OUTPUT: {
+                        "type": "boolean"
+                    },
                 },
             },
             side_effects="read_or_guarded_write",
@@ -582,6 +652,8 @@ def register_workflow_mcp_tool_actions(registry: ActionRegistry) -> None:
 
 
 __all__ = [
+    "EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REQUESTED_OUTPUT",
+    "SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT",
     "WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID",
     "extract_static_workflow_mcp_tool_name",
     "register_workflow_mcp_tool_actions",
diff --git a/tests/backend/test_kr_materialisation_guard_service.py b/tests/backend/test_kr_materialisation_guard_service.py
index 73466a7a7..e83945eaa 100644
--- a/tests/backend/test_kr_materialisation_guard_service.py
+++ b/tests/backend/test_kr_materialisation_guard_service.py
@@ -3,6 +3,7 @@
 from copy import deepcopy
 
 from src.backend.services.kr_materialisation_guard_service import (
+    GUARDED_CREATE_DUPLICATE_RESOLUTION_MODE,
     validate_kr_materialisation_guard,
 )
 
@@ -117,6 +118,7 @@ def test_optional_guard_preserves_unguarded_generic_kr_behaviour() -> None:
 
     assert result["guard_applied"] is False
     assert result["guard_passed"] is True
+    assert result["duplicate_resolution_mode"] is None
 
 
 def test_guard_accepts_exact_create_and_changed_record_reuse_plan() -> None:
@@ -143,6 +145,11 @@ def test_guard_accepts_exact_create_and_changed_record_reuse_plan() -> None:
 
     assert plan["guard_passed"] is True
     assert resolved["guard_passed"] is True
+    assert plan["duplicate_resolution_mode"] == GUARDED_CREATE_DUPLICATE_RESOLUTION_MODE
+    assert (
+        resolved["duplicate_resolution_mode"]
+        == GUARDED_CREATE_DUPLICATE_RESOLUTION_MODE
+    )
 
 
 def test_guard_enforces_canonical_existence_bound_decisions() -> None:
@@ -173,9 +180,9 @@ def test_guard_enforces_canonical_existence_bound_decisions() -> None:
         relationship_specs=_relationship_specs(),
     )
     assert rejected_reuse["guard_passed"] is False
+    assert rejected_reuse["duplicate_resolution_mode"] is None
     assert (
-        rejected_reuse["error_code"]
-        == "kr_materialisation_guard_concept_spec_rejected"
+        rejected_reuse["error_code"] == "kr_materialisation_guard_concept_spec_rejected"
     )
 
     invalid_create = _concept_specs()
@@ -238,10 +245,7 @@ def test_guard_normalises_predicate_id_alias_before_downstream_execution() -> No
     assert plan["guard_passed"] is True
     assert resolved["guard_passed"] is True
     assert relationship_specs[0]["predicate"] == "#V#has_doctoral_programme"
-    assert (
-        resolved_relationship_specs[0]["predicate"]
-        == "#V#has_doctoral_programme"
-    )
+    assert resolved_relationship_specs[0]["predicate"] == "#V#has_doctoral_programme"
 
 
 def test_guard_rejects_conflicting_predicate_aliases() -> None:
@@ -272,10 +276,7 @@ def test_guard_rejects_conflicting_predicate_aliases() -> None:
     )
 
     assert plan["guard_passed"] is False
-    assert (
-        plan["error_code"]
-        == "kr_materialisation_guard_relationship_spec_invalid"
-    )
+    assert plan["error_code"] == "kr_materialisation_guard_relationship_spec_invalid"
     assert resolved["guard_passed"] is False
     assert (
         resolved["error_code"]
@@ -325,8 +326,7 @@ def test_guard_rejects_injected_unrelated_concept_before_any_write() -> None:
     )
     assert reuse_result["guard_passed"] is False
     assert (
-        reuse_result["error_code"]
-        == "kr_materialisation_guard_reuse_payload_rejected"
+        reuse_result["error_code"] == "kr_materialisation_guard_reuse_payload_rejected"
     )
 
 
@@ -358,6 +358,5 @@ def test_guard_rejects_injected_edge_and_ungrounded_resolved_endpoint() -> None:
     assert plan["error_code"] == "kr_materialisation_guard_relationship_rejected"
     assert resolved["guard_passed"] is False
     assert (
-        resolved["error_code"]
-        == "kr_materialisation_guard_resolved_endpoint_rejected"
+        resolved["error_code"] == "kr_materialisation_guard_resolved_endpoint_rejected"
     )
diff --git a/tests/backend/test_kr_materialisation_workflow_vontology_service.py b/tests/backend/test_kr_materialisation_workflow_vontology_service.py
index e221a0a2a..8f7200ddc 100644
--- a/tests/backend/test_kr_materialisation_workflow_vontology_service.py
+++ b/tests/backend/test_kr_materialisation_workflow_vontology_service.py
@@ -130,8 +130,7 @@ def test_kr_seed_bundle_uses_prompt_concepts_not_inline_prompt_text() -> None:
         for step in llm_steps
     }
     llm_policy_by_state = {
-        step.get("state_id"): step.get("llm_policy") or {}
-        for step in llm_steps
+        step.get("state_id"): step.get("llm_policy") or {} for step in llm_steps
     }
     assert prompt_ids_by_state["extract_kr_materialisation_plan"] == (
         mod.PROMPT_KR_DESIGN_MATERIALISATION_PLAN_ID,
@@ -151,8 +150,7 @@ def test_kr_prompt_seed_assets_hold_operational_prompt_content() -> None:
 
     assert "materialise bounded knowledge representation designs" in plan_prompt
     assert (
-        "Return JSON only with keys: decision ('materialise' or 'block')"
-        in plan_prompt
+        "Return JSON only with keys: decision ('materialise' or 'block')" in plan_prompt
     )
     assert "Resolve KR relationship endpoint references" in relationship_prompt
     assert (
@@ -163,14 +161,13 @@ def test_kr_prompt_seed_assets_hold_operational_prompt_content() -> None:
 
 def test_kr_seed_applies_optional_guard_before_each_write_phase() -> None:
     bundle = json.loads(_SEED_BUNDLE_PATH.read_text(encoding="utf-8"))
+    assert bundle["seed_version"] == "4"
     workflow = next(
         row
         for row in bundle["workflows"]
         if row["workflow_id"] == mod.KR_DESIGN_MATERIALISATION_WORKFLOW_ID
     )
-    states = {
-        row["state_id"]: row for row in workflow["publication_spec"]["steps"]
-    }
+    states = {row["state_id"]: row for row in workflow["publication_spec"]["steps"]}
 
     assert workflow["launch_input_contract"]["excluded_ambient_input_keys"] == [
         "invocations"
@@ -180,18 +177,69 @@ def test_kr_seed_applies_optional_guard_before_each_write_phase() -> None:
         and row["required"] is False
         for row in workflow["launch_input_contract"]["input_mappings"]
     )
-    assert states["extract_kr_materialisation_plan"]["conditional_transitions"][
-        0
-    ]["to_state"] == "validate_materialisation_plan"
+    assert (
+        states["extract_kr_materialisation_plan"]["conditional_transitions"][0][
+            "to_state"
+        ]
+        == "validate_materialisation_plan"
+    )
     assert states["validate_materialisation_plan"]["action_id"] == (
         "workflow_control.kr_materialisation_guard"
     )
     assert states["validate_materialisation_plan"]["static_input_bindings"] == [
         ["phase", "plan"]
     ]
-    assert states["resolve_relationship_specs"]["conditional_transitions"][0][
-        "to_state"
-    ] == "validate_resolved_relationships"
+    guard_outputs = {
+        row["tool_output_field"]: row["context_key"]
+        for row in states["validate_materialisation_plan"]["tool_output_mapping_specs"]
+    }
+    assert guard_outputs["duplicate_resolution_mode"] == (
+        "kr_concept_duplicate_resolution_mode"
+    )
+    assert (
+        "kr_concept_duplicate_resolution_mode"
+        in states["validate_materialisation_plan"]["writes_context_keys"]
+    )
+    concept_item = next(
+        row
+        for row in bundle["workflows"]
+        if row["workflow_id"] == mod.KR_DESIGN_CONCEPT_MATERIALISATION_ITEM_WORKFLOW_ID
+    )
+    concept_item_states = {
+        row["state_id"]: row for row in concept_item["publication_spec"]["steps"]
+    }
+    duplicate_resolution_mapping = next(
+        row
+        for row in concept_item_states["create_concept"]["context_input_mapping_specs"]
+        if row["tool_param"] == "duplicate_resolution_mode"
+    )
+    assert duplicate_resolution_mapping == {
+        "concept_id": (
+            "#V#workflow_mapping_kr_design_concept_materialisation_item_workflow_"
+            "create_concept_kr_concept_duplicate_resolution_mode_to_"
+            "duplicate_resolution_mode_parameter"
+        ),
+        "context_key": "kr_concept_duplicate_resolution_mode",
+        "required": False,
+        "tool_param": "duplicate_resolution_mode",
+    }
+    assert not any(
+        (
+            isinstance(binding, list)
+            and len(binding) == 2
+            and binding[0] == "duplicate_resolution_mode"
+        )
+        for binding in concept_item_states["create_concept"]["static_input_bindings"]
+    )
+    concept_iteration_bindings = dict(
+        states["materialise_concepts"]["static_input_bindings"]
+    )
+    assert concept_iteration_bindings["max_concurrency"] == 1
+    assert concept_iteration_bindings["stop_on_error"] is True
+    assert (
+        states["resolve_relationship_specs"]["conditional_transitions"][0]["to_state"]
+        == "validate_resolved_relationships"
+    )
     assert states["validate_resolved_relationships"]["action_id"] == (
         "workflow_control.kr_materialisation_guard"
     )
@@ -200,6 +248,42 @@ def test_kr_seed_applies_optional_guard_before_each_write_phase() -> None:
     ]
 
 
+def test_kr_seed_suppresses_event_fan_out_only_for_owned_mutations() -> None:
+    bundle = json.loads(_SEED_BUNDLE_PATH.read_text(encoding="utf-8"))
+    suppression_states: set[tuple[str, str]] = set()
+
+    for workflow in bundle["workflows"]:
+        for step in workflow["publication_spec"]["steps"]:
+            bindings = dict(step.get("static_input_bindings") or [])
+            if "suppress_event_workflow_launches" not in bindings:
+                continue
+            assert bindings["suppress_event_workflow_launches"] is True
+            suppression_states.add((workflow["workflow_id"], step["state_id"]))
+
+    assert suppression_states == {
+        (
+            mod.KR_DESIGN_CONCEPT_MATERIALISATION_ITEM_WORKFLOW_ID,
+            "create_concept",
+        ),
+        (
+            mod.KR_DESIGN_CONCEPT_MATERIALISATION_ITEM_WORKFLOW_ID,
+            "assert_type_parent_relationship",
+        ),
+        (
+            mod.KR_DESIGN_CONCEPT_MATERIALISATION_ITEM_WORKFLOW_ID,
+            "assert_instance_parent_relationship",
+        ),
+        (
+            mod.KR_DESIGN_CONCEPT_MATERIALISATION_ITEM_WORKFLOW_ID,
+            "attach_description",
+        ),
+        (
+            mod.KR_DESIGN_RELATIONSHIP_ASSERTION_ITEM_WORKFLOW_ID,
+            "assert_relationship",
+        ),
+    }
+
+
 def test_kr_seed_bundle_validates_as_workflow_contracts() -> None:
     report = mod.validate_kr_materialisation_seed_bundle()
 
diff --git a/tests/backend/test_mcp_create_concepts_duplicate_resolution_mode.py b/tests/backend/test_mcp_create_concepts_duplicate_resolution_mode.py
new file mode 100644
index 000000000..a448f958c
--- /dev/null
+++ b/tests/backend/test_mcp_create_concepts_duplicate_resolution_mode.py
@@ -0,0 +1,322 @@
+"""Focused create_concepts duplicate-mode and cancellation regressions."""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+
+from src.backend.integrations.internal_mcp.catalogue import (
+    _concepts_create_input_schema,
+    _create_concepts,
+)
+from src.backend.integrations.internal_mcp.schemas import validate_payload
+from src.backend.integrations.internal_mcp.transport import (
+    InternalMCPHandlerCancelled,
+)
+from src.backend.services.create_concepts_parent_resolution_service import (
+    ParentResolutionResult,
+)
+from src.backend.utils.concept_id_utils import canonicalise_vontology_concept_id
+
+
+_PARENT_ID = "#V#abstract_object"
+
+
+def _patch_common_preflights(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        "src.backend.services.create_concepts_parent_resolution_service."
+        "resolve_parent_for_create_concepts",
+        lambda parent_id: ParentResolutionResult(
+            requested_parent_id=parent_id,
+            canonical_parent_id=_PARENT_ID,
+            resolved_parent_id=_PARENT_ID,
+            fallback_used=False,
+            fallback_candidates_checked=(),
+            fallback_selected_parent_id=None,
+        ),
+    )
+    monkeypatch.setattr(
+        "src.backend.services.workflow_event_integration_service."
+        "resolve_event_actor_context",
+        lambda **_kwargs: (None, None),
+    )
+
+
+def _created_payload(**kwargs: Any) -> dict[str, Any]:
+    name = str(kwargs["new_concept_name"])
+    concept_id = canonicalise_vontology_concept_id(name)
+    assert concept_id
+    return {
+        "success": True,
+        "message": "created",
+        "concept": {"concept_id": concept_id},
+        "canonical_concept_id": concept_id,
+        "input_name": name,
+    }
+
+
+def _not_found_resolution(**_kwargs: Any) -> dict[str, Any]:
+    return {
+        "success": True,
+        "status": "not_found",
+        "resolved_concept_id": None,
+        "candidates": [],
+        "audit": [],
+    }
+
+
+def test_canonical_id_only_skips_semantic_resolution_after_exact_miss(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_common_preflights(monkeypatch)
+    semantic_calls: list[str] = []
+    exact_queries: list[dict[str, Any]] = []
+
+    def _find_one(
+        query: dict[str, Any],
+        _projection: dict[str, Any] | None = None,
+    ) -> None:
+        exact_queries.append(query)
+        return None
+
+    def _resolve_concept_by_name(**kwargs: Any) -> dict[str, Any]:
+        semantic_calls.append(str(kwargs.get("name")))
+        return _not_found_resolution()
+
+    monkeypatch.setattr(
+        "src.backend.db.repositories.concepts_repository.ConceptsRepository.find_one",
+        _find_one,
+    )
+    monkeypatch.setattr(
+        "src.backend.services.concept_resolution_service.resolve_concept_by_name",
+        _resolve_concept_by_name,
+    )
+    monkeypatch.setattr(
+        "src.backend.vontology.utils_vontology.create_vontology_concept",
+        _created_payload,
+    )
+
+    result = _create_concepts(
+        parent_id=_PARENT_ID,
+        concepts=[{"name": "stable represented identity", "kind": "type"}],
+        duplicate_resolution_mode="canonical_id_only",
+    )
+
+    assert result["successful"] == 1
+    assert semantic_calls == []
+    assert exact_queries == [{"concept_id": "#V#stable_represented_identity"}]
+
+
+def test_default_mode_retains_semantic_duplicate_resolution(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_common_preflights(monkeypatch)
+    semantic_calls: list[str] = []
+
+    monkeypatch.setattr(
+        "src.backend.db.repositories.concepts_repository.ConceptsRepository.find_one",
+        lambda _query, _projection=None: None,
+    )
+
+    def _resolve_concept_by_name(**kwargs: Any) -> dict[str, Any]:
+        semantic_calls.append(str(kwargs.get("name")))
+        return _not_found_resolution()
+
+    monkeypatch.setattr(
+        "src.backend.services.concept_resolution_service.resolve_concept_by_name",
+        _resolve_concept_by_name,
+    )
+    monkeypatch.setattr(
+        "src.backend.vontology.utils_vontology.create_vontology_concept",
+        _created_payload,
+    )
+
+    result = _create_concepts(
+        parent_id=_PARENT_ID,
+        concepts=[{"name": "human supplied label", "kind": "type"}],
+    )
+
+    assert result["successful"] == 1
+    assert semantic_calls == ["human supplied label"]
+
+
+def test_canonical_id_only_still_reuses_exact_existing_identity(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_common_preflights(monkeypatch)
+    semantic_calls: list[str] = []
+    create_calls: list[dict[str, Any]] = []
+
+    existing = {
+        "concept_id": "#V#stable_existing_identity",
+        "relationships": {
+            "is_a_type_of": [_PARENT_ID],
+            "is_an_instance_of": [],
+        },
+    }
+    monkeypatch.setattr(
+        "src.backend.db.repositories.concepts_repository.ConceptsRepository.find_one",
+        lambda query, _projection=None: (
+            existing if query.get("concept_id") == existing["concept_id"] else None
+        ),
+    )
+
+    def _resolve_concept_by_name(**kwargs: Any) -> dict[str, Any]:
+        semantic_calls.append(str(kwargs.get("name")))
+        return _not_found_resolution()
+
+    monkeypatch.setattr(
+        "src.backend.services.concept_resolution_service.resolve_concept_by_name",
+        _resolve_concept_by_name,
+    )
+    monkeypatch.setattr(
+        "src.backend.vontology.utils_vontology.create_vontology_concept",
+        lambda **kwargs: create_calls.append(kwargs),
+    )
+
+    result = _create_concepts(
+        parent_id=_PARENT_ID,
+        concepts=[{"name": "stable existing identity", "kind": "type"}],
+        duplicate_resolution_mode="canonical_id_only",
+    )
+
+    item = result["results"][0]
+    assert result["successful"] == 0
+    assert result["already_existed"] == 1
+    assert item["existing_concept_id"] == existing["concept_id"]
+    assert item["duplicate_match_source"] == "canonical_concept_id"
+    assert semantic_calls == []
+    assert create_calls == []
+
+
+def test_duplicate_resolution_mode_schema_and_handler_reject_unknown_value() -> None:
+    schema = _concepts_create_input_schema()
+    assert schema.enum_values["duplicate_resolution_mode"] == [
+        "canonical_id_only",
+        None,
+    ]
+    valid, errors = validate_payload(
+        schema,
+        {
+            "parent_id": _PARENT_ID,
+            "concepts": [],
+            "duplicate_resolution_mode": "canonical_id_only",
+        },
+    )
+    assert valid is True
+    assert errors == []
+
+    valid_null, errors = validate_payload(
+        schema,
+        {
+            "parent_id": _PARENT_ID,
+            "concepts": [],
+            "duplicate_resolution_mode": None,
+        },
+    )
+    assert valid_null is True
+    assert errors == []
+
+    invalid, errors = validate_payload(
+        schema,
+        {
+            "parent_id": _PARENT_ID,
+            "concepts": [],
+            "duplicate_resolution_mode": "semantic",
+        },
+    )
+    assert invalid is False
+    assert any("canonical_id_only" in error for error in errors)
+
+    handler_result = _create_concepts(
+        parent_id=_PARENT_ID,
+        concepts=[],
+        duplicate_resolution_mode="semantic",
+    )
+    assert handler_result["error_code"] == "invalid_parameter"
+    assert handler_result["error_details"]["supported_duplicate_resolution_modes"] == [
+        "canonical_id_only"
+    ]
+
+
+def test_cancellation_after_duplicate_preflight_prevents_create(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_common_preflights(monkeypatch)
+    duplicate_preflight_completed = False
+    create_calls: list[dict[str, Any]] = []
+
+    def _duplicate_guard(**_kwargs: Any) -> None:
+        nonlocal duplicate_preflight_completed
+        duplicate_preflight_completed = True
+        return None
+
+    def _raise_when_preflight_completed() -> None:
+        if duplicate_preflight_completed:
+            raise InternalMCPHandlerCancelled("cancelled before create")
+
+    monkeypatch.setattr(
+        "src.backend.services.create_concepts_duplicate_guard_service."
+        "find_existing_concept_for_create_concepts",
+        _duplicate_guard,
+    )
+    monkeypatch.setattr(
+        "src.backend.integrations.internal_mcp.transport."
+        "raise_if_internal_mcp_cancelled",
+        _raise_when_preflight_completed,
+    )
+    monkeypatch.setattr(
+        "src.backend.vontology.utils_vontology.create_vontology_concept",
+        lambda **kwargs: create_calls.append(kwargs),
+    )
+
+    with pytest.raises(InternalMCPHandlerCancelled):
+        _create_concepts(
+            parent_id=_PARENT_ID,
+            concepts=[{"name": "cancelled identity", "kind": "type"}],
+        )
+
+    assert create_calls == []
+
+
+def test_cancellation_between_batch_items_prevents_later_creates(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    _patch_common_preflights(monkeypatch)
+    created_names: list[str] = []
+
+    monkeypatch.setattr(
+        "src.backend.services.create_concepts_duplicate_guard_service."
+        "find_existing_concept_for_create_concepts",
+        lambda **_kwargs: None,
+    )
+
+    def _cancel_after_first_complete() -> None:
+        if created_names:
+            raise InternalMCPHandlerCancelled("cancelled between items")
+
+    def _create(**kwargs: Any) -> dict[str, Any]:
+        created_names.append(str(kwargs["new_concept_name"]))
+        return _created_payload(**kwargs)
+
+    monkeypatch.setattr(
+        "src.backend.integrations.internal_mcp.transport."
+        "raise_if_internal_mcp_cancelled",
+        _cancel_after_first_complete,
+    )
+    monkeypatch.setattr(
+        "src.backend.vontology.utils_vontology.create_vontology_concept",
+        _create,
+    )
+
+    with pytest.raises(InternalMCPHandlerCancelled):
+        _create_concepts(
+            parent_id=_PARENT_ID,
+            concepts=[
+                {"name": "first complete identity", "kind": "type"},
+                {"name": "later blocked identity", "kind": "type"},
+            ],
+        )
+
+    assert created_names == ["first complete identity"]
diff --git a/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py b/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py
index 28b74d384..3964fab41 100644
--- a/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py
+++ b/tests/backend/test_spreadsheet_programme_workflow_vontology_service.py
@@ -162,8 +162,9 @@ def test_bootstrap_materialises_model_led_spreadsheet_workflow_family(
         "spreadsheet_write_authority_contract.v1"
     )
     assert "#V#person" in write_authority["allowed_concept_parent_ids"]
-    assert "#V#has_doctoral_supervisor" in (
-        write_authority["allowed_relationship_predicate_ids"]
+    assert (
+        "#V#has_doctoral_supervisor"
+        in (write_authority["allowed_relationship_predicate_ids"])
     )
     for_each = _state(main, "materialise_records").actions[0]
     assert for_each.action_id == "workflow_control.for_each"
@@ -181,9 +182,7 @@ def test_bootstrap_materialises_model_led_spreadsheet_workflow_family(
     assert child_materialise.inputs["workflow_id"] == (
         "#V#kr_design_materialisation_workflow"
     )
-    assert child_materialise.inputs[
-        "conversation_turn_llm_timeout_override_sec"
-    ] == 180
+    assert child_materialise.inputs["conversation_turn_llm_timeout_override_sec"] == 180
     assert child_materialise.inputs["materialisation_guard"]["$context_key"] == (
         "spreadsheet_record_materialisation_guard"
     )
@@ -218,24 +217,26 @@ def test_bootstrap_materialises_model_led_spreadsheet_workflow_family(
             "previous_spreadsheet_record_marker_id",
         )
     }
-    unchanged_assignments = _state(item, "emit_unchanged_skip").actions[0].inputs[
-        "assignments"
-    ]
+    unchanged_assignments = (
+        _state(item, "emit_unchanged_skip").actions[0].inputs["assignments"]
+    )
     unchanged_marker_assignment = next(
         row for row in unchanged_assignments if row.get("key") == "record_marker_id"
     )
     assert unchanged_marker_assignment["value_from_context"] == (
         "previous_spreadsheet_record_marker_id"
     )
-    marker_fingerprint = _state(item, "read_record_marker").actions[0].inputs[
-        "processing_authority_fingerprint"
-    ]
+    marker_fingerprint = (
+        _state(item, "read_record_marker")
+        .actions[0]
+        .inputs["processing_authority_fingerprint"]
+    )
     assert marker_fingerprint.startswith("sha256:")
     assert len(marker_fingerprint) == 71
 
-    required_effects = (
-        main.metadata.get("required_effects_contract") or {}
-    ).get("required_effects") or []
+    required_effects = (main.metadata.get("required_effects_contract") or {}).get(
+        "required_effects"
+    ) or []
     reconciliation = next(
         effect
         for effect in required_effects
@@ -246,8 +247,9 @@ def test_bootstrap_materialises_model_led_spreadsheet_workflow_family(
         "build_spreadsheet_batch_completion_evidence",
         "record_source_processing_marker",
     ]
-    assert "build_spreadsheet_record_materialisation_request" not in (
-        reconciliation["required_tools"]
+    assert (
+        "build_spreadsheet_record_materialisation_request"
+        not in (reconciliation["required_tools"])
     )
 
 
@@ -267,9 +269,7 @@ def test_bootstrap_fails_closed_when_support_parent_is_missing(
     assert report["support_parent_concepts"]["missing_concept_ids"] == [
         "#V#binary_predicate"
     ]
-    assert (
-        db.concepts.find_one({"concept_id": "#V#has_doctoral_supervisor"}) is None
-    )
+    assert db.concepts.find_one({"concept_id": "#V#has_doctoral_supervisor"}) is None
 
 
 def test_spreadsheet_planner_prompt_pins_untrusted_data_and_versioning(
@@ -343,6 +343,7 @@ def test_processing_authority_fingerprint_tracks_exact_kr_dependency(
     from scripts import generate_spreadsheet_programme_workflow_seed as generator
 
     baseline = generator.build_bundle()
+    assert baseline["seed_version"] == "17"
     baseline_dependency = baseline["processing_authority_dependencies"][
         "kr_materialisation_workflow_seed_bundle"
     ]
@@ -351,6 +352,7 @@ def test_processing_authority_fingerprint_tracks_exact_kr_dependency(
     )
     assert baseline_dependency["family_id"] == source_payload["family_id"]
     assert baseline_dependency["seed_version"] == source_payload["seed_version"]
+    assert baseline_dependency["seed_version"] == "4"
     assert baseline_dependency["canonical_payload_sha256"].startswith("sha256:")
 
     changed_payload = dict(source_payload)
diff --git a/tests/backend/test_workflow_event_integration_service.py b/tests/backend/test_workflow_event_integration_service.py
index a3d8c6a65..e2dc0063d 100644
--- a/tests/backend/test_workflow_event_integration_service.py
+++ b/tests/backend/test_workflow_event_integration_service.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timedelta, timezone
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
@@ -9,6 +10,9 @@
 from src.backend.services import (
     workflow_event_integration_service as workflow_event_service,
 )
+from src.backend.services.feature_flags import (
+    get_workflow_discovery_cache_invalidation_enabled,
+)
 from src.backend.workflows.durable.models import EventWorkflowBinding
 from src.backend.workflows.durable.workflow_instance_submission_service import (
     WorkflowInstanceSubmissionResult,
@@ -21,6 +25,7 @@
     EVENT_TYPE_TYPE_CREATED,
     EVENT_TYPE_VONTOLOGY_MUTATED,
     EVENT_TYPE_WORKFLOW_INSTANCE_TERMINAL,
+    current_event_workflow_launch_suppression_reason,
     launch_event_workflow,
     maybe_launch_episode_evaluation_for_turn_completion_gate,
     maybe_launch_episode_evaluation_for_workflow_terminal,
@@ -29,6 +34,7 @@
     maybe_launch_type_created_workflow,
     maybe_launch_vontology_mutation_workflow,
     maybe_launch_task_status_workflow,
+    suppress_event_workflow_launches,
 )
 
 
@@ -52,6 +58,72 @@ def _submission_result(
     )
 
 
+def test_event_workflow_launch_suppression_is_nested_and_thread_local() -> None:
+    assert current_event_workflow_launch_suppression_reason() is None
+
+    with suppress_event_workflow_launches("outer"):
+        assert current_event_workflow_launch_suppression_reason() == "outer"
+        with suppress_event_workflow_launches("inner"):
+            assert current_event_workflow_launch_suppression_reason() == "inner"
+        assert current_event_workflow_launch_suppression_reason() == "outer"
+
+        with ThreadPoolExecutor(max_workers=1) as executor:
+            isolated_reason = executor.submit(
+                current_event_workflow_launch_suppression_reason
+            ).result()
+        assert isolated_reason is None
+
+    assert current_event_workflow_launch_suppression_reason() is None
+
+
+def test_suppressed_event_launch_returns_before_binding_reads(
+    monkeypatch,
+) -> None:
+    monkeypatch.setenv("VON_EVENT_WORKFLOW_INTEGRATION_ENABLE", "1")
+    monkeypatch.setenv("VON_DURABLE_WORKFLOWS_ENABLE", "1")
+    binding_lookup = MagicMock(side_effect=AssertionError("binding read attempted"))
+    monkeypatch.setattr(
+        workflow_event_service,
+        "_resolve_bindings_for_event",
+        binding_lookup,
+    )
+
+    with suppress_event_workflow_launches("owned_materialisation_mutation"):
+        result = launch_event_workflow(
+            event_type=EVENT_TYPE_TASK_CREATED,
+            event_id="task-1",
+            user_id="#V#user_alice",
+            org_id="#V#org_nao",
+        )
+
+    assert result == {
+        "success": True,
+        "triggered": False,
+        "outcome": "suppressed",
+        "event_type": EVENT_TYPE_TASK_CREATED,
+        "event_id": "task-1",
+        "reason": "event_workflow_launch_suppressed",
+        "suppression_reason": "owned_materialisation_mutation",
+        "event_workflow_launch_suppressed": True,
+    }
+    binding_lookup.assert_not_called()
+
+
+def test_event_launch_suppression_disables_discovery_cache_invalidation_locally(
+    monkeypatch,
+) -> None:
+    monkeypatch.delenv(
+        "VON_WORKFLOW_DISCOVERY_CACHE_INVALIDATION_ENABLE",
+        raising=False,
+    )
+    assert get_workflow_discovery_cache_invalidation_enabled(default=True) is True
+
+    with suppress_event_workflow_launches("owned_materialisation_mutation"):
+        assert get_workflow_discovery_cache_invalidation_enabled(default=True) is False
+
+    assert get_workflow_discovery_cache_invalidation_enabled(default=True) is True
+
+
 def test_launch_event_workflow_integration_flag_disables_trigger(monkeypatch) -> None:
     monkeypatch.setenv("VON_EVENT_WORKFLOW_INTEGRATION_ENABLE", "0")
     monkeypatch.setenv("VON_DURABLE_WORKFLOWS_ENABLE", "1")
diff --git a/tests/backend/test_workflow_mcp_tool_actions.py b/tests/backend/test_workflow_mcp_tool_actions.py
index 7edcb3dba..193073065 100644
--- a/tests/backend/test_workflow_mcp_tool_actions.py
+++ b/tests/backend/test_workflow_mcp_tool_actions.py
@@ -11,6 +11,9 @@
 )
 from src.backend.integrations.internal_mcp.schemas import Schema
 from src.backend.integrations.internal_mcp.transport import InternalMCPTransport
+from src.backend.services.workflow_event_integration_service import (
+    current_event_workflow_launch_suppression_reason,
+)
 from src.backend.workflows.action_registry import (
     ActionRegistry,
     ActionSpec,
@@ -21,6 +24,8 @@
     register_subworkflow_actions,
 )
 from src.backend.workflows.workflow_mcp_tool_actions import (
+    EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REQUESTED_OUTPUT,
+    SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT,
     WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID,
     register_workflow_mcp_tool_actions,
 )
@@ -152,6 +157,84 @@ def test_workflow_mcp_action_invokes_read_tool_and_maps_structured_output():
     ]
 
 
+def test_workflow_mcp_event_launch_control_propagates_without_payload_forwarding(
+    monkeypatch,
+) -> None:
+    monkeypatch.setattr(
+        workflow_mcp_mod,
+        "validate_tool_target_contract",
+        lambda **_kwargs: SimpleNamespace(ok=True),
+    )
+    observations: list[tuple[str, str | None]] = []
+
+    def _handler(*, message: str) -> dict[str, object]:
+        observations.append(
+            (message, current_event_workflow_launch_suppression_reason())
+        )
+        return {"success": True, "message": message}
+
+    catalogue = MethodCatalogue()
+    catalogue.register(
+        MethodDefinition(
+            name="synthetic_event_suppression_probe",
+            handler=_handler,
+            input_schema=Schema(
+                required={"message": str},
+                optional={},
+                allow_unknown=False,
+            ),
+            output_schema=None,
+            category="read",
+        )
+    )
+    gateway = InternalMCPGateway(
+        catalogue=catalogue,
+        transport=InternalMCPTransport(),
+        enabled=True,
+    )
+    registry = ActionRegistry()
+    register_workflow_mcp_tool_actions(registry)
+
+    default_result = registry.execute(
+        WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID,
+        inputs={
+            "tool_name": "synthetic_event_suppression_probe",
+            "tool_arguments": {"message": "default"},
+        },
+        context={},
+        env=WorkflowEnvironment(llm_client=None, gateway=gateway),
+    )
+    suppressed_result = registry.execute(
+        WORKFLOW_MCP_INVOKE_TOOL_ACTION_ID,
+        inputs={
+            "tool_name": "synthetic_event_suppression_probe",
+            "tool_arguments": {
+                "message": "suppressed",
+                SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT: True,
+            },
+            SUPPRESS_EVENT_WORKFLOW_LAUNCHES_INPUT: True,
+        },
+        context={},
+        env=WorkflowEnvironment(llm_client=None, gateway=gateway),
+    )
+
+    assert default_result.status == "success"
+    assert (
+        default_result.outputs[EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REQUESTED_OUTPUT]
+        is False
+    )
+    assert suppressed_result.status == "success"
+    assert (
+        suppressed_result.outputs[EVENT_WORKFLOW_LAUNCH_SUPPRESSION_REQUESTED_OUTPUT]
+        is True
+    )
+    assert observations == [
+        ("default", None),
+        ("suppressed", "workflow_mcp.invoke_tool_owned_mutation"),
+    ]
+    assert current_event_workflow_launch_suppression_reason() is None
+
+
 def test_workflow_mcp_action_preserves_typed_gateway_schema_failure() -> None:
     gateway = InternalMCPGateway(
         catalogue=build_default_catalogue(),