From eaa942801d61dd3b3939fbe9c75bf6d4e319c915 Mon Sep 17 00:00:00 2001
From: witbrock <witbrock@gmail.com>
Date: Mon, 27 Jul 2026 14:53:15 +0100
Subject: [PATCH] Harden adaptive provider and effect seams

---
 .../integrations/internal_mcp/catalogue.py    | 116 ++++-
 .../integrations/internal_mcp/gateway.py      |  12 +
 .../integrations/internal_mcp/transport.py    |  53 +++
 src/backend/languagemodels/llm_interface.py   |   8 +-
 .../structured_tool_calling/client.py         |   2 +
 .../providers/openai_client.py                | 121 +++++-
 .../structured_tool_calling/transport.py      | 215 +++++++---
 src/backend/mcp_server/mcp_stdio_server.py    |   7 +
 src/backend/services/adaptive_turn_service.py |  48 +++
 src/backend/services/concept_service.py       |  57 ++-
 ...eate_concepts_parent_resolution_service.py |  28 +-
 .../services/model_parameter_service.py       |  15 +
 .../services/model_registry_service.py        | 116 +++--
 src/backend/services/text_value_service.py    |  28 +-
 .../services/turn_execution_record_service.py | 248 +++++++++--
 tests/backend/test_adaptive_turn_service.py   |  69 +++
 .../test_concept_service_read_enrichment.py   | 195 ++++++++-
 .../test_internal_mcp_transport_deadlines.py  |  35 ++
 .../test_llm_openai_temperature_guards.py     | 138 +++++-
 ...est_mcp_create_concepts_parent_recovery.py |  35 +-
 tests/backend/test_model_parameter_service.py |  27 ++
 tests/backend/test_model_registry_service.py  | 135 ++++++
 .../test_openai_structured_tool_transport.py  | 398 +++++++++++++++++-
 ...g_turn_execution_records_mcp_read_tools.py |  78 ++++
 ...cution_record_service_execution_summary.py | 147 ++++++-
 tests/backend/test_von_chat_run_timeout.py    |  23 +-
 26 files changed, 2171 insertions(+), 183 deletions(-)

diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py
index 93b7f2554..316118b10 100644
--- a/src/backend/integrations/internal_mcp/catalogue.py
+++ b/src/backend/integrations/internal_mcp/catalogue.py
@@ -1093,6 +1093,49 @@ def _create_concepts(**kwargs):
                 "Provide an array of concept objects, e.g., concepts=[{name: 'MyType'}]"
             ],
         )
+    requested_non_predicate_kinds = {
+        (
+            str(concept_data.get("kind") or "type").strip().lower()
+            if isinstance(concept_data, dict)
+            else "invalid"
+        )
+        for concept_data in concepts
+        if not (
+            isinstance(concept_data, dict)
+            and str(concept_data.get("kind") or "type").strip().lower()
+            == "predicate"
+        )
+    }
+    if (
+        parent_resolution.resolved_parent_kind in {"individual", "instance"}
+        and requested_non_predicate_kinds
+    ):
+        return make_error_response(
+            "parent_is_not_a_type",
+            (
+                f"Parent concept '{validated_parent_id}' is an individual, not "
+                "a semantic type. No concepts were created."
+            ),
+            details={
+                "original_parent_id": parent_id,
+                "resolved_parent_id": validated_parent_id,
+                "resolved_parent_kind": parent_resolution.resolved_parent_kind,
+                "requested_child_kinds": sorted(requested_non_predicate_kinds),
+            },
+            suggestions=[
+                "Choose an existing semantic type for parent_id",
+                (
+                    "Use scope_mode or organisation_concept_id for ownership and "
+                    "visibility; do not use an organisation or owner individual "
+                    "as parent_id"
+                ),
+                (
+                    "If the intended type does not exist, create that type first "
+                    "under a semantically close type parent"
+                ),
+            ],
+            related_concept_ids=[validated_parent_id],
+        )
 
     results = []
     with defer_relationship_extent_index_sync():
@@ -7172,7 +7215,8 @@ def _concepts_create_input_schema() -> Schema:
         },
         allow_unknown=True,
         description=(
-            "create_concepts input: parent_id (str, parent concept_id), concepts (list of {name, kind?, description?, notes?}). "
+            "create_concepts input: parent_id (str, semantic type concept_id; not an owner, organisation, user, or container individual), "
+            "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. "
@@ -10082,6 +10126,65 @@ def _summarise_turn_execution_tool_invocations(
     return summary
 
 
+def _summarise_late_effect_observations(
+    raw_observations: Any,
+    *,
+    max_items: int = 32,
+) -> tuple[list[dict[str, Any]], int]:
+    """Return an actor-safe bounded receipt projection for late effects."""
+
+    if not isinstance(raw_observations, list):
+        return [], 0
+
+    summary: list[dict[str, Any]] = []
+    for raw_entry in raw_observations[-max_items:]:
+        if not isinstance(raw_entry, Mapping):
+            continue
+        entry = {
+            key: raw_entry.get(key)
+            for key in (
+                "schema_version",
+                "observation_id",
+                "effect_id",
+                "execution_id",
+                "call_id",
+                "capability_name",
+                "method_name",
+                "outcome",
+                "effect_status",
+                "changed",
+                "observed_at_utc",
+                "output_schema_validation",
+                "output_schema_valid",
+                "output_schema_error",
+                "payload_truncated",
+                "payload_redacted",
+                "storage_transformed",
+                "error_type",
+                "error",
+            )
+            if key in raw_entry
+        }
+        payload = raw_entry.get("payload")
+        if isinstance(payload, Mapping):
+            entry["receipt"] = {
+                key: payload.get(key)
+                for key in (
+                    "success",
+                    "status",
+                    "effect_status",
+                    "changed",
+                    "error",
+                    "error_code",
+                    "mutation_outcome",
+                    "partial_failures",
+                )
+                if key in payload
+            }
+        summary.append(entry)
+    return summary, len(raw_observations)
+
+
 def _extract_turn_execution_tool_invocation_summary(
     item: Mapping[str, Any],
 ) -> list[dict[str, Any]]:
@@ -21670,6 +21773,12 @@ def _rag_get_item(**kwargs):
                 source="turn_execution_get.turn_execution_record",
             )
         )
+        (
+            late_effect_observations,
+            late_effect_observation_count,
+        ) = _summarise_late_effect_observations(
+            doc.get("late_effect_observations")
+        )
 
         payload = {
             "collection": collection,
@@ -21712,6 +21821,11 @@ def _rag_get_item(**kwargs):
             "tool_invocation_summary": _summarise_turn_execution_tool_invocations(
                 tool_invocations
             ),
+            "late_effect_observation_count": late_effect_observation_count,
+            "late_effect_observations": late_effect_observations,
+            "late_effect_observations_truncated": (
+                late_effect_observation_count > len(late_effect_observations)
+            ),
             "required_effects": required_effects,
             "postcondition_checks": postcondition_checks,
             "critic": critic_payload,
diff --git a/src/backend/integrations/internal_mcp/gateway.py b/src/backend/integrations/internal_mcp/gateway.py
index bc68052b5..a56ee78e2 100644
--- a/src/backend/integrations/internal_mcp/gateway.py
+++ b/src/backend/integrations/internal_mcp/gateway.py
@@ -343,6 +343,7 @@ def invoke(
         payload: Optional[MutableMapping[str, Any]] = None,
         *,
         deadline_monotonic: float | None = None,
+        require_configured_timeout: bool = False,
         late_completion_observer: LateCompletionObserver | None = None,
     ) -> TransportResult:
         if not self._enabled:
@@ -520,6 +521,7 @@ def _observe_validated_late_completion(
                         category=definition.category,
                         advisory_timeout_sec=advisory_timeout,
                         deadline_monotonic=deadline_monotonic,
+                        require_configured_timeout=require_configured_timeout,
                         log_tag=self._log_tag,
                         late_completion_observer=observed_late_completion,
                     )
@@ -643,6 +645,16 @@ def get_method_definition(self, method_name: str) -> MethodDefinition | None:
         except Exception:
             return None
 
+    def get_method_timeout_sec(self, method_name: str) -> float | None:
+        """Return the configured hard window for one registered method."""
+
+        definition = self.get_method_definition(method_name)
+        return (
+            definition.resolved_timeout(self._transport)
+            if definition is not None
+            else None
+        )
+
     def extend_catalogue(self, definitions: Iterable[MethodDefinition]) -> None:
         for definition in definitions:
             self._catalogue.register(definition)
diff --git a/src/backend/integrations/internal_mcp/transport.py b/src/backend/integrations/internal_mcp/transport.py
index 846356fd8..500d17905 100644
--- a/src/backend/integrations/internal_mcp/transport.py
+++ b/src/backend/integrations/internal_mcp/transport.py
@@ -676,6 +676,7 @@ def execute(
         category: str = "read",
         advisory_timeout_sec: float | None = None,
         deadline_monotonic: float | None = None,
+        require_configured_timeout: bool = False,
         log_tag: str = "[mcp_gateway]",
         late_completion_observer: LateCompletionObserver | None = None,
     ) -> TransportResult:
@@ -729,6 +730,58 @@ def execute(
             else "discard_from_turn"
         )
 
+        if (
+            require_configured_timeout
+            and deadline_monotonic is not None
+            and hard_timeout_sec < configured_hard_timeout_sec
+        ):
+            is_write = str(category or "").strip().lower() == "write"
+            payload: Dict[str, Any] = {
+                "success": False,
+                "status": "not_started",
+                "error": (
+                    f"{method_name!r} was not started because the caller's "
+                    "remaining window is shorter than its configured hard "
+                    "execution window."
+                ),
+                "error_code": (
+                    "insufficient_effect_window"
+                    if is_write
+                    else "insufficient_execution_window"
+                ),
+                "error_type": "admission_denied",
+                "retryable": True,
+                "execution_id": execution_id,
+                "configured_execution_window_seconds": (
+                    configured_hard_timeout_sec
+                ),
+                "remaining_execution_window_seconds": hard_timeout_sec,
+                "timeout_phase": "pre_dispatch",
+                "outcome_finality": "terminal_for_turn",
+                "recovery_affordances": [
+                    {
+                        "action_type": (
+                            "return_bounded_failure_or_retry_in_new_turn"
+                        )
+                    }
+                ],
+            }
+            if is_write:
+                payload["mutation_outcome"] = "not_started"
+            return TransportResult(
+                payload=payload,
+                duration_ms=max(0.0, (time.perf_counter() - submitted_at) * 1000.0),
+                execution_id=execution_id,
+                outcome="not_started",
+                timeout_sec=hard_timeout_sec,
+                advisory_timeout_sec=advisory_sec,
+                queue_duration_ms=0.0,
+                handler_duration_ms=None,
+                handler_elapsed_ms=None,
+                transport_overhead_ms=0.0,
+                timeout_phase="pre_dispatch",
+            )
+
         if hard_timeout_sec <= 0.0:
             with self._diagnostics_lock:
                 self._timeout_count += 1
diff --git a/src/backend/languagemodels/llm_interface.py b/src/backend/languagemodels/llm_interface.py
index 3e307cbcd..d45380689 100644
--- a/src/backend/languagemodels/llm_interface.py
+++ b/src/backend/languagemodels/llm_interface.py
@@ -1785,17 +1785,15 @@ def _split_request_timeout_from_llm_params(
     def _get_structured_client_config(self, model: Optional[str]) -> LLMClientConfig:
         """Get configuration for structured tool calling client (JVNAUTOSCI-799)."""
         resolved_model = resolve_openai_model_name(model)
-        safe_temperature = resolve_safe_temperature_for_model(
-            resolved_model or self.DEFAULT_MODEL,
-            0.7,
-        )
         return LLMClientConfig(
             model=resolved_model or self.DEFAULT_MODEL,
             provider="openai",
             api_key=self.api_key,
             connection_id="#V#openai_provider",
             deployment_id=resolved_model or self.DEFAULT_MODEL,
-            temperature=safe_temperature,
+            # The effective API surface is selected by the structured adapter.
+            # Defer optional-parameter policy until that surface is known.
+            temperature=0.7,
         )
 
     def validate_model_response(self, response, model: str) -> str:
diff --git a/src/backend/languagemodels/structured_tool_calling/client.py b/src/backend/languagemodels/structured_tool_calling/client.py
index 69565c05c..5dc02fa2f 100644
--- a/src/backend/languagemodels/structured_tool_calling/client.py
+++ b/src/backend/languagemodels/structured_tool_calling/client.py
@@ -64,6 +64,7 @@ def resolve_safe_temperature_for_model(
     temperature: Optional[float],
     *,
     api_surface: str = "chat_completions",
+    profile_concept_id: str | None = None,
 ) -> Optional[float]:
     """Return a temperature that is safe to send for the given model.
 
@@ -80,6 +81,7 @@ def resolve_safe_temperature_for_model(
         parameter="temperature",
         value=temperature,
         api_surface=api_surface,
+        profile_concept_id=profile_concept_id,
     )
     return sanitised if isinstance(sanitised, (int, float)) else None
 
diff --git a/src/backend/languagemodels/structured_tool_calling/providers/openai_client.py b/src/backend/languagemodels/structured_tool_calling/providers/openai_client.py
index 60656781a..940b8337b 100644
--- a/src/backend/languagemodels/structured_tool_calling/providers/openai_client.py
+++ b/src/backend/languagemodels/structured_tool_calling/providers/openai_client.py
@@ -68,6 +68,35 @@ def _provider_item_mapping(item: Any) -> dict[str, Any]:
     return {}
 
 
+def _raw_reasoning_effort(raw: Any) -> Any:
+    if not isinstance(raw, Mapping):
+        return None
+    payload = raw.get("model_parameters")
+    if not isinstance(payload, Mapping):
+        payload = raw
+    effort = payload.get("reasoning_effort")
+    if effort is None:
+        effort = payload.get("reasoningEffort")
+    nested = payload.get("reasoning")
+    if effort is None and isinstance(nested, Mapping):
+        effort = nested.get("effort")
+    return effort
+
+
+def _merge_provider_parameter_kwargs(
+    target: dict[str, Any],
+    source: Mapping[str, Any],
+) -> None:
+    """Merge provider projections without discarding adjacent nested options."""
+
+    for key, value in source.items():
+        current = target.get(key)
+        if isinstance(current, Mapping) and isinstance(value, Mapping):
+            target[key] = {**dict(current), **dict(value)}
+        else:
+            target[key] = value
+
+
 class OpenAIClient(LLMClient):
     """OpenAI client whose wire surface is selected from represented profiles."""
 
@@ -456,14 +485,33 @@ async def _generate_chat_completions(
             dict(llm_params) if isinstance(llm_params, Mapping) else {}
         )
         effective_parameters: dict[str, Any] = {}
+        direct_reasoning_effort = request_kwargs.pop("reasoning_effort", None)
+        if (
+            _raw_reasoning_effort(llm_params) is None
+            and direct_reasoning_effort is not None
+        ):
+            requested_parameters["reasoning_effort"] = direct_reasoning_effort
+            _merge_provider_parameter_kwargs(
+                effective_parameters,
+                openai_chat_completions_kwargs_from_model_parameters(
+                    {"reasoning_effort": direct_reasoning_effort},
+                    model=request_model,
+                    profile_concept_id=decision.profile_concept_id,
+                ),
+            )
 
         if self.config.max_tokens is not None:
             request_kwargs["max_tokens"] = self.config.max_tokens
         if isinstance(llm_params, Mapping) and llm_params:
-            effective_parameters = openai_chat_completions_kwargs_from_model_parameters(
-                llm_params, model=request_model
+            _merge_provider_parameter_kwargs(
+                effective_parameters,
+                openai_chat_completions_kwargs_from_model_parameters(
+                    llm_params,
+                    model=request_model,
+                    profile_concept_id=decision.profile_concept_id,
+                ),
             )
-            request_kwargs.update(effective_parameters)
+        _merge_provider_parameter_kwargs(request_kwargs, effective_parameters)
         if (
             tools
             and "reasoning_effort" in request_kwargs
@@ -475,13 +523,26 @@ async def _generate_chat_completions(
                 "Omitting non-zero reasoning_effort for a represented Chat "
                 "Completions structured-tool profile."
             )
+        if (
+            "temperature" not in requested_parameters
+            and "temperature" in request_kwargs
+        ):
+            requested_parameters["temperature"] = request_kwargs["temperature"]
+        requested_temperature = requested_parameters.get(
+            "temperature",
+            request_kwargs.get("temperature", self.config.temperature),
+        )
+        request_kwargs.pop("temperature", None)
+        effective_parameters.pop("temperature", None)
         safe_temperature = resolve_safe_temperature_for_model(
             request_model,
-            self.config.temperature,
+            requested_temperature,
             api_surface=API_SURFACE_CHAT_COMPLETIONS,
+            profile_concept_id=decision.profile_concept_id,
         )
         if safe_temperature is not None:
             request_kwargs["temperature"] = safe_temperature
+            effective_parameters["temperature"] = safe_temperature
 
         response = await request_client.chat.completions.create(
             model=request_model,
@@ -542,6 +603,31 @@ async def _generate_responses(
             dict(llm_params) if isinstance(llm_params, Mapping) else {}
         )
         effective_parameters: dict[str, Any] = {}
+        direct_reasoning_effort = request_kwargs.pop("reasoning_effort", None)
+        raw_direct_reasoning = request_kwargs.pop("reasoning", None)
+        if isinstance(raw_direct_reasoning, Mapping):
+            direct_reasoning = dict(raw_direct_reasoning)
+            nested_effort = direct_reasoning.pop("effort", None)
+            if direct_reasoning_effort is None:
+                direct_reasoning_effort = nested_effort
+            if direct_reasoning:
+                request_kwargs["reasoning"] = direct_reasoning
+        elif raw_direct_reasoning is not None:
+            # An invalid raw shape must not bypass represented reasoning policy.
+            requested_parameters["reasoning"] = raw_direct_reasoning
+        if (
+            _raw_reasoning_effort(llm_params) is None
+            and direct_reasoning_effort is not None
+        ):
+            requested_parameters["reasoning_effort"] = direct_reasoning_effort
+            _merge_provider_parameter_kwargs(
+                effective_parameters,
+                openai_responses_kwargs_from_model_parameters(
+                    {"reasoning_effort": direct_reasoning_effort},
+                    model=request_model,
+                    profile_concept_id=decision.profile_concept_id,
+                ),
+            )
         input_items = self._build_responses_input(
             prompt=prompt,
             context=context,
@@ -551,18 +637,35 @@ async def _generate_responses(
         if self.config.max_tokens is not None:
             request_kwargs["max_output_tokens"] = self.config.max_tokens
         if isinstance(llm_params, Mapping) and llm_params:
-            effective_parameters = openai_responses_kwargs_from_model_parameters(
-                llm_params,
-                model=request_model,
+            _merge_provider_parameter_kwargs(
+                effective_parameters,
+                openai_responses_kwargs_from_model_parameters(
+                    llm_params,
+                    model=request_model,
+                    profile_concept_id=decision.profile_concept_id,
+                ),
             )
-            request_kwargs.update(effective_parameters)
+        _merge_provider_parameter_kwargs(request_kwargs, effective_parameters)
+        if (
+            "temperature" not in requested_parameters
+            and "temperature" in request_kwargs
+        ):
+            requested_parameters["temperature"] = request_kwargs["temperature"]
+        requested_temperature = requested_parameters.get(
+            "temperature",
+            request_kwargs.get("temperature", self.config.temperature),
+        )
+        request_kwargs.pop("temperature", None)
+        effective_parameters.pop("temperature", None)
         safe_temperature = resolve_safe_temperature_for_model(
             request_model,
-            self.config.temperature,
+            requested_temperature,
             api_surface=API_SURFACE_RESPONSES,
+            profile_concept_id=decision.profile_concept_id,
         )
         if safe_temperature is not None:
             request_kwargs["temperature"] = safe_temperature
+            effective_parameters["temperature"] = safe_temperature
         request_kwargs["store"] = decision.store
         if decision.continuation_mode == "stateless" and not decision.store:
             requested_include = request_kwargs.get("include")
diff --git a/src/backend/languagemodels/structured_tool_calling/transport.py b/src/backend/languagemodels/structured_tool_calling/transport.py
index d8048da6b..8eff1c5f5 100644
--- a/src/backend/languagemodels/structured_tool_calling/transport.py
+++ b/src/backend/languagemodels/structured_tool_calling/transport.py
@@ -324,7 +324,9 @@ def resolve_structured_tool_transport(
     Unknown profiles preserve existing provider behaviour.  For OpenAI and
     OpenAI-compatible clients that means Chat Completions.  Responses is chosen
     only when a matching represented profile explicitly advertises structured
-    tool support (or requires it).
+    tool support (or requires it).  A represented required surface also remains
+    authoritative for text-only calls so that surface-specific parameter
+    constraints are not lost when a caller deliberately supplies no tools.
     """
 
     provider_key = _normalise(provider) or "unknown"
@@ -347,21 +349,6 @@ def resolve_structured_tool_transport(
             parameter_projection=projection,
         )
 
-    if not tools_present:
-        return StructuredToolTransportDecision(
-            provider=provider_key,
-            model=model_name,
-            effective_api_surface=API_SURFACE_CHAT_COMPLETIONS,
-            status="compatible",
-            reason="no_structured_tools_in_request",
-            capability_class="no_tool_generation",
-            tools_present=False,
-            requested_api_surface=requested_surface,
-            connection_id=connection_id,
-            deployment_id=deployment_id,
-            parameter_projection=projection,
-        )
-
     resolved = resolve_model_api_profiles(model=model_name, provider=provider_key)
     profiles: Sequence[Mapping[str, Any]] = ()
     if isinstance(resolved, Mapping):
@@ -454,51 +441,12 @@ def resolve_structured_tool_transport(
         invalid_applicable = [
             item for item in invalid_applicable if item[3] == max_specificity
         ]
-    advertised = [
-        (item[0], item[1], item[2], item[3], item[5])
-        for item in applicable
-        if item[2] != "unsupported"
-    ]
-    explicitly_unsupported = [
-        item[1] for item in applicable if item[2] == "unsupported"
-    ]
-
-    conflicting_surface = next(
-        (
-            surface
-            for surface in sorted(_OPENAI_STRUCTURED_TOOL_API_SURFACES)
-            if surface in explicitly_unsupported
-            and any(item[1] == surface for item in advertised)
-        ),
-        None,
-    )
-    required_surfaces = {item[1] for item in advertised if item[2] == "required"}
-
-    selected: tuple[Mapping[str, Any], str, str, str, bool] | None = None
-    required = [item for item in advertised if item[2] == "required"]
-    if requested_surface != "auto" and allow_advertised_surface_override:
-        selected = next(
-            (item for item in advertised if item[1] == requested_surface),
-            None,
-        )
-    elif required:
-        selected = required[0]
-    elif requested_surface != "auto":
-        selected = next(
-            (item for item in advertised if item[1] == requested_surface),
-            None,
-        )
-    if selected is None and advertised:
-        selected = next(
-            (item for item in advertised if item[1] == API_SURFACE_CHAT_COMPLETIONS),
-            advertised[0],
-        )
 
     provenance = dict(resolved or {}) if isinstance(resolved, Mapping) else {}
     common = {
         "provider": provider_key,
         "model": model_name,
-        "tools_present": True,
+        "tools_present": tools_present,
         "requested_api_surface": requested_surface,
         "registry_concept_id": provenance.get("registry_concept_id"),
         "registry_entry_id": provenance.get("registry_entry_id"),
@@ -525,6 +473,28 @@ def resolve_structured_tool_transport(
             advertised_alternatives=(),
         )
 
+    advertised = [
+        (item[0], item[1], item[2], item[3], item[5])
+        for item in applicable
+        if item[2] != "unsupported"
+    ]
+    explicitly_unsupported = [
+        item[1] for item in applicable if item[2] == "unsupported"
+    ]
+    conflicting_surface = next(
+        (
+            surface
+            for surface in sorted(_OPENAI_STRUCTURED_TOOL_API_SURFACES)
+            if surface in explicitly_unsupported
+            and any(item[1] == surface for item in advertised)
+        ),
+        None,
+    )
+    required_surfaces = {item[1] for item in advertised if item[2] == "required"}
+
+    # Contradictory represented authority is invalid independently of whether
+    # this particular request carries tools.  Answer-only traffic must not
+    # silently choose one side of a same-surface contradiction.
     if conflicting_surface is not None:
         return StructuredToolTransportDecision(
             **common,
@@ -547,6 +517,139 @@ def resolve_structured_tool_transport(
             advertised_alternatives=(),
         )
 
+    if not tools_present:
+        required = [item for item in applicable if item[2] == "required"]
+        selected_no_tool: tuple[
+            Mapping[str, Any],
+            str,
+            str,
+            str,
+            str,
+            bool,
+            tuple[int, int, int],
+        ] | None = None
+        if requested_surface != "auto" and allow_advertised_surface_override:
+            selected_no_tool = next(
+                (item for item in applicable if item[1] == requested_surface),
+                None,
+            )
+        elif required:
+            selected_no_tool = required[0]
+        elif requested_surface != "auto":
+            selected_no_tool = next(
+                (item for item in applicable if item[1] == requested_surface),
+                None,
+            )
+
+        if selected_no_tool is not None:
+            (
+                profile,
+                surface,
+                capability,
+                continuation_mode,
+                _storage_policy,
+                store,
+                _specificity,
+            ) = selected_no_tool
+            alternatives = tuple(
+                item[1] for item in applicable if item[1] != surface
+            )
+            return StructuredToolTransportDecision(
+                **common,
+                effective_api_surface=surface,
+                status="compatible",
+                reason=f"represented_profile_{capability}",
+                capability_class=(
+                    "responses_required"
+                    if surface == API_SURFACE_RESPONSES
+                    and capability == "required"
+                    else "represented_text_surface"
+                ),
+                capability_source=str(
+                    provenance.get("source") or "model_registry"
+                ),
+                profile_concept_id=(
+                    str(profile.get("profile_concept_id"))
+                    if profile.get("profile_concept_id")
+                    else None
+                ),
+                continuation_mode=continuation_mode,
+                store=store,
+                advertised_alternatives=alternatives,
+            )
+
+        # A profile can prohibit structured tools while still governing ordinary
+        # text generation on its surface. Preserve that exact profile when the
+        # conservative answer-only surface is Chat Completions so downstream
+        # parameter policy cannot drift to another connection's Chat profile.
+        conservative_chat_profile = next(
+            (
+                item
+                for item in applicable
+                if item[1] == API_SURFACE_CHAT_COMPLETIONS
+            ),
+            None,
+        )
+        if conservative_chat_profile is not None:
+            (
+                profile,
+                surface,
+                _capability,
+                continuation_mode,
+                _storage_policy,
+                store,
+                _specificity,
+            ) = conservative_chat_profile
+            return StructuredToolTransportDecision(
+                **common,
+                effective_api_surface=surface,
+                status="compatible",
+                reason="no_structured_tools_in_request",
+                capability_class="no_tool_generation",
+                capability_source=str(
+                    provenance.get("source") or "model_registry"
+                ),
+                profile_concept_id=(
+                    str(profile.get("profile_concept_id"))
+                    if profile.get("profile_concept_id")
+                    else None
+                ),
+                continuation_mode=continuation_mode,
+                store=store,
+                advertised_alternatives=tuple(
+                    item[1] for item in applicable if item[1] != surface
+                ),
+            )
+
+        return StructuredToolTransportDecision(
+            **common,
+            effective_api_surface=API_SURFACE_CHAT_COMPLETIONS,
+            status="compatible",
+            reason="no_structured_tools_in_request",
+            capability_class="no_tool_generation",
+            capability_source="conservative_client_default",
+        )
+
+    selected: tuple[Mapping[str, Any], str, str, str, bool] | None = None
+    required = [item for item in advertised if item[2] == "required"]
+    if requested_surface != "auto" and allow_advertised_surface_override:
+        selected = next(
+            (item for item in advertised if item[1] == requested_surface),
+            None,
+        )
+    elif required:
+        selected = required[0]
+    elif requested_surface != "auto":
+        selected = next(
+            (item for item in advertised if item[1] == requested_surface),
+            None,
+        )
+    if selected is None and advertised:
+        selected = next(
+            (item for item in advertised if item[1] == API_SURFACE_CHAT_COMPLETIONS),
+            advertised[0],
+        )
+
     if selected is not None:
         profile, surface, capability, continuation_mode, store = selected
         alternatives = tuple(item[1] for item in advertised if item[1] != surface)
diff --git a/src/backend/mcp_server/mcp_stdio_server.py b/src/backend/mcp_server/mcp_stdio_server.py
index e80ebc80d..299d97945 100644
--- a/src/backend/mcp_server/mcp_stdio_server.py
+++ b/src/backend/mcp_server/mcp_stdio_server.py
@@ -947,12 +947,17 @@ def get_method_definition(self, method_name: str) -> Any:
             return definition
         return None
 
+    def get_method_timeout_sec(self, method_name: str) -> float | None:
+        return self._gateway.get_method_timeout_sec(method_name)
+
     def invoke(
         self,
         method_name: str,
         payload: dict[str, Any] | None = None,
         *,
         deadline_monotonic: float | None = None,
+        late_completion_observer: Any | None = None,
+        require_configured_timeout: bool = False,
     ):
         if not self._allow_writes:
             meta = self._gateway.describe_methods().get(method_name) or {}
@@ -966,6 +971,8 @@ def invoke(
             method_name,
             payload,
             deadline_monotonic=deadline_monotonic,
+            late_completion_observer=late_completion_observer,
+            require_configured_timeout=require_configured_timeout,
         )
 
 
diff --git a/src/backend/services/adaptive_turn_service.py b/src/backend/services/adaptive_turn_service.py
index 56eb2e36b..efe7877d7 100644
--- a/src/backend/services/adaptive_turn_service.py
+++ b/src/backend/services/adaptive_turn_service.py
@@ -2706,6 +2706,53 @@ def invoke_and_contain(
                         canonical_name,
                         is_effect,
                     )
+            if is_effect:
+                configured_effect_window = gateway.get_method_timeout_sec(
+                    canonical_name
+                )
+                remaining_research_window = max(
+                    0.0,
+                    research_deadline - clock(),
+                )
+                if (
+                    configured_effect_window is not None
+                    and configured_effect_window > remaining_research_window
+                ):
+                    return index, (
+                        {
+                            **_error_payload(
+                                "insufficient_effect_window",
+                                (
+                                    f"{canonical_name!r} was not started because "
+                                    "the remaining research window is shorter "
+                                    "than its configured hard execution window."
+                                ),
+                                retryable=True,
+                            ),
+                            "status": "not_started",
+                            "mutation_outcome": "not_started",
+                            "outcome_finality": "terminal_for_turn",
+                            "configured_effect_window_seconds": round(
+                                configured_effect_window,
+                                6,
+                            ),
+                            "remaining_research_window_seconds": round(
+                                remaining_research_window,
+                                6,
+                            ),
+                            "recovery_affordances": [
+                                {
+                                    "action_type": (
+                                        "return_bounded_failure_or_retry_in_new_turn"
+                                    )
+                                }
+                            ],
+                        },
+                        None,
+                        arguments,
+                        canonical_name,
+                        is_effect,
+                    )
             try:
                 effect_identifier = (
                     _effect_id(
@@ -2733,6 +2780,7 @@ def invoke_and_contain(
                             if effect_identifier is not None
                             else None
                         ),
+                        require_configured_timeout=is_effect,
                     )
                 raw_payload = transport_result.payload
             except SchemaValidationError as exc:
diff --git a/src/backend/services/concept_service.py b/src/backend/services/concept_service.py
index a8aa7ecd7..087efabb2 100644
--- a/src/backend/services/concept_service.py
+++ b/src/backend/services/concept_service.py
@@ -1120,7 +1120,7 @@ def enrich_concept_with_text_relations(
     never migrates or deletes them.  Schema migration belongs to an explicit,
     authorised maintenance operation rather than an ordinary fetch.
     """
-    from .text_value_service import get_texts_for_concept
+    from .text_value_service import get_texts_for_concept, get_texts_for_concepts
 
     if not logger:
         logger = globals().get("logger")
@@ -1139,10 +1139,53 @@ def enrich_concept_with_text_relations(
 
     legacy_names = concept.get("names", [])
 
-    # Fetch names from text relations (authoritative source)
-    names_from_relations = get_texts_for_concept(
-        subject_concept_id=concept_id, predicate="hasName", limit=100
+    # The ordinary fetch path needs three predicates. Fetch them with one
+    # relation/text join while the bounded batch is known to be complete. If
+    # the batch reaches its cap, retain the older per-predicate reads so a
+    # name-heavy concept cannot crowd out content or notes.
+    enrichment_batch_cap = 102
+    batch_query_metadata: Dict[str, Any] = {}
+    batched_rows = get_texts_for_concepts(
+        [concept_id],
+        predicates=("hasName", "hasContent", "hasNote"),
+        limit_per_concept=enrichment_batch_cap,
+        query_metadata=batch_query_metadata,
+    ).get(concept_id, [])
+    relation_query_truncated = batch_query_metadata.get(
+        "relation_query_truncated"
     )
+    batch_complete = (
+        not relation_query_truncated
+        if isinstance(relation_query_truncated, bool)
+        else len(batched_rows) < enrichment_batch_cap
+    )
+    if batch_complete:
+        names_from_relations = [
+            row for row in batched_rows if row.get("predicate") == "hasName"
+        ][:100]
+        content_relations = [
+            row for row in batched_rows if row.get("predicate") == "hasContent"
+        ][:1]
+        note_relations = [
+            row for row in batched_rows if row.get("predicate") == "hasNote"
+        ][:1]
+    else:
+        names_from_relations = get_texts_for_concept(
+            subject_concept_id=concept_id,
+            predicate="hasName",
+            limit=100,
+        )
+        content_relations = get_texts_for_concept(
+            subject_concept_id=concept_id,
+            predicate="hasContent",
+            limit=1,
+        )
+        note_relations = get_texts_for_concept(
+            subject_concept_id=concept_id,
+            predicate="hasNote",
+            limit=1,
+        )
+
     # Convert to frontend format
     concept["names"] = [
         {
@@ -1258,16 +1301,10 @@ def _add_code_name(text: Optional[str]) -> None:
         _add_code_name(maybe_guid)
 
     # Fetch content from text relations (for diary entries, articles, etc.)
-    content_relations = get_texts_for_concept(
-        subject_concept_id=concept_id, predicate="hasContent", limit=1
-    )
     if content_relations:
         concept["content"] = content_relations[0].get("text", "")
 
     # Fetch notes from text relations
-    note_relations = get_texts_for_concept(
-        subject_concept_id=concept_id, predicate="hasNote", limit=1
-    )
     if note_relations:
         concept["note"] = note_relations[0].get("text", "")
 
diff --git a/src/backend/services/create_concepts_parent_resolution_service.py b/src/backend/services/create_concepts_parent_resolution_service.py
index a73636281..6c8c9858d 100644
--- a/src/backend/services/create_concepts_parent_resolution_service.py
+++ b/src/backend/services/create_concepts_parent_resolution_service.py
@@ -11,7 +11,7 @@
 from __future__ import annotations
 
 from dataclasses import dataclass
-from typing import Any, Iterable, Tuple
+from typing import Any, Iterable, Mapping, Tuple
 
 from ..db.repositories.concepts_repository import ConceptsRepository
 from ..utils.concept_id_utils import canonicalise_vontology_concept_id
@@ -32,6 +32,7 @@ class ParentResolutionResult:
     fallback_used: bool
     fallback_candidates_checked: Tuple[str, ...]
     fallback_selected_parent_id: str | None
+    resolved_parent_kind: str | None = None
 
     @property
     def success(self) -> bool:
@@ -45,11 +46,23 @@ def to_dict(self) -> dict[str, Any]:
             "fallback_used": self.fallback_used,
             "fallback_candidates_checked": list(self.fallback_candidates_checked),
             "fallback_selected_parent_id": self.fallback_selected_parent_id,
+            "resolved_parent_kind": self.resolved_parent_kind,
         }
 
 
-def _concept_exists(concept_id: str) -> bool:
-    return bool(ConceptsRepository.find_one({"concept_id": concept_id}))
+def _find_concept(concept_id: str) -> Mapping[str, Any] | None:
+    concept = ConceptsRepository.find_one({"concept_id": concept_id})
+    return concept if isinstance(concept, Mapping) else None
+
+
+def _concept_kind(concept: Mapping[str, Any] | None) -> str | None:
+    if not isinstance(concept, Mapping):
+        return None
+    raw_kind = concept.get("kind")
+    if not isinstance(raw_kind, str):
+        return None
+    kind = raw_kind.strip().lower()
+    return kind or None
 
 
 def _canonicalise_candidates(candidates: Iterable[str]) -> tuple[str, ...]:
@@ -91,7 +104,8 @@ def resolve_parent_for_create_concepts(parent_id: str) -> ParentResolutionResult
     """
 
     canonical_parent = canonicalise_vontology_concept_id(parent_id)
-    if canonical_parent and _concept_exists(canonical_parent):
+    canonical_concept = _find_concept(canonical_parent) if canonical_parent else None
+    if canonical_parent and canonical_concept is not None:
         return ParentResolutionResult(
             requested_parent_id=parent_id,
             canonical_parent_id=canonical_parent,
@@ -99,6 +113,7 @@ def resolve_parent_for_create_concepts(parent_id: str) -> ParentResolutionResult
             fallback_used=False,
             fallback_candidates_checked=(),
             fallback_selected_parent_id=None,
+            resolved_parent_kind=_concept_kind(canonical_concept),
         )
 
     fallback_candidates = _fallback_candidates_for_parent(canonical_parent)
@@ -107,7 +122,8 @@ def resolve_parent_for_create_concepts(parent_id: str) -> ParentResolutionResult
         if candidate == canonical_parent:
             continue
         checked.append(candidate)
-        if _concept_exists(candidate):
+        candidate_concept = _find_concept(candidate)
+        if candidate_concept is not None:
             return ParentResolutionResult(
                 requested_parent_id=parent_id,
                 canonical_parent_id=canonical_parent,
@@ -115,6 +131,7 @@ def resolve_parent_for_create_concepts(parent_id: str) -> ParentResolutionResult
                 fallback_used=True,
                 fallback_candidates_checked=tuple(checked),
                 fallback_selected_parent_id=candidate,
+                resolved_parent_kind=_concept_kind(candidate_concept),
             )
 
     return ParentResolutionResult(
@@ -125,4 +142,3 @@ def resolve_parent_for_create_concepts(parent_id: str) -> ParentResolutionResult
         fallback_candidates_checked=tuple(checked),
         fallback_selected_parent_id=None,
     )
-
diff --git a/src/backend/services/model_parameter_service.py b/src/backend/services/model_parameter_service.py
index dc2f46733..3cda0bfe4 100644
--- a/src/backend/services/model_parameter_service.py
+++ b/src/backend/services/model_parameter_service.py
@@ -164,6 +164,7 @@ def _registry_parameter_policy(
     model: str | None,
     parameter: str,
     api_surface: str | None,
+    profile_concept_id: str | None = None,
 ) -> Mapping[str, Any] | None:
     if not model:
         return None
@@ -177,6 +178,7 @@ def _registry_parameter_policy(
             provider=provider,
             parameter=parameter,
             api_surface=api_surface,
+            profile_concept_id=profile_concept_id,
         )
     except Exception:
         return None
@@ -188,6 +190,7 @@ def build_model_parameter_capabilities(
     model: str | None,
     api_surface: str | None = "responses",
     include_registry: bool = False,
+    profile_concept_id: str | None = None,
 ) -> dict[str, Any]:
     provider_name = _normalise_provider(provider)
     model_id = _clean_text(model)
@@ -211,6 +214,7 @@ def build_model_parameter_capabilities(
             model=model_id,
             parameter=MODEL_PARAMETER_REASONING_EFFORT,
             api_surface=surface,
+            profile_concept_id=profile_concept_id,
         )
         if isinstance(reasoning_policy, Mapping):
             action = str(reasoning_policy.get("action") or "").strip().lower()
@@ -257,6 +261,7 @@ def _normalise_reasoning_effort(
     model: str | None,
     api_surface: str | None,
     include_registry: bool,
+    profile_concept_id: str | None = None,
 ) -> str | None:
     effort = _clean_text(value)
     if not effort:
@@ -269,6 +274,7 @@ def _normalise_reasoning_effort(
         model=model,
         api_surface=api_surface,
         include_registry=include_registry,
+        profile_concept_id=profile_concept_id,
     )
     reasoning = _extract_mapping(capability.get("parameters")).get(
         MODEL_PARAMETER_REASONING_EFFORT
@@ -324,6 +330,7 @@ def provider_kwargs_from_model_parameters(
     model: str | None,
     api_surface: str | None,
     include_registry: bool = True,
+    profile_concept_id: str | None = None,
 ) -> dict[str, Any]:
     surface = _normalise_api_surface(api_surface)
     params = normalise_model_parameters_for_storage(
@@ -332,6 +339,7 @@ def provider_kwargs_from_model_parameters(
         model=model,
         api_surface=surface,
         include_registry=include_registry,
+        profile_concept_id=profile_concept_id,
     )
     if not params:
         return {}
@@ -341,6 +349,7 @@ def provider_kwargs_from_model_parameters(
         model=model,
         api_surface=surface,
         include_registry=include_registry,
+        profile_concept_id=profile_concept_id,
     )
     capability_params = _extract_mapping(capabilities.get("parameters"))
     kwargs: dict[str, Any] = {}
@@ -363,6 +372,7 @@ def normalise_model_parameters_for_storage(
     model: str | None = None,
     api_surface: str | None = "responses",
     include_registry: bool = False,
+    profile_concept_id: str | None = None,
 ) -> dict[str, Any]:
     payload = _extract_model_parameters_payload(raw)
     if not payload:
@@ -395,6 +405,7 @@ def normalise_model_parameters_for_storage(
             model=model,
             api_surface=api_surface,
             include_registry=include_registry,
+            profile_concept_id=profile_concept_id,
         )
         if effort:
             params[MODEL_PARAMETER_REASONING_EFFORT] = effort
@@ -429,6 +440,7 @@ def openai_responses_kwargs_from_model_parameters(
     raw: Any,
     *,
     model: str | None,
+    profile_concept_id: str | None = None,
 ) -> dict[str, Any]:
     return provider_kwargs_from_model_parameters(
         raw,
@@ -436,6 +448,7 @@ def openai_responses_kwargs_from_model_parameters(
         model=model,
         api_surface="responses",
         include_registry=True,
+        profile_concept_id=profile_concept_id,
     )
 
 
@@ -443,6 +456,7 @@ def openai_chat_completions_kwargs_from_model_parameters(
     raw: Any,
     *,
     model: str | None,
+    profile_concept_id: str | None = None,
 ) -> dict[str, Any]:
     return provider_kwargs_from_model_parameters(
         raw,
@@ -450,4 +464,5 @@ def openai_chat_completions_kwargs_from_model_parameters(
         model=model,
         api_surface="chat_completions",
         include_registry=True,
+        profile_concept_id=profile_concept_id,
     )
diff --git a/src/backend/services/model_registry_service.py b/src/backend/services/model_registry_service.py
index d55027ffe..e60e65352 100644
--- a/src/backend/services/model_registry_service.py
+++ b/src/backend/services/model_registry_service.py
@@ -968,47 +968,61 @@ def _entry_model_match_score(
 
 
 def _iter_parameter_constraints_for_entry(
-    entry: Mapping[str, Any], *, api_surface: str | None = None
+    entry: Mapping[str, Any],
+    *,
+    api_surface: str | None = None,
+    profile_concept_id: str | None = None,
 ) -> list[Mapping[str, Any]]:
     profiles = entry.get("api_profiles")
     if not isinstance(profiles, Sequence) or isinstance(profiles, str):
         return []
 
     requested_surface = _normalise_lookup_token(api_surface)
+    requested_profile_id = (
+        profile_concept_id.strip()
+        if isinstance(profile_concept_id, str) and profile_concept_id.strip()
+        else None
+    )
     matching_constraints: list[Mapping[str, Any]] = []
     generic_constraints: list[Mapping[str, Any]] = []
 
     for profile in profiles:
         if not isinstance(profile, Mapping):
             continue
+        if requested_profile_id is not None and (
+            str(profile.get("profile_concept_id") or "").strip()
+            != requested_profile_id
+        ):
+            continue
         constraints = profile.get("parameter_constraints")
         if not isinstance(constraints, Sequence) or isinstance(constraints, str):
             continue
+        projected_constraints = [
+            {
+                **constraint,
+                "profile_concept_id": (
+                    constraint.get("profile_concept_id")
+                    or profile.get("profile_concept_id")
+                ),
+            }
+            for constraint in constraints
+            if isinstance(constraint, Mapping)
+        ]
 
         profile_surface = _normalise_lookup_token(profile.get("api_surface"))
         if not requested_surface:
-            matching_constraints.extend(
-                constraint
-                for constraint in constraints
-                if isinstance(constraint, Mapping)
-            )
+            matching_constraints.extend(projected_constraints)
             continue
 
         if not profile_surface:
-            generic_constraints.extend(
-                constraint
-                for constraint in constraints
-                if isinstance(constraint, Mapping)
-            )
+            generic_constraints.extend(projected_constraints)
             continue
 
         if profile_surface == requested_surface:
-            matching_constraints.extend(
-                constraint
-                for constraint in constraints
-                if isinstance(constraint, Mapping)
-            )
+            matching_constraints.extend(projected_constraints)
 
+    if requested_profile_id is not None:
+        return matching_constraints
     return matching_constraints or generic_constraints
 
 
@@ -1018,6 +1032,7 @@ def resolve_model_parameter_policy(
     parameter: str,
     provider: str | None = None,
     api_surface: str | None = None,
+    profile_concept_id: str | None = None,
     preferred_language: str | None = None,
 ) -> Mapping[str, Any] | None:
     registry_snapshot = get_model_registry_snapshot(
@@ -1031,35 +1046,48 @@ def resolve_model_parameter_policy(
     if not requested_parameter:
         return None
 
-    for entry in models:
-        if not isinstance(entry, Mapping):
-            continue
-        if not _entry_matches_model(entry, model=model, provider=provider):
-            continue
+    matching_entries = [
+        entry
+        for entry in models
+        if isinstance(entry, Mapping)
+        and _entry_matches_model(entry, model=model, provider=provider)
+    ]
+    if not matching_entries:
+        return None
+    entry = max(
+        matching_entries,
+        key=lambda candidate: _entry_model_match_score(
+            candidate,
+            model=model,
+            provider=provider,
+        ),
+    )
 
-        for constraint in _iter_parameter_constraints_for_entry(
-            entry, api_surface=api_surface
-        ):
-            parameter_name = _normalise_parameter_name(
-                constraint.get("parameter") or constraint.get("parameter_concept_id")
-            )
-            if parameter_name != requested_parameter:
-                continue
+    for constraint in _iter_parameter_constraints_for_entry(
+        entry,
+        api_surface=api_surface,
+        profile_concept_id=profile_concept_id,
+    ):
+        parameter_name = _normalise_parameter_name(
+            constraint.get("parameter") or constraint.get("parameter_concept_id")
+        )
+        if parameter_name != requested_parameter:
+            continue
 
-            return {
-                "parameter": parameter_name,
-                "action": constraint.get("action"),
-                "fixed_value": constraint.get("fixed_value"),
-                "allowed_values": list(constraint.get("allowed_values") or []),
-                "constraint_concept_id": constraint.get("constraint_concept_id"),
-                "parameter_concept_id": constraint.get("parameter_concept_id"),
-                "profile_concept_id": constraint.get("profile_concept_id"),
-                "registry_entry_id": entry.get("registry_entry_id"),
-                "concept_id": entry.get("concept_id"),
-                "provider": entry.get("provider"),
-                "model_id": entry.get("model_id"),
-                "source": registry_snapshot.get("source"),
-            }
+        return {
+            "parameter": parameter_name,
+            "action": constraint.get("action"),
+            "fixed_value": constraint.get("fixed_value"),
+            "allowed_values": list(constraint.get("allowed_values") or []),
+            "constraint_concept_id": constraint.get("constraint_concept_id"),
+            "parameter_concept_id": constraint.get("parameter_concept_id"),
+            "profile_concept_id": constraint.get("profile_concept_id"),
+            "registry_entry_id": entry.get("registry_entry_id"),
+            "concept_id": entry.get("concept_id"),
+            "provider": entry.get("provider"),
+            "model_id": entry.get("model_id"),
+            "source": registry_snapshot.get("source"),
+        }
 
     return None
 
@@ -1150,6 +1178,7 @@ def sanitise_model_parameter_value(
     value: Any,
     provider: str | None = None,
     api_surface: str | None = None,
+    profile_concept_id: str | None = None,
     preferred_language: str | None = None,
 ) -> Any:
     if value is None:
@@ -1160,6 +1189,7 @@ def sanitise_model_parameter_value(
         parameter=parameter,
         provider=provider,
         api_surface=api_surface,
+        profile_concept_id=profile_concept_id,
         preferred_language=preferred_language,
     )
     if not isinstance(policy, Mapping):
diff --git a/src/backend/services/text_value_service.py b/src/backend/services/text_value_service.py
index ace47e72c..f3c6e834e 100644
--- a/src/backend/services/text_value_service.py
+++ b/src/backend/services/text_value_service.py
@@ -424,6 +424,7 @@ def get_texts_for_concepts(
     limit_per_concept: int = 50,
     recent_first: bool = False,
     max_time_ms: Optional[int] = None,
+    query_metadata: Optional[Dict[str, Any]] = None,
 ) -> Dict[str, List[Dict[str, Any]]]:
     """Fetch linked TextValues for many concepts with one relation/text join.
 
@@ -450,6 +451,14 @@ def get_texts_for_concepts(
     ]
 
     if not ordered_subject_ids:
+        if query_metadata is not None:
+            query_metadata.update(
+                {
+                    "raw_relation_count": 0,
+                    "relation_query_limit": 0,
+                    "relation_query_truncated": False,
+                }
+            )
         return {}
 
     rel_filter: Dict[str, Any] = {"subject_concept_id": {"$in": ordered_subject_ids}}
@@ -465,7 +474,16 @@ def get_texts_for_concepts(
             rel_filter["predicate"] = {"$in": predicate_values}
 
     sort = [("updated_at", -1), ("created_at", -1)] if recent_first else None
-    relation_limit = max(len(ordered_subject_ids) * max(limit_per_concept, 1), 1)
+    per_concept_relation_limit = max(limit_per_concept, 1)
+    if query_metadata is not None:
+        # One extra raw relation is a completeness sentinel. Joined text rows
+        # cannot provide this signal because dangling TextValue references are
+        # deliberately omitted from the projection below.
+        per_concept_relation_limit += 1
+    relation_limit = max(
+        len(ordered_subject_ids) * per_concept_relation_limit,
+        1,
+    )
     relations = list(
         TextRelationsRepository.find(
             rel_filter,
@@ -474,6 +492,14 @@ def get_texts_for_concepts(
             max_time_ms=max_time_ms,
         )
     )
+    if query_metadata is not None:
+        query_metadata.update(
+            {
+                "raw_relation_count": len(relations),
+                "relation_query_limit": relation_limit,
+                "relation_query_truncated": len(relations) >= relation_limit,
+            }
+        )
     rows_by_concept: Dict[str, List[Dict[str, Any]]] = {
         subject_id: [] for subject_id in ordered_subject_ids
     }
diff --git a/src/backend/services/turn_execution_record_service.py b/src/backend/services/turn_execution_record_service.py
index 56ad9ec98..3e18c14ab 100644
--- a/src/backend/services/turn_execution_record_service.py
+++ b/src/backend/services/turn_execution_record_service.py
@@ -18,7 +18,7 @@
 from typing import Any, Mapping, Sequence, cast
 
 from pymongo import ASCENDING, DESCENDING
-from pymongo.errors import OperationFailure, PyMongoError
+from pymongo.errors import DuplicateKeyError, OperationFailure, PyMongoError
 
 from ..db.mongo_client import get_db
 from .arxiv_paper_link_service import extract_arxiv_id_candidates
@@ -10477,13 +10477,13 @@ def upsert_turn_execution_record_projection(
     payload.pop("late_effect_observations", None)
     payload["request_id"] = request_id
     if _safe_str(user_id):
-        payload.setdefault("user_id", _safe_str(user_id))
+        payload["user_id"] = _safe_str(user_id)
     if _safe_str(session_id):
-        payload.setdefault("session_id", _safe_str(session_id))
+        payload["session_id"] = _safe_str(session_id)
     if _safe_str(namespace):
-        payload.setdefault("namespace", _safe_str(namespace))
+        payload["namespace"] = _safe_str(namespace)
     if _safe_str(org_id):
-        payload.setdefault("org_id", _safe_str(org_id))
+        payload["org_id"] = _safe_str(org_id)
     payload.setdefault("schema_version", TURN_EXECUTION_RECORD_SCHEMA_VERSION)
     payload.setdefault("created_at_utc", _iso_utc(now))
     payload["updated_at_utc"] = _iso_utc(now)
@@ -10497,13 +10497,31 @@ def upsert_turn_execution_record_projection(
     if isinstance(compacted_payload.payload, Mapping):
         payload = dict(compacted_payload.payload)
     payload["request_id"] = request_id
+    for field_name, raw_value in (
+        ("namespace", namespace),
+        ("user_id", user_id),
+        ("org_id", org_id),
+    ):
+        clean_value = _safe_str(raw_value)
+        if clean_value:
+            payload[field_name] = clean_value
     payload.setdefault("schema_version", TURN_EXECUTION_RECORD_SCHEMA_VERSION)
     payload["updated_at_utc"] = _iso_utc(now)
 
+    scope_query: dict[str, Any] = {"request_id": request_id}
+    for field_name, raw_value in (
+        ("namespace", namespace),
+        ("user_id", user_id),
+        ("org_id", org_id),
+    ):
+        clean_value = _safe_str(raw_value)
+        if clean_value:
+            scope_query[field_name] = clean_value
+
     try:
         result = _turn_execution_update_one(
             coll,
-            {"request_id": request_id},
+            scope_query,
             {
                 "$set": payload,
                 "$setOnInsert": {"inserted_at": now},
@@ -10520,6 +10538,19 @@ def upsert_turn_execution_record_projection(
             "matched": matched,
             "request_id": request_id,
         }
+    except DuplicateKeyError:
+        # request_id is globally unique. A scoped upsert that collides with an
+        # existing record therefore denotes a different actor scope, not a
+        # record this caller may replace.
+        logger.warning(
+            "Refused turn_execution_records scope collision for request_id=%s",
+            request_id,
+        )
+        return {
+            "updated": False,
+            "reason": "actor_scope_mismatch",
+            "request_id": request_id,
+        }
     except PyMongoError as exc:
         logger.warning(
             "Failed to upsert turn_execution_records projection for request_id=%s: %s",
@@ -10547,15 +10578,14 @@ def append_late_effect_observation(
     """Append one bounded, identity-bearing late-effect observation.
 
     The entry is additive to the ordinary turn projection: a later full-record
-    ``$set`` upsert does not remove it.  ``$addToSet`` also makes an exact retry
-    of the same observation harmless.  The caller remains responsible for
-    interpreting a handler receipt versus canonical state.
+    ``$set`` upsert does not remove it.  The conditional append is idempotent
+    by ``observation_id``, including when a retry changes volatile receipt
+    fields.  The caller remains responsible for interpreting a handler receipt
+    versus canonical state.
 
     Entries are individually bounded, but the array has no arbitrary count
-    ceiling.  Replacing this atomic idempotent append with ``$push/$slice``
-    would admit duplicates, while a read-trim-write sequence would lose
-    concurrent observations.  A per-turn retention limit should therefore use
-    an atomic pipeline or a separately indexed collection if observed growth
+    ceiling.  A per-turn retention limit should therefore use an atomic
+    pipeline or a separately indexed collection if observed growth
     demonstrates the need.
     """
 
@@ -10575,6 +10605,22 @@ def append_late_effect_observation(
     if not isinstance(observation, Mapping):
         return {"updated": False, "reason": "invalid_observation"}
 
+    actor_scope: dict[str, str] = {}
+    for field_name, raw_value in (
+        ("namespace", namespace),
+        ("user_id", user_id),
+        ("org_id", org_id),
+    ):
+        clean_value = _safe_str(raw_value)
+        if clean_value:
+            actor_scope[field_name] = clean_value
+    if not actor_scope:
+        return {
+            "updated": False,
+            "reason": "missing_actor_scope",
+            "request_id": clean_request_id,
+        }
+
     coll = get_turn_execution_records_collection()
     if coll is None:
         return {
@@ -10583,8 +10629,9 @@ def append_late_effect_observation(
             "request_id": clean_request_id,
         }
 
+    source_observation = dict(observation)
     bounded_observation = _compact_final_answer_projection_payload(
-        dict(observation),
+        source_observation,
         max_depth=4,
         max_items=32,
         max_string_chars=2_000,
@@ -10592,6 +10639,9 @@ def append_late_effect_observation(
     if not isinstance(bounded_observation, Mapping):
         bounded_observation = {"value": bounded_observation}
     bounded_observation = dict(bounded_observation)
+    storage_transformed = _hash_payload(source_observation) != _hash_payload(
+        bounded_observation
+    )
 
     observed_at_utc = (
         _safe_str(bounded_observation.get("observed_at_utc"))
@@ -10611,6 +10661,7 @@ def append_late_effect_observation(
             "effect_id": clean_effect_id,
             "execution_id": clean_execution_id,
             "observed_at_utc": observed_at_utc,
+            "storage_transformed": storage_transformed,
         }
     )
 
@@ -10632,25 +10683,164 @@ def append_late_effect_observation(
         if clean_value:
             insert_payload[field_name] = clean_value
 
+    record_scope_query = {
+        "request_id": clean_request_id,
+        **actor_scope,
+    }
+    append_query = {
+        **record_scope_query,
+        "late_effect_observations": {
+            "$not": {"$elemMatch": {"observation_id": observation_id}}
+        },
+    }
+    persisted_at_utc = _iso_utc(now)
+    append_update = {
+        "$push": {"late_effect_observations": entry},
+        "$max": {
+            "updated_at_utc": persisted_at_utc,
+            "late_effect_updated_at_utc": persisted_at_utc,
+        },
+    }
+
     try:
         result = _turn_execution_update_one(
             coll,
-            {"request_id": clean_request_id},
-            {
-                "$addToSet": {"late_effect_observations": entry},
-                "$setOnInsert": insert_payload,
-            },
+            append_query,
+            append_update,
             operation="append_late_effect_observation.update_one",
             detail=clean_execution_id,
-            upsert=True,
+            upsert=False,
         )
         appended = bool(getattr(result, "modified_count", 0) > 0)
-        inserted = getattr(result, "upserted_id", None) is not None
         matched = bool(getattr(result, "matched_count", 0) > 0)
+        inserted = False
+        if not matched:
+            existing = _turn_execution_find_one(
+                coll,
+                record_scope_query,
+                projection={
+                    "_id": 0,
+                    "late_effect_observations.observation_id": 1,
+                },
+                operation="append_late_effect_observation.find_existing",
+                detail=clean_execution_id,
+            )
+            existing_observations = (
+                existing.get("late_effect_observations")
+                if isinstance(existing, Mapping)
+                and isinstance(existing.get("late_effect_observations"), list)
+                else []
+            )
+            duplicate = any(
+                isinstance(item, Mapping)
+                and item.get("observation_id") == observation_id
+                for item in existing_observations
+            )
+            if duplicate:
+                return {
+                    "updated": False,
+                    "appended": False,
+                    "duplicate": True,
+                    "inserted": False,
+                    "matched": True,
+                    "request_id": clean_request_id,
+                    "effect_id": clean_effect_id,
+                    "execution_id": clean_execution_id,
+                    "observation_id": observation_id,
+                }
+
+            if existing is None:
+                request_id_collision = _turn_execution_find_one(
+                    coll,
+                    {"request_id": clean_request_id},
+                    projection={
+                        "_id": 0,
+                        "namespace": 1,
+                        "user_id": 1,
+                        "org_id": 1,
+                    },
+                    operation="append_late_effect_observation.find_scope_collision",
+                    detail=clean_execution_id,
+                )
+                if request_id_collision is not None:
+                    return {
+                        "updated": False,
+                        "appended": False,
+                        "duplicate": False,
+                        "inserted": False,
+                        "matched": False,
+                        "reason": "actor_scope_mismatch",
+                        "request_id": clean_request_id,
+                        "effect_id": clean_effect_id,
+                        "execution_id": clean_execution_id,
+                        "observation_id": observation_id,
+                    }
+                ensured = _turn_execution_update_one(
+                    coll,
+                    record_scope_query,
+                    {"$setOnInsert": insert_payload},
+                    operation="append_late_effect_observation.ensure_record",
+                    detail=clean_execution_id,
+                    upsert=True,
+                )
+                inserted = getattr(ensured, "upserted_id", None) is not None
+
+            # Re-evaluate the identity predicate after the record existence
+            # check so concurrent submissions still admit at most one entry.
+            result = _turn_execution_update_one(
+                coll,
+                append_query,
+                append_update,
+                operation="append_late_effect_observation.retry_update_one",
+                detail=clean_execution_id,
+                upsert=False,
+            )
+            appended = bool(getattr(result, "modified_count", 0) > 0)
+            matched = bool(getattr(result, "matched_count", 0) > 0)
+
+        duplicate = False
+        if not appended and not matched:
+            final_existing = _turn_execution_find_one(
+                coll,
+                record_scope_query,
+                projection={
+                    "_id": 0,
+                    "late_effect_observations.observation_id": 1,
+                },
+                operation="append_late_effect_observation.confirm_identity",
+                detail=clean_execution_id,
+            )
+            final_observations = (
+                final_existing.get("late_effect_observations")
+                if isinstance(final_existing, Mapping)
+                and isinstance(
+                    final_existing.get("late_effect_observations"), list
+                )
+                else []
+            )
+            duplicate = any(
+                isinstance(item, Mapping)
+                and item.get("observation_id") == observation_id
+                for item in final_observations
+            )
+            if not duplicate:
+                return {
+                    "updated": False,
+                    "appended": False,
+                    "duplicate": False,
+                    "inserted": inserted,
+                    "matched": False,
+                    "reason": "append_not_acknowledged",
+                    "request_id": clean_request_id,
+                    "effect_id": clean_effect_id,
+                    "execution_id": clean_execution_id,
+                    "observation_id": observation_id,
+                }
+
         return {
-            "updated": appended or inserted,
-            "appended": appended or inserted,
-            "duplicate": matched and not appended,
+            "updated": appended,
+            "appended": appended,
+            "duplicate": duplicate,
             "inserted": inserted,
             "matched": matched,
             "request_id": clean_request_id,
@@ -10689,7 +10879,15 @@ def submit_late_effect_observation(**kwargs: Any) -> bool:
 
     def _append() -> None:
         try:
-            append_late_effect_observation(**kwargs)
+            outcome = append_late_effect_observation(**kwargs)
+            if not outcome.get("updated") and not outcome.get("duplicate"):
+                logger.warning(
+                    "Late-effect observation was not persisted; request_id=%s "
+                    "effect_id=%s reason=%s",
+                    _safe_str(kwargs.get("request_id")),
+                    _safe_str(kwargs.get("effect_id")),
+                    _safe_str(outcome.get("reason")) or "append_not_acknowledged",
+                )
         except Exception:
             logger.exception(
                 "Late-effect observation append failed; request_id=%s "
diff --git a/tests/backend/test_adaptive_turn_service.py b/tests/backend/test_adaptive_turn_service.py
index f2d62663a..227d20d43 100644
--- a/tests/backend/test_adaptive_turn_service.py
+++ b/tests/backend/test_adaptive_turn_service.py
@@ -1090,6 +1090,75 @@ def append_observation(**kwargs: Any) -> dict[str, Any]:
     assert result.tool_invocations[0]["effect_status"] == "indeterminate"
 
 
+def test_effect_is_not_started_without_its_configured_execution_window() -> None:
+    started = time.monotonic()
+    clock = _ManualClock(started)
+    invoked: list[str] = []
+
+    def handler(name: str, _arguments: dict[str, Any]) -> dict[str, Any]:
+        invoked.append(name)
+        return {
+            "success": True,
+            "effect_status": "succeeded",
+            "changed": True,
+        }
+
+    client = _TimedSequenceClient(
+        clock,
+        (
+            started + 7.5,
+            LLMResponse(
+                text_response="",
+                tool_calls=[
+                    ToolCall(
+                        tool_name="turn_invoke_capability",
+                        call_id="effect-with-clipped-window",
+                        payload={
+                            "name": "create_concepts",
+                            "arguments": {
+                                "concepts": [{"name": "Must not start late"}]
+                            },
+                        },
+                    )
+                ],
+            ),
+        ),
+        (
+            started + 7.6,
+            LLMResponse(
+                text_response=(
+                    "The write was not started because its full bounded window "
+                    "was no longer available."
+                )
+            ),
+        ),
+    )
+
+    result = execute_adaptive_turn(
+        gateway=_effect_gateway(handler, write_timeout_sec=1.0),
+        prompt="Represent this concept.",
+        context=[],
+        llm_client=client,
+        model="test-model",
+        user_namespace="#V#person@org",
+        user_concept_id="#V#person",
+        org_concept_id="#V#org",
+        turn_id="effect-window-admission",
+        turn_budget_seconds=10,
+        final_synthesis_reserve_seconds=2,
+        clock=clock,
+    )
+
+    assert invoked == []
+    assert result.response_text.startswith("The write was not started")
+    assert result.tool_invocations[0]["effect_status"] == "failed"
+    assert result.tool_invocations[0]["changed"] is False
+    assert "insufficient_effect_window" in (
+        result.tool_invocations[0]["evidence"]["preview"]
+    )
+    assert "not_started" in result.tool_invocations[0]["evidence"]["preview"]
+
+
 @pytest.mark.parametrize(
     "relationships",
     [
diff --git a/tests/backend/test_concept_service_read_enrichment.py b/tests/backend/test_concept_service_read_enrichment.py
index 4e918d139..719b50863 100644
--- a/tests/backend/test_concept_service_read_enrichment.py
+++ b/tests/backend/test_concept_service_read_enrichment.py
@@ -2,25 +2,33 @@
 
 
 def test_enrichment_is_read_only_and_preserves_legacy_text(monkeypatch):
-    calls: list[str] = []
+    calls: list[tuple[list[str], tuple[str, ...], int]] = []
 
-    def _get_texts(*, predicate, **_kwargs):
-        calls.append(predicate)
-        if predicate == "hasName":
-            return [
+    def _get_texts(concept_ids, *, predicates, limit_per_concept, **_kwargs):
+        calls.append((list(concept_ids), tuple(predicates), limit_per_concept))
+        return {
+            "#V#legacy_projection": [
                 {
                     "text": "Relation name",
                     "lang": "en-NZ",
+                    "predicate": "hasName",
                     "context": {"name_type": "NL"},
                     "relation_id": "rel-1",
                 }
             ]
-        return []
+        }
 
     def _unexpected_write(*_args, **_kwargs):
         raise AssertionError("read enrichment must not write")
 
-    monkeypatch.setattr(text_value_service, "get_texts_for_concept", _get_texts)
+    monkeypatch.setattr(text_value_service, "get_texts_for_concepts", _get_texts)
+    monkeypatch.setattr(
+        text_value_service,
+        "get_texts_for_concept",
+        lambda **_kwargs: (_ for _ in ()).throw(
+            AssertionError("complete batch must not issue per-predicate reads")
+        ),
+    )
     monkeypatch.setattr(text_value_service, "upsert_text_for_concept", _unexpected_write)
     monkeypatch.setattr(
         concept_service.ConceptsRepository,
@@ -44,23 +52,30 @@ def _unexpected_write(*_args, **_kwargs):
     ]
     assert enriched["names"][1]["relation_id"] is None
     assert enriched["description"] == "Legacy description"
-    assert calls == ["hasName", "hasContent", "hasNote"]
+    assert calls == [
+        (
+            ["#V#legacy_projection"],
+            ("hasName", "hasContent", "hasNote"),
+            102,
+        )
+    ]
 
 
 def test_enrichment_deduplicates_legacy_name_already_in_relations(monkeypatch):
-    def _get_texts(*, predicate, **_kwargs):
-        if predicate == "hasName":
-            return [
+    def _get_texts(concept_ids, **_kwargs):
+        return {
+            concept_ids[0]: [
                 {
                     "text": "Same name",
                     "lang": "en",
+                    "predicate": "hasName",
                     "context": {"name_type": "NL"},
                     "relation_id": "rel-1",
                 }
             ]
-        return []
+        }
 
-    monkeypatch.setattr(text_value_service, "get_texts_for_concept", _get_texts)
+    monkeypatch.setattr(text_value_service, "get_texts_for_concepts", _get_texts)
 
     enriched = concept_service.enrich_concept_with_text_relations(
         {
@@ -70,3 +85,157 @@ def _get_texts(*, predicate, **_kwargs):
     )
 
     assert [item["name"] for item in enriched["names"]].count("Same name") == 1
+
+
+def test_enrichment_falls_back_when_combined_batch_reaches_cap(monkeypatch):
+    batched_rows = [
+        {
+            "text": f"Batch name {index}",
+            "lang": "en",
+            "predicate": "hasName",
+            "context": {"name_type": "NL"},
+            "relation_id": f"batch-{index}",
+        }
+        for index in range(102)
+    ]
+    single_calls: list[str] = []
+
+    def _truncated_batch(concept_ids, *, query_metadata, **_kwargs):
+        query_metadata["relation_query_truncated"] = True
+        return {concept_ids[0]: batched_rows}
+
+    monkeypatch.setattr(
+        text_value_service,
+        "get_texts_for_concepts",
+        _truncated_batch,
+    )
+
+    def _get_texts(*, predicate, **_kwargs):
+        single_calls.append(predicate)
+        return {
+            "hasName": [
+                {
+                    "text": "Exact fallback name",
+                    "lang": "en",
+                    "context": {"name_type": "NL"},
+                    "relation_id": "fallback-name",
+                }
+            ],
+            "hasContent": [{"text": "Exact fallback content"}],
+            "hasNote": [{"text": "Exact fallback note"}],
+        }[predicate]
+
+    monkeypatch.setattr(text_value_service, "get_texts_for_concept", _get_texts)
+
+    enriched = concept_service.enrich_concept_with_text_relations(
+        {"concept_id": "#V#name_heavy_projection"}
+    )
+
+    assert single_calls == ["hasName", "hasContent", "hasNote"]
+    assert enriched["names"][0]["name"] == "Exact fallback name"
+    assert enriched["content"] == "Exact fallback content"
+    assert enriched["note"] == "Exact fallback note"
+
+
+def test_enrichment_uses_complete_102_relation_batch_without_fallback(monkeypatch):
+    batched_rows = [
+        {
+            "text": f"Name {index}",
+            "lang": "en",
+            "predicate": "hasName",
+            "context": {"name_type": "NL"},
+            "relation_id": f"name-{index}",
+        }
+        for index in range(100)
+    ]
+    batched_rows.extend(
+        [
+            {
+                "text": "Complete content",
+                "lang": "en",
+                "predicate": "hasContent",
+                "relation_id": "content-1",
+            },
+            {
+                "text": "Complete note",
+                "lang": "en",
+                "predicate": "hasNote",
+                "relation_id": "note-1",
+            },
+        ]
+    )
+
+    def _complete_batch(concept_ids, *, query_metadata, **_kwargs):
+        query_metadata["relation_query_truncated"] = False
+        return {concept_ids[0]: batched_rows}
+
+    monkeypatch.setattr(
+        text_value_service,
+        "get_texts_for_concepts",
+        _complete_batch,
+    )
+    monkeypatch.setattr(
+        text_value_service,
+        "get_texts_for_concept",
+        lambda **_kwargs: (_ for _ in ()).throw(
+            AssertionError("complete raw relation batch must not fall back")
+        ),
+    )
+
+    enriched = concept_service.enrich_concept_with_text_relations(
+        {"concept_id": "#V#complete_102_projection"}
+    )
+
+    assert len(
+        [item for item in enriched["names"] if item["relation_id"] is not None]
+    ) == 100
+    assert enriched["content"] == "Complete content"
+    assert enriched["note"] == "Complete note"
+
+
+def test_text_batch_truncation_uses_raw_relations_not_joined_rows(monkeypatch):
+    concept_id = "#V#raw_relation_sentinel"
+    relations = [
+        {
+            "_id": f"rel-{index}",
+            "subject_concept_id": concept_id,
+            "object_text_id": f"tv-{index}",
+            "predicate": "hasName",
+        }
+        for index in range(103)
+    ]
+    text_values = [
+        {"_id": f"tv-{index}", "text": f"Name {index}", "lang": "en"}
+        for index in range(102)
+    ]
+
+    monkeypatch.setattr(
+        text_value_service,
+        "filter_accessible_concept_ids",
+        lambda concept_ids: set(concept_ids),
+    )
+    monkeypatch.setattr(
+        text_value_service.TextRelationsRepository,
+        "find",
+        lambda *_args, **_kwargs: relations,
+    )
+    monkeypatch.setattr(
+        text_value_service.TextValuesRepository,
+        "find",
+        lambda *_args, **_kwargs: text_values,
+    )
+    metadata = {}
+
+    rows = text_value_service.get_texts_for_concepts(
+        [concept_id],
+        predicates=("hasName", "hasContent", "hasNote"),
+        limit_per_concept=102,
+        query_metadata=metadata,
+    )[concept_id]
+
+    assert len(rows) == 102
+    assert metadata == {
+        "raw_relation_count": 103,
+        "relation_query_limit": 103,
+        "relation_query_truncated": True,
+    }
diff --git a/tests/backend/test_internal_mcp_transport_deadlines.py b/tests/backend/test_internal_mcp_transport_deadlines.py
index d92b3d379..22e259fa4 100644
--- a/tests/backend/test_internal_mcp_transport_deadlines.py
+++ b/tests/backend/test_internal_mcp_transport_deadlines.py
@@ -26,6 +26,7 @@ def _gateway_for(
     transport: InternalMCPTransport,
     category: str = "read",
     output_schema: Schema | None = None,
+    timeout_sec: float | None = None,
 ) -> InternalMCPGateway:
     catalogue = MethodCatalogue()
     catalogue.register(
@@ -35,6 +36,7 @@ def _gateway_for(
             input_schema=Schema(required={}, optional={}, allow_unknown=False),
             output_schema=output_schema,
             category=category,
+            timeout_sec=timeout_sec,
         )
     )
     return InternalMCPGateway(
@@ -187,6 +189,39 @@ def _handler():
     assert cancellation_seen.wait(timeout=1.0)
 
 
+def test_required_configured_write_window_is_denied_atomically_before_dispatch() -> None:
+    handler_called = Event()
+    executor = _BoundedHandlerExecutor(worker_count=1, queue_capacity=1)
+    transport = InternalMCPTransport(
+        write_timeout_sec=0.5,
+        write_advisory_timeout_sec=0.1,
+        handler_executor=executor,
+    )
+    gateway = _gateway_for(
+        method_name="synthetic_full_window_write",
+        handler=lambda: handler_called.set(),
+        transport=transport,
+        category="write",
+        timeout_sec=0.2,
+    )
+
+    result = gateway.invoke(
+        "synthetic_full_window_write",
+        {},
+        deadline_monotonic=time.monotonic() + 0.19,
+        require_configured_timeout=True,
+    )
+
+    assert result.outcome == "not_started"
+    assert result.timeout_phase == "pre_dispatch"
+    assert result.payload["error_code"] == "insufficient_effect_window"
+    assert result.payload["mutation_outcome"] == "not_started"
+    assert result.payload["configured_execution_window_seconds"] == 0.2
+    assert 0.0 < result.payload["remaining_execution_window_seconds"] < 0.2
+    assert handler_called.is_set() is False
+    assert executor.diagnostics()["submitted_count"] == 0
+
+
 def test_result_completed_after_absolute_deadline_is_discarded() -> None:
     class _AfterDeadlineExecutor:
         @staticmethod
diff --git a/tests/backend/test_llm_openai_temperature_guards.py b/tests/backend/test_llm_openai_temperature_guards.py
index 2d41f3dbc..8f66a5894 100644
--- a/tests/backend/test_llm_openai_temperature_guards.py
+++ b/tests/backend/test_llm_openai_temperature_guards.py
@@ -4,6 +4,8 @@
 
 import types
 
+import pytest
+
 import src.backend.services  # noqa: F401
 
 
@@ -55,6 +57,43 @@ def _install_openai_temperature_registry(monkeypatch) -> None:
                     }
                 ],
             },
+            {
+                "model_id": "gpt-5.6-terra",
+                "provider": "openai",
+                "concept_id": "#V#openai_gpt_5_6_terra",
+                "registry_entry_id": "#V#openai_gpt_5_6_terra_registry_entry",
+                "api_profiles": [
+                    {
+                        "profile_concept_id": (
+                            "#V#openai_gpt_5_6_terra_responses_profile"
+                        ),
+                        "api_surface": "responses",
+                        "structured_tool_calling": "required",
+                        "tool_continuation_mode": "stateless",
+                        "response_storage_policy": "disabled",
+                        "parameter_constraints": [
+                            {
+                                "constraint_concept_id": (
+                                    "#V#terra_responses_temperature_omit"
+                                ),
+                                "parameter_concept_id": "#V#temperature_parameter",
+                                "parameter": "temperature",
+                                "action": "omit",
+                            }
+                        ],
+                    },
+                    {
+                        "profile_concept_id": (
+                            "#V#openai_gpt_5_6_terra_chat_completions_profile"
+                        ),
+                        "api_surface": "chat_completions",
+                        "structured_tool_calling": "unsupported",
+                        "tool_continuation_mode": "stateless",
+                        "response_storage_policy": "disabled",
+                        "parameter_constraints": [],
+                    },
+                ],
+            },
         ],
     }
     monkeypatch.setattr(
@@ -76,7 +115,7 @@ def _build_fake_openai_response(model: str, content: str = "ok"):
     )
 
 
-def test_llm_interface_openai_structured_config_omits_temperature_for_gpt5_mini(
+def test_llm_interface_openai_structured_config_defers_temperature_policy(
     monkeypatch,
 ) -> None:
     import src.backend.languagemodels.llm_interface as mod
@@ -92,7 +131,102 @@ def test_llm_interface_openai_structured_config_omits_temperature_for_gpt5_mini(
     config = client._get_structured_client_config("gpt-5-mini")
 
     assert config.model == "gpt-5-mini"
-    assert config.temperature is None
+    assert config.temperature == 0.7
+
+
+def test_llm_interface_answer_only_uses_required_responses_profile(
+    monkeypatch,
+) -> None:
+    import src.backend.languagemodels.llm_interface as mod
+    from src.backend.languagemodels.structured_tool_calling.providers import (
+        openai_client as provider_module,
+    )
+
+    _install_openai_temperature_registry(monkeypatch)
+    captured: dict[str, list[dict[str, object]]] = {
+        "responses": [],
+        "chat": [],
+        "client_options": [],
+    }
+
+    class _FakeAsyncOpenAI:
+        def __init__(self, **_kwargs) -> None:
+            self.responses = types.SimpleNamespace(create=self._responses_create)
+            self.chat = types.SimpleNamespace(
+                completions=types.SimpleNamespace(create=self._chat_create)
+            )
+
+        def with_options(self, **kwargs):
+            captured["client_options"].append(dict(kwargs))
+            return self
+
+        async def close(self) -> None:
+            return None
+
+        async def _responses_create(self, **kwargs):
+            captured["responses"].append(dict(kwargs))
+            return {
+                "id": "resp-answer-only",
+                "model": "gpt-5.6-terra",
+                "output": [
+                    {
+                        "id": "msg-answer-only",
+                        "type": "message",
+                        "role": "assistant",
+                        "content": [
+                            {
+                                "type": "output_text",
+                                "text": "A grounded answer-only response.",
+                            }
+                        ],
+                    }
+                ],
+                "usage": {
+                    "input_tokens": 20,
+                    "output_tokens": 6,
+                    "total_tokens": 26,
+                },
+            }
+
+        async def _chat_create(self, **kwargs):
+            captured["chat"].append(dict(kwargs))
+            raise AssertionError("Answer-only request unexpectedly used Chat")
+
+    monkeypatch.setattr(
+        provider_module.openai,
+        "AsyncOpenAI",
+        _FakeAsyncOpenAI,
+    )
+    monkeypatch.setattr(mod.openai, "OpenAI", lambda **_kwargs: object())
+
+    client = mod.OpenAIClient(api_key="test-key")
+    result = client.generate_with_tools(
+        prompt="Produce the final grounded answer.",
+        available_tools=[],
+        context=[
+            {
+                "role": "user",
+                "content": "Use the bounded evidence already provided.",
+            }
+        ],
+        model="gpt-5.6-terra",
+        system_message="Answer from the supplied evidence. Do not call tools.",
+        llm_params={"request_timeout_seconds": 20.0},
+    )
+
+    assert result.text_response == "A grounded answer-only response."
+    assert captured["chat"] == []
+    assert len(captured["responses"]) == 1
+    assert "temperature" not in captured["responses"][0]
+    assert captured["responses"][0]["store"] is False
+    assert result.transport_metadata["tools_present"] is False
+    assert result.transport_metadata["effective_api_surface"] == "responses"
+    assert result.transport_metadata["profile_concept_id"] == (
+        "#V#openai_gpt_5_6_terra_responses_profile"
+    )
+    assert captured["client_options"] == [
+        {"timeout": pytest.approx(20.0, abs=0.2), "max_retries": 0}
+    ]
 
 
 def test_llm_interface_openai_generate_omits_temperature_for_gpt5_mini(
diff --git a/tests/backend/test_mcp_create_concepts_parent_recovery.py b/tests/backend/test_mcp_create_concepts_parent_recovery.py
index 8e4df8a69..15a34ef4b 100644
--- a/tests/backend/test_mcp_create_concepts_parent_recovery.py
+++ b/tests/backend/test_mcp_create_concepts_parent_recovery.py
@@ -93,6 +93,40 @@ def test_create_concepts_parent_not_found_includes_recovery_details(monkeypatch)
     assert any("workflow supertype" in str(item) for item in suggestions)
 
 
+def test_create_concepts_rejects_individual_as_semantic_parent(monkeypatch):
+    created: list[dict[str, Any]] = []
+    monkeypatch.setattr(
+        "src.backend.db.repositories.concepts_repository.ConceptsRepository.find_one",
+        lambda query: (
+            {
+                "concept_id": "#V#example_research_lab",
+                "kind": "individual",
+            }
+            if query.get("concept_id") == "#V#example_research_lab"
+            else None
+        ),
+    )
+    monkeypatch.setattr(
+        "src.backend.vontology.utils_vontology.create_vontology_concept",
+        lambda **kwargs: created.append(dict(kwargs)) or {"success": True},
+    )
+
+    result = _create_concepts(
+        parent_id="#V#example_research_lab",
+        concepts=[{"name": "Synthetic scenario", "kind": "individual"}],
+    )
+
+    assert result.get("success") is False
+    assert result.get("error_code") == "parent_is_not_a_type"
+    assert result.get("error_details") == {
+        "original_parent_id": "#V#example_research_lab",
+        "resolved_parent_id": "#V#example_research_lab",
+        "resolved_parent_kind": "individual",
+        "requested_child_kinds": ["individual"],
+    }
+    assert created == []
+
+
 def test_create_concepts_recovery_works_through_gateway(monkeypatch):
     _patch_create_concept_success(monkeypatch)
     monkeypatch.setattr(
@@ -161,4 +195,3 @@ async def _invoke() -> Any:
     assert (payload.get("results") or [{}])[0].get("concept_id") == "#V#stdio_workflow_recovery_type"
     assert payload.get("parent_id_used") == "#V#durable_workflow"
     assert (payload.get("parent_resolution") or {}).get("fallback_used") is True
-
diff --git a/tests/backend/test_model_parameter_service.py b/tests/backend/test_model_parameter_service.py
index ce0ade0b9..349570f76 100644
--- a/tests/backend/test_model_parameter_service.py
+++ b/tests/backend/test_model_parameter_service.py
@@ -92,3 +92,30 @@ def registry_lookup(**_kwargs):
         {"reasoning_effort": "future"},
         model="gpt-5.5",
     ) == {"reasoning_effort": "future"}
+
+
+def test_provider_mapping_scopes_registry_policy_to_selected_profile(
+    monkeypatch,
+) -> None:
+    import src.backend.services.model_parameter_service as mod
+
+    profile_ids: list[str | None] = []
+
+    def registry_lookup(**kwargs):
+        profile_ids.append(kwargs.get("profile_concept_id"))
+        return {
+            "action": "allow",
+            "allowed_values": ["low"],
+            "fixed_value": None,
+            "source": "vontology_graph",
+        }
+
+    monkeypatch.setattr(mod, "_registry_parameter_policy", registry_lookup)
+
+    assert mod.openai_responses_kwargs_from_model_parameters(
+        {"reasoning_effort": "low"},
+        model="gpt-5.6-terra",
+        profile_concept_id="#V#selected_responses_profile",
+    ) == {"reasoning": {"effort": "low"}}
+    assert profile_ids
+    assert set(profile_ids) == {"#V#selected_responses_profile"}
diff --git a/tests/backend/test_model_registry_service.py b/tests/backend/test_model_registry_service.py
index e1fdbe907..55a8042bb 100644
--- a/tests/backend/test_model_registry_service.py
+++ b/tests/backend/test_model_registry_service.py
@@ -288,6 +288,141 @@ def test_runtime_parameter_policy_uses_graph_registry_snapshot(monkeypatch) -> N
     )
 
 
+def test_runtime_parameter_policy_is_scoped_to_selected_profile(monkeypatch) -> None:
+    import src.backend.services.model_registry_service as mod
+
+    snapshot = {
+        "source": "vontology_graph",
+        "models": [
+            {
+                "model_id": "same-surface-deployment",
+                "provider": "openai",
+                "registry_entry_id": "#V#same_surface_entry",
+                "api_profiles": [
+                    {
+                        "profile_concept_id": "#V#connection_a_responses",
+                        "api_surface": "responses",
+                        "parameter_constraints": [
+                            {
+                                "parameter": "temperature",
+                                "action": "omit",
+                            }
+                        ],
+                    },
+                    {
+                        "profile_concept_id": "#V#connection_b_responses",
+                        "api_surface": "responses",
+                        "parameter_constraints": [
+                            {
+                                "parameter": "temperature",
+                                "action": "fixed_value",
+                                "fixed_value": "0.25",
+                            }
+                        ],
+                    },
+                ],
+            }
+        ],
+    }
+    monkeypatch.setattr(
+        mod,
+        "get_model_registry_snapshot",
+        lambda *, preferred_language=None: snapshot,
+    )
+
+    assert (
+        mod.sanitise_model_parameter_value(
+            model="same-surface-deployment",
+            provider="openai",
+            parameter="temperature",
+            value=0.7,
+            api_surface="responses",
+            profile_concept_id="#V#connection_a_responses",
+        )
+        is None
+    )
+    assert (
+        mod.sanitise_model_parameter_value(
+            model="same-surface-deployment",
+            provider="openai",
+            parameter="temperature",
+            value=0.7,
+            api_surface="responses",
+            profile_concept_id="#V#connection_b_responses",
+        )
+        == 0.25
+    )
+    assert (
+        mod.resolve_model_parameter_policy(
+            model="same-surface-deployment",
+            provider="openai",
+            parameter="temperature",
+            api_surface="responses",
+            profile_concept_id="#V#connection_b_responses",
+        )["profile_concept_id"]
+        == "#V#connection_b_responses"
+    )
+
+
+def test_runtime_parameter_policy_prefers_exact_entry_over_family(monkeypatch) -> None:
+    import src.backend.services.model_registry_service as mod
+
+    snapshot = {
+        "source": "vontology_graph",
+        "models": [
+            {
+                "model_id": "gpt-5",
+                "provider": "openai",
+                "registry_entry_id": "#V#gpt_5_family_entry",
+                "api_profiles": [
+                    {
+                        "profile_concept_id": "#V#gpt_5_family_responses",
+                        "api_surface": "responses",
+                        "parameter_constraints": [
+                            {"parameter": "temperature", "action": "omit"}
+                        ],
+                    }
+                ],
+            },
+            {
+                "model_id": "gpt-5.6-terra",
+                "provider": "openai",
+                "registry_entry_id": "#V#terra_exact_entry",
+                "api_profiles": [
+                    {
+                        "profile_concept_id": "#V#terra_exact_responses",
+                        "api_surface": "responses",
+                        "parameter_constraints": [
+                            {
+                                "parameter": "temperature",
+                                "action": "fixed_value",
+                                "fixed_value": "0.4",
+                            }
+                        ],
+                    }
+                ],
+            },
+        ],
+    }
+    monkeypatch.setattr(
+        mod,
+        "get_model_registry_snapshot",
+        lambda *, preferred_language=None: snapshot,
+    )
+
+    policy = mod.resolve_model_parameter_policy(
+        model="gpt-5.6-terra",
+        provider="openai",
+        parameter="temperature",
+        api_surface="responses",
+        profile_concept_id="#V#terra_exact_responses",
+    )
+
+    assert policy is not None
+    assert policy["registry_entry_id"] == "#V#terra_exact_entry"
+    assert policy["action"] == "fixed_value"
+
+
 def test_runtime_parameter_policy_rejects_values_outside_graph_allowed_set(
     monkeypatch,
 ) -> None:
diff --git a/tests/backend/test_openai_structured_tool_transport.py b/tests/backend/test_openai_structured_tool_transport.py
index 2970be5a5..69757878e 100644
--- a/tests/backend/test_openai_structured_tool_transport.py
+++ b/tests/backend/test_openai_structured_tool_transport.py
@@ -2380,9 +2380,11 @@ def test_malformed_structured_transport_profile_fails_closed_before_request(
 
 
 @pytest.mark.parametrize("unsupported_first", [False, True])
+@pytest.mark.parametrize("tools_present", [False, True], ids=["answer_only", "tools"])
 def test_conflicting_same_surface_capabilities_fail_closed_before_request(
     monkeypatch: pytest.MonkeyPatch,
     unsupported_first: bool,
+    tools_present: bool,
 ) -> None:
     supported = _responses_profile(capability="supported")
     unsupported = _responses_profile(capability="unsupported")
@@ -2403,8 +2405,8 @@ def test_conflicting_same_surface_capabilities_fail_closed_before_request(
     with pytest.raises(UnsupportedStructuredToolTransportError) as exc_info:
         asyncio.run(
             client.generate_with_tools(
-                prompt="Use the tool.",
-                available_tools=[_tool()],
+                prompt="Answer or use the tool.",
+                available_tools=[_tool()] if tools_present else [],
             )
         )
 
@@ -2488,17 +2490,240 @@ def test_legacy_profile_without_structured_capability_remains_conservative(
     )
 
 
-def test_no_tool_request_preserves_chat_surface_even_with_responses_profile(
+def test_no_tool_request_preserves_required_responses_profile_and_policy(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
-    model = "responses-profile-without-tools"
-    _install_profiles(monkeypatch, _registry_profiles(_responses_profile()))
+    from src.backend.services import model_registry_service
+
+    model = "gpt-5.6-terra"
+    responses_profile = _responses_profile()
+    responses_profile["profile_concept_id"] = (
+        "#V#openai_gpt_5_6_terra_responses_profile"
+    )
+    responses_profile["parameter_constraints"] = [
+        {
+            "constraint_concept_id": "#V#terra_responses_temperature_omit",
+            "parameter_concept_id": "#V#temperature_parameter",
+            "parameter": "temperature",
+            "action": "omit",
+        }
+    ]
+    chat_profile = _chat_profile(capability="unsupported")
+    _install_profiles(
+        monkeypatch,
+        _registry_profiles(responses_profile, chat_profile),
+    )
+    monkeypatch.setattr(
+        model_registry_service,
+        "get_model_registry_snapshot",
+        lambda *, preferred_language=None: {
+            "source": "vontology_graph",
+            "registry_concept_id": "#V#default_model_registry",
+            "models": [
+                {
+                    "model_id": model,
+                    "provider": "openai",
+                    "concept_id": "#V#openai_gpt_5_6_terra",
+                    "registry_entry_id": "#V#terra_registry_entry",
+                    "api_profiles": [responses_profile, chat_profile],
+                }
+            ],
+        },
+    )
+    captured = _install_fake_openai(
+        monkeypatch,
+        responses=[_text_response(model=model, text="grounded answer")],
+    )
+    client = OpenAIClient(
+        LLMClientConfig(
+            model=model,
+            provider="openai",
+            api_key="test-key",
+            temperature=0.7,
+        )
+    )
+
+    result = asyncio.run(
+        client.generate_with_tools(
+            prompt="Answer without tools",
+            available_tools=[],
+            temperature=0.7,
+        )
+    )
+
+    assert captured["chat"] == []
+    assert len(captured["responses"]) == 1
+    assert "temperature" not in captured["responses"][0]
+    assert result.text_response == "grounded answer"
+    assert result.transport_metadata["tools_present"] is False
+    assert result.transport_metadata["effective_api_surface"] == "responses"
+    assert result.transport_metadata["profile_concept_id"] == (
+        "#V#openai_gpt_5_6_terra_responses_profile"
+    )
+    assert result.transport_metadata["reason"] == "represented_profile_required"
+    assert result.transport_metadata["requested_model_parameters"] == {
+        "temperature": 0.7
+    }
+    assert result.transport_metadata["effective_provider_parameters"] == {}
+    assert result.transport_metadata["omitted_model_parameter_names"] == [
+        "temperature"
+    ]
+
+
+@pytest.mark.parametrize(
+    ("connection_id", "expected_temperature"),
+    [
+        ("#V#connection_a", None),
+        ("#V#connection_b", 0.25),
+    ],
+)
+def test_temperature_policy_follows_exact_selected_connection_profile(
+    monkeypatch: pytest.MonkeyPatch,
+    connection_id: str,
+    expected_temperature: float | None,
+) -> None:
+    from src.backend.services import model_registry_service
+
+    model = "same-surface-connection-model"
+    connection_a = _responses_profile()
+    connection_a.update(
+        {
+            "profile_concept_id": "#V#connection_a_responses",
+            "connection_id": "#V#connection_a",
+            "parameter_constraints": [
+                {"parameter": "temperature", "action": "omit"}
+            ],
+        }
+    )
+    connection_b = _responses_profile()
+    connection_b.update(
+        {
+            "profile_concept_id": "#V#connection_b_responses",
+            "connection_id": "#V#connection_b",
+            "parameter_constraints": [
+                {
+                    "parameter": "temperature",
+                    "action": "fixed_value",
+                    "fixed_value": "0.25",
+                }
+            ],
+        }
+    )
+    registry = _registry_profiles(connection_a, connection_b)
+    _install_profiles(monkeypatch, registry)
+    monkeypatch.setattr(
+        model_registry_service,
+        "get_model_registry_snapshot",
+        lambda *, preferred_language=None: {
+            "source": "vontology_graph",
+            "models": [
+                {
+                    "model_id": model,
+                    "provider": "openai",
+                    "registry_entry_id": "#V#same_surface_connection_entry",
+                    "api_profiles": [connection_a, connection_b],
+                }
+            ],
+        },
+    )
+    captured = _install_fake_openai(
+        monkeypatch,
+        responses=[_text_response(model=model)],
+    )
+    client = OpenAIClient(
+        LLMClientConfig(
+            model=model,
+            provider="openai",
+            api_key="test-key",
+            connection_id=connection_id,
+            temperature=0.7,
+        )
+    )
+
+    result = asyncio.run(
+        client.generate_with_tools(
+            prompt="Answer without tools",
+            available_tools=[],
+        )
+    )
+
+    request = captured["responses"][0]
+    if expected_temperature is None:
+        assert "temperature" not in request
+        assert result.transport_metadata["effective_provider_parameters"] == {}
+    else:
+        assert request["temperature"] == expected_temperature
+        assert result.transport_metadata["effective_provider_parameters"] == {
+            "temperature": expected_temperature
+        }
+    assert result.transport_metadata["profile_concept_id"] == (
+        f"#V#{connection_id.removeprefix('#V#')}_responses"
+    )
+
+
+@pytest.mark.parametrize(
+    ("connection_id", "expected_effort"),
+    [
+        ("#V#connection_a", None),
+        ("#V#connection_b", "medium"),
+    ],
+)
+def test_answer_only_chat_direct_reasoning_uses_exact_connection_profile(
+    monkeypatch: pytest.MonkeyPatch,
+    connection_id: str,
+    expected_effort: str | None,
+) -> None:
+    from src.backend.services import model_registry_service
+
+    model = "gpt-5-answer-only-chat"
+    connection_a = _chat_profile(capability="unsupported")
+    connection_a.update(
+        {
+            "profile_concept_id": "#V#connection_a_chat",
+            "connection_id": "#V#connection_a",
+            "parameter_constraints": [
+                {"parameter": "reasoning_effort", "action": "omit"}
+            ],
+        }
+    )
+    connection_b = _chat_profile(capability="unsupported")
+    connection_b.update(
+        {
+            "profile_concept_id": "#V#connection_b_chat",
+            "connection_id": "#V#connection_b",
+            "parameter_constraints": [
+                {
+                    "parameter": "reasoning_effort",
+                    "action": "fixed_value",
+                    "fixed_value": "medium",
+                }
+            ],
+        }
+    )
+    profiles = [connection_a, connection_b]
+    _install_profiles(monkeypatch, _registry_profiles(*profiles))
+    monkeypatch.setattr(
+        model_registry_service,
+        "get_model_registry_snapshot",
+        lambda *, preferred_language=None: {
+            "source": "vontology_graph",
+            "models": [
+                {
+                    "model_id": model,
+                    "provider": "openai",
+                    "registry_entry_id": "#V#answer_only_chat_entry",
+                    "api_profiles": profiles,
+                }
+            ],
+        },
+    )
     captured = _install_fake_openai(monkeypatch)
     client = OpenAIClient(
         LLMClientConfig(
             model=model,
             provider="openai",
             api_key="test-key",
+            connection_id=connection_id,
             temperature=None,
         )
     )
@@ -2507,12 +2732,173 @@ def test_no_tool_request_preserves_chat_surface_even_with_responses_profile(
         client.generate_with_tools(
             prompt="Answer without tools",
             available_tools=[],
+            reasoning_effort="high",
+        )
+    )
+
+    request = captured["chat"][0]
+    if expected_effort is None:
+        assert "reasoning_effort" not in request
+        assert result.transport_metadata["effective_provider_parameters"] == {}
+    else:
+        assert request["reasoning_effort"] == expected_effort
+        assert result.transport_metadata["effective_provider_parameters"] == {
+            "reasoning_effort": expected_effort
+        }
+    assert result.transport_metadata["profile_concept_id"] == (
+        f"#V#{connection_id.removeprefix('#V#')}_chat"
+    )
+    assert result.transport_metadata["requested_model_parameters"] == {
+        "reasoning_effort": "high"
+    }
+
+
+@pytest.mark.parametrize(
+    ("action", "fixed_value", "expected_effort"),
+    [
+        ("omit", None, None),
+        ("fixed_value", "medium", "medium"),
+    ],
+)
+def test_direct_responses_reasoning_obeys_selected_profile_policy(
+    monkeypatch: pytest.MonkeyPatch,
+    action: str,
+    fixed_value: str | None,
+    expected_effort: str | None,
+) -> None:
+    from src.backend.services import model_registry_service
+
+    model = "gpt-5-direct-responses-reasoning"
+    profile = _responses_profile()
+    constraint = {
+        "parameter": "reasoning_effort",
+        "action": action,
+    }
+    if fixed_value is not None:
+        constraint["fixed_value"] = fixed_value
+    profile["parameter_constraints"] = [constraint]
+    _install_profiles(monkeypatch, _registry_profiles(profile))
+    monkeypatch.setattr(
+        model_registry_service,
+        "get_model_registry_snapshot",
+        lambda *, preferred_language=None: {
+            "source": "vontology_graph",
+            "models": [
+                {
+                    "model_id": model,
+                    "provider": "openai",
+                    "registry_entry_id": "#V#direct_responses_reasoning_entry",
+                    "api_profiles": [profile],
+                }
+            ],
+        },
+    )
+    captured = _install_fake_openai(
+        monkeypatch,
+        responses=[_text_response(model=model)],
+    )
+    client = OpenAIClient(
+        LLMClientConfig(
+            model=model,
+            provider="openai",
+            api_key="test-key",
+            temperature=None,
+        )
+    )
+
+    result = asyncio.run(
+        client.generate_with_tools(
+            prompt="Answer without tools",
+            available_tools=[],
+            reasoning={"effort": "high", "summary": "auto"},
+        )
+    )
+
+    request_reasoning = captured["responses"][0]["reasoning"]
+    if expected_effort is None:
+        assert request_reasoning == {"summary": "auto"}
+        assert result.transport_metadata["effective_provider_parameters"] == {}
+    else:
+        assert request_reasoning == {
+            "summary": "auto",
+            "effort": expected_effort,
+        }
+        assert result.transport_metadata["effective_provider_parameters"] == {
+            "reasoning": {"effort": expected_effort}
+        }
+    assert result.transport_metadata["requested_model_parameters"] == {
+        "reasoning_effort": "high"
+    }
+
+
+def test_unknown_profile_no_tool_request_preserves_conservative_chat(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    model = "unknown-text-model"
+    _install_profiles(monkeypatch, None)
+    captured = _install_fake_openai(monkeypatch)
+    client = OpenAIClient(
+        LLMClientConfig(
+            model=model,
+            provider="openai",
+            api_key="test-key",
+            temperature=None,
+        )
+    )
+
+    result = asyncio.run(
+        client.generate_with_tools(
+            prompt="Answer without tools",
+            available_tools=[],
+            temperature=0.2,
         )
     )
 
     assert len(captured["chat"]) == 1
+    assert captured["chat"][0]["temperature"] == 0.2
     assert captured["responses"] == []
-    assert result.transport_metadata["effective_api_surface"] == ("chat_completions")
+    assert result.transport_metadata["effective_api_surface"] == "chat_completions"
+    assert result.transport_metadata["capability_source"] == (
+        "conservative_client_default"
+    )
+    assert result.transport_metadata["reason"] == "no_structured_tools_in_request"
+    assert result.transport_metadata["requested_model_parameters"] == {
+        "temperature": 0.2
+    }
+    assert result.transport_metadata["effective_provider_parameters"] == {
+        "temperature": 0.2
+    }
+
+
+def test_chat_tools_unsupported_profile_still_allows_text_only_chat(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    model = "chat-text-only-model"
+    _install_profiles(
+        monkeypatch,
+        _registry_profiles(_chat_profile(capability="unsupported")),
+    )
+    captured = _install_fake_openai(monkeypatch)
+    client = OpenAIClient(
+        LLMClientConfig(
+            model=model,
+            provider="openai",
+            api_key="test-key",
+            temperature=None,
+        )
+    )
+
+    result = asyncio.run(
+        client.generate_with_tools(
+            prompt="Answer without tools",
+            available_tools=[],
+        )
+    )
+
+    assert len(captured["chat"]) == 1
+    assert captured["responses"] == []
+    assert result.text_response == "chat ok"
+    assert result.transport_metadata["effective_api_surface"] == "chat_completions"
     assert result.transport_metadata["reason"] == "no_structured_tools_in_request"
 
 
diff --git a/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py b/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py
index 20893509b..442fe0385 100644
--- a/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py
+++ b/tests/backend/test_rag_turn_execution_records_mcp_read_tools.py
@@ -1041,6 +1041,26 @@ def test_rag_get_item_supports_turn_execution_records(monkeypatch):
             {"check_id": "check_effect_2", "status": "inconclusive"}
         ],
         "critic": {"summary": {"inconclusive_count": 1}},
+        "late_effect_observations": [
+            {
+                "schema_version": "late_effect_observation.v1",
+                "observation_id": "late-observation-9",
+                "effect_id": "effect_2",
+                "execution_id": "mcp_late_9",
+                "capability_name": "add_relationship",
+                "outcome": "late_success",
+                "effect_status": "succeeded",
+                "changed": True,
+                "observed_at_utc": "2026-02-19T01:00:02Z",
+                "storage_transformed": True,
+                "payload": {
+                    "success": True,
+                    "effect_status": "succeeded",
+                    "changed": True,
+                    "private_detail": "not part of the actor receipt projection",
+                },
+            }
+        ],
     }
 
     coll = _TurnExecutionCollection([doc])
@@ -1074,6 +1094,64 @@ def test_rag_get_item_supports_turn_execution_records(monkeypatch):
         result["workflow_routing_diagnostics"]["dispatch"]["last_successful_boundary"]
         == "workflow_handoff"
     )
+    assert result["late_effect_observation_count"] == 1
+    assert result["late_effect_observations_truncated"] is False
+    assert result["late_effect_observations"] == [
+        {
+            "schema_version": "late_effect_observation.v1",
+            "observation_id": "late-observation-9",
+            "effect_id": "effect_2",
+            "execution_id": "mcp_late_9",
+            "capability_name": "add_relationship",
+            "outcome": "late_success",
+            "effect_status": "succeeded",
+            "changed": True,
+            "observed_at_utc": "2026-02-19T01:00:02Z",
+            "storage_transformed": True,
+            "receipt": {
+                "success": True,
+                "effect_status": "succeeded",
+                "changed": True,
+            },
+        }
+    ]
+
+
+def test_rag_get_item_does_not_expose_late_effects_across_namespaces(
+    monkeypatch,
+):
+    from src.backend.integrations.internal_mcp import catalogue as cat
+
+    coll = _TurnExecutionCollection(
+        [
+                {
+                    "request_id": "req-private-late-effect",
+                    "session_id": "chat-private-late-effect",
+                    "namespace": "#V#user@org_a",
+                "late_effect_observations": [
+                    {
+                        "observation_id": "private-observation",
+                        "effect_id": "private-effect",
+                        "outcome": "late_success",
+                    }
+                ],
+            }
+        ]
+    )
+    monkeypatch.setattr(
+        "src.backend.db.connection_manager.get_db",
+        lambda: _DB({"turn_execution_records": coll}),
+    )
+
+    result = cat._rag_get_item(
+        namespace="#V#user@org_b",
+        collection="turn_execution_records",
+        session_id="req-private-late-effect",
+    )
+
+    assert result["success"] is False
+    assert result["error_code"] == "not_found"
+    assert "late_effect_observations" not in result
 
 
 def test_rag_list_indexed_supports_episode_critique_improvement_suggestion_summary(
diff --git a/tests/backend/test_turn_execution_record_service_execution_summary.py b/tests/backend/test_turn_execution_record_service_execution_summary.py
index f537a72ce..9bac20dff 100644
--- a/tests/backend/test_turn_execution_record_service_execution_summary.py
+++ b/tests/backend/test_turn_execution_record_service_execution_summary.py
@@ -144,8 +144,37 @@ def test_late_effect_observation_is_bounded_idempotent_and_survives_full_upsert(
     class _Collection:
         document = None
 
-        def update_one(self, _query, update, **_kwargs):
+        def update_one(self, query, update, **_kwargs):
             existed = self.document is not None
+            if not existed and not _kwargs.get("upsert"):
+                return SimpleNamespace(
+                    modified_count=0,
+                    matched_count=0,
+                    upserted_id=None,
+                )
+            if existed and query.get("request_id") != self.document.get("request_id"):
+                return SimpleNamespace(
+                    modified_count=0,
+                    matched_count=0,
+                    upserted_id=None,
+                )
+            late_filter = query.get("late_effect_observations")
+            if existed and isinstance(late_filter, dict):
+                observation_id = (
+                    (late_filter.get("$not") or {})
+                    .get("$elemMatch", {})
+                    .get("observation_id")
+                )
+                if any(
+                    item.get("observation_id") == observation_id
+                    for item in self.document.get("late_effect_observations", [])
+                    if isinstance(item, dict)
+                ):
+                    return SimpleNamespace(
+                        modified_count=0,
+                        matched_count=0,
+                        upserted_id=None,
+                    )
             if self.document is None:
                 self.document = dict(update.get("$setOnInsert") or {})
             modified = False
@@ -159,10 +188,22 @@ def update_one(self, _query, update, **_kwargs):
                 if value not in values:
                     values.append(value)
                     modified = True
+            for key, value in (update.get("$push") or {}).items():
+                self.document.setdefault(key, []).append(value)
+                modified = True
+            for key, value in (update.get("$max") or {}).items():
+                current = self.document.get(key)
+                if current is None or current < value:
+                    self.document[key] = value
+                    modified = True
             return SimpleNamespace(
                 modified_count=1 if modified and existed else 0,
                 matched_count=1 if existed else 0,
-                upserted_id=None if existed else "late-observation-1",
+                upserted_id=(
+                    None
+                    if existed or not _kwargs.get("upsert")
+                    else "late-observation-1"
+                ),
             )
 
         def find_one(self, query, **_kwargs):
@@ -218,7 +259,14 @@ def find_one(self, query, **_kwargs):
         request_id="req-late-effect",
         effect_id="effect-late-1",
         execution_id="mcp_late_effect_1",
-        observation=late_observation,
+        observation={
+            **late_observation,
+            "observed_at_utc": "2026-07-27T10:00:02Z",
+            "payload": {
+                **late_observation["payload"],
+                "changed": False,
+            },
+        },
         user_id="#V#user",
         session_id="session-late-effect",
         namespace="#V#user@org",
@@ -238,6 +286,10 @@ def find_one(self, query, **_kwargs):
     assert len(json.dumps(stored_observation)) < 64_000
     assert "source_observation_sha256" not in stored_observation
     assert "storage_truncated" not in stored_observation
+    assert stored_observation["storage_transformed"] is True
+    assert collection.document["late_effect_updated_at_utc"] == (
+        collection.document["updated_at_utc"]
+    )
 
     full_record = build_turn_execution_record(
         request_id="req-late-effect",
@@ -294,6 +346,95 @@ def find_one(self, query, **_kwargs):
     assert readback["late_effect_observations"][0]["payload"]["changed"] is True
 
 
+def test_late_effect_observation_refuses_cross_actor_request_id_collision(
+    monkeypatch,
+) -> None:
+    class _Collection:
+        document = {
+            "request_id": "shared-client-request-id",
+            "namespace": "#V#actor_a@org",
+            "user_id": "#V#actor_a",
+            "org_id": "#V#org",
+            "late_effect_observations": [],
+        }
+
+        def update_one(self, _query, _update, **_kwargs):
+            return SimpleNamespace(
+                modified_count=0,
+                matched_count=0,
+                upserted_id=None,
+            )
+
+        def find_one(self, query, **_kwargs):
+            for field_name, expected in query.items():
+                if self.document.get(field_name) != expected:
+                    return None
+            return dict(self.document)
+
+    collection = _Collection()
+    import src.backend.services.turn_execution_record_service as record_service
+
+    monkeypatch.setattr(
+        record_service,
+        "get_turn_execution_records_collection",
+        lambda: collection,
+    )
+
+    outcome = append_late_effect_observation(
+        request_id="shared-client-request-id",
+        effect_id="effect-b",
+        execution_id="execution-b",
+        observation={
+            "outcome": "late_success",
+            "payload": {"success": True, "changed": True},
+        },
+        namespace="#V#actor_b@org",
+        user_id="#V#actor_b",
+        org_id="#V#org",
+    )
+
+    assert outcome["updated"] is False
+    assert outcome["reason"] == "actor_scope_mismatch"
+    assert collection.document["late_effect_observations"] == []
+
+
+def test_turn_record_upsert_refuses_cross_actor_request_id_collision(
+    monkeypatch,
+) -> None:
+    from pymongo.errors import DuplicateKeyError
+
+    class _Collection:
+        def update_one(self, query, _update, **_kwargs):
+            assert query == {
+                "request_id": "shared-upsert-request-id",
+                "namespace": "#V#actor_b@org",
+                "user_id": "#V#actor_b",
+                "org_id": "#V#org",
+            }
+            raise DuplicateKeyError("request_id_unique")
+
+    import src.backend.services.turn_execution_record_service as record_service
+
+    monkeypatch.setattr(
+        record_service,
+        "get_turn_execution_records_collection",
+        lambda: _Collection(),
+    )
+
+    outcome = upsert_turn_execution_record_projection(
+        record={"request_id": "shared-upsert-request-id"},
+        namespace="#V#actor_b@org",
+        user_id="#V#actor_b",
+        org_id="#V#org",
+    )
+
+    assert outcome == {
+        "updated": False,
+        "reason": "actor_scope_mismatch",
+        "request_id": "shared-upsert-request-id",
+    }
+
+
 def test_projection_field_telemetry_preserves_bounded_collection_row_index() -> None:
     entries = _normalise_projection_field_entries(
         [
diff --git a/tests/backend/test_von_chat_run_timeout.py b/tests/backend/test_von_chat_run_timeout.py
index 88346dbe8..a595e60fd 100644
--- a/tests/backend/test_von_chat_run_timeout.py
+++ b/tests/backend/test_von_chat_run_timeout.py
@@ -87,7 +87,7 @@ async def _timed_call() -> object:
     time.sleep(0.25)
 
 
-def test_restricted_gateway_forwards_absolute_read_deadline() -> None:
+def test_restricted_gateway_forwards_adaptive_read_transport_options() -> None:
     from src.backend.mcp_server.mcp_stdio_server import _RestrictedGateway
 
     captured: dict[str, object] = {}
@@ -101,22 +101,39 @@ def describe_methods(self):
         def get_method_definition(self, method_name):
             return SimpleNamespace(name=method_name, category="read")
 
-        def invoke(self, method_name, payload, *, deadline_monotonic=None):
+        def get_method_timeout_sec(self, method_name):
+            assert method_name == "safe_read"
+            return 17.0
+
+        def invoke(
+            self,
+            method_name,
+            payload,
+            *,
+            deadline_monotonic=None,
+            late_completion_observer=None,
+            require_configured_timeout=False,
+        ):
             captured.update(
                 {
                     "method_name": method_name,
                     "payload": payload,
                     "deadline_monotonic": deadline_monotonic,
+                    "late_completion_observer": late_completion_observer,
+                    "require_configured_timeout": require_configured_timeout,
                 }
             )
             return "result"
 
     gateway = _RestrictedGateway(gateway=_BaseGateway(), allow_writes=False)
+    assert gateway.get_method_timeout_sec("safe_read") == 17.0
 
     result = gateway.invoke(
         "safe_read",
         {"query": "bounded"},
         deadline_monotonic=123.5,
+        late_completion_observer=None,
+        require_configured_timeout=False,
     )
 
     assert result == "result"
@@ -124,6 +141,8 @@ def invoke(self, method_name, payload, *, deadline_monotonic=None):
         "method_name": "safe_read",
         "payload": {"query": "bounded"},
         "deadline_monotonic": 123.5,
+        "late_completion_observer": None,
+        "require_configured_timeout": False,
     }