Skip to content

JVNAUTOSCI-2600 A3r4: preserve late effect finality #311

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
witbrock wants to merge 1 commit into from
Closed

JVNAUTOSCI-2600 A3r4: preserve late effect finality #311

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
175 changes: 92 additions & 83 deletions src/backend/integrations/internal_mcp/catalogue.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -877,6 +877,9 @@ def _create_concepts(**kwargs):
from ...services.create_concepts_parent_resolution_service import (
resolve_parent_for_create_concepts,
)
from ...services.relationship_extent_index_service import (
defer_relationship_extent_index_sync,
)

parent_id: str | None = kwargs.get("parent_id")
concepts = kwargs.get("concepts", [])
Expand Down Expand Up @@ -1092,95 +1095,101 @@ def _create_concepts(**kwargs):
)

results = []
for concept_data in concepts:
# This is both the batch boundary and the safe cancellation point after
# the preceding concept's complete logical write bundle.
raise_if_internal_mcp_cancelled()
if not isinstance(concept_data, dict):
results.append({"error": "Concept must be an object", "data": concept_data})
continue
with defer_relationship_extent_index_sync():
for concept_data in concepts:
# This is both the batch boundary and the safe cancellation point after
# the preceding concept's complete logical write bundle.
raise_if_internal_mcp_cancelled()
if not isinstance(concept_data, dict):
results.append(
{"error": "Concept must be an object", "data": concept_data}
)
continue

name = concept_data.get("name")
kind = (concept_data.get("kind") or "type").strip().lower()
name = concept_data.get("name")
kind = (concept_data.get("kind") or "type").strip().lower()

if not name:
results.append(
{"error": "Concept missing required 'name' field", "data": concept_data}
)
continue
if not name:
results.append(
{
"error": "Concept missing required 'name' field",
"data": concept_data,
}
)
continue

# Map kind to create_as_instance parameter
if kind == "individual":
kind = "instance"
# Map kind to create_as_instance parameter
if kind == "individual":
kind = "instance"

if kind == "predicate":
create_as_instance = True
parent_id_for_concept = PREDICATE_TYPE_ID
else:
create_as_instance = kind == "instance"
parent_id_for_concept = validated_parent_id

duplicate_match = find_existing_concept_for_create_concepts(
concept_name=str(name),
kind=kind,
parent_id_for_concept=parent_id_for_concept,
preferred_language="en-NZ",
allow_duplicate_instances=allow_duplicate_instances,
duplicate_resolution_mode=(
_CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY
if canonical_id_only
else None
),
)
if duplicate_match is not None:
result = build_duplicate_prevented_create_concepts_result(
requested_name=str(name),
requested_kind=kind,
existing_concept_id=duplicate_match.existing_concept_id,
guard_scope=duplicate_match.guard_scope,
match_source=duplicate_match.match_source,
if kind == "predicate":
create_as_instance = True
parent_id_for_concept = PREDICATE_TYPE_ID
else:
create_as_instance = kind == "instance"
parent_id_for_concept = validated_parent_id

duplicate_match = find_existing_concept_for_create_concepts(
concept_name=str(name),
kind=kind,
parent_id_for_concept=parent_id_for_concept,
preferred_language="en-NZ",
allow_duplicate_instances=allow_duplicate_instances,
duplicate_resolution_mode=(
_CREATE_CONCEPTS_DUPLICATE_RESOLUTION_CANONICAL_ID_ONLY
if canonical_id_only
else None
),
)
result["concept_id"] = duplicate_match.existing_concept_id
results.append(result)
continue
if duplicate_match is not None:
result = build_duplicate_prevented_create_concepts_result(
requested_name=str(name),
requested_kind=kind,
existing_concept_id=duplicate_match.existing_concept_id,
guard_scope=duplicate_match.guard_scope,
match_source=duplicate_match.match_source,
)
result["concept_id"] = duplicate_match.existing_concept_id
results.append(result)
continue

# Duplicate preflights are read-only and may consume the entire
# transport budget. Never start an insert after cancellation.
raise_if_internal_mcp_cancelled()
result = create_vontology_concept(
parent_id=parent_id_for_concept,
new_concept_name=name,
create_as_instance=create_as_instance,
description=concept_data.get("description"),
notes=concept_data.get("notes"),
created_by_concept_id=actor_user_id,
organisation_concept_id=actor_org_id,
event_namespace=namespace,
visibility_scope_mode=scope_mode,
)
# Enrich result with the requested name for traceability and surface
# the canonical created concept_id at a stable top-level key so UI
# summaries can reliably name what was created.
result["requested_name"] = name
result["requested_kind"] = kind
concept_id_value = result.get("concept_id")
if not isinstance(concept_id_value, str) or not concept_id_value.strip():
nested_concept = result.get("concept")
if isinstance(nested_concept, dict):
nested_id = nested_concept.get("concept_id")
if isinstance(nested_id, str) and nested_id.strip():
concept_id_value = nested_id
if not isinstance(concept_id_value, str) or not concept_id_value.strip():
canonical_id = result.get("canonical_concept_id")
if isinstance(canonical_id, str) and canonical_id.strip():
concept_id_value = canonical_id
if not isinstance(concept_id_value, str) or not concept_id_value.strip():
existing_id = result.get("existing_concept_id")
if isinstance(existing_id, str) and existing_id.strip():
concept_id_value = existing_id
if isinstance(concept_id_value, str) and concept_id_value.strip():
result["concept_id"] = concept_id_value.strip()
results.append(result)
# Duplicate preflights are read-only and may consume the entire
# transport budget. Never start an insert after cancellation.
raise_if_internal_mcp_cancelled()
result = create_vontology_concept(
parent_id=parent_id_for_concept,
new_concept_name=name,
create_as_instance=create_as_instance,
description=concept_data.get("description"),
notes=concept_data.get("notes"),
created_by_concept_id=actor_user_id,
organisation_concept_id=actor_org_id,
event_namespace=namespace,
visibility_scope_mode=scope_mode,
)
# Enrich result with the requested name for traceability and surface
# the canonical created concept_id at a stable top-level key so UI
# summaries can reliably name what was created.
result["requested_name"] = name
result["requested_kind"] = kind
concept_id_value = result.get("concept_id")
if not isinstance(concept_id_value, str) or not concept_id_value.strip():
nested_concept = result.get("concept")
if isinstance(nested_concept, dict):
nested_id = nested_concept.get("concept_id")
if isinstance(nested_id, str) and nested_id.strip():
concept_id_value = nested_id
if not isinstance(concept_id_value, str) or not concept_id_value.strip():
canonical_id = result.get("canonical_concept_id")
if isinstance(canonical_id, str) and canonical_id.strip():
concept_id_value = canonical_id
if not isinstance(concept_id_value, str) or not concept_id_value.strip():
existing_id = result.get("existing_concept_id")
if isinstance(existing_id, str) and existing_id.strip():
concept_id_value = existing_id
if isinstance(concept_id_value, str) and concept_id_value.strip():
result["concept_id"] = concept_id_value.strip()
results.append(result)

# Count different outcome types for summary
successful = sum(1 for r in results if isinstance(r, dict) and r.get("success"))
Expand Down
64 changes: 63 additions & 1 deletion src/backend/integrations/internal_mcp/gateway.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,11 @@
normalise_payload_aliases,
validate_payload,
)
from .transport import InternalMCPTransport, TransportResult
from .transport import (
InternalMCPTransport,
LateCompletionObserver,
TransportResult,
)

logger = logging.getLogger(__name__)

Expand Down Expand Up @@ -339,6 +343,7 @@ def invoke(
payload: Optional[MutableMapping[str, Any]] = None,
*,
deadline_monotonic: float | None = None,
late_completion_observer: LateCompletionObserver | None = None,
) -> TransportResult:
if not self._enabled:
raise GatewayDisabledError("Internal MCP gateway is disabled.")
Expand Down Expand Up @@ -382,6 +387,62 @@ def invoke(
definition.category,
hard_timeout_sec=float(timeout or self._transport.read_timeout_sec),
)
observed_late_completion = late_completion_observer
if late_completion_observer is not None:

def _observe_validated_late_completion(
observation: Dict[str, Any],
) -> None:
enriched = dict(observation)
result_payload = enriched.get("payload")
if enriched.get("outcome") != "late_success":
validation = "handler_error"
valid: bool | None = None
validation_error = None
elif enriched.get("payload_truncated") is True:
validation = "indeterminate_truncated"
valid = None
validation_error = "Late handler payload was truncated."
elif definition.output_schema is None:
validation = "not_configured"
valid = None
validation_error = None
elif not isinstance(result_payload, MutableMapping):
validation = "invalid"
valid = False
validation_error = (
"Output schema provided but late handler returned "
"a non-mapping payload."
)
elif result_payload.get("success") is False:
validation = "standard_error_response"
valid = True
validation_error = None
else:
try:
valid, validation_errors = validate_payload(
definition.output_schema,
result_payload,
)
except Exception as exc:
valid = False
validation_errors = [type(exc).__name__]
validation = "valid" if valid else "invalid"
validation_error = (
None
if valid
else "; ".join(validation_errors)[:2_000]
)
enriched.update(
{
"output_schema_validation": validation,
"output_schema_valid": valid,
"output_schema_error": validation_error,
}
)
late_completion_observer(enriched)

observed_late_completion = _observe_validated_late_completion
try:
from src.backend.security.access_control import (
get_effective_organisation_concept_id,
Expand Down Expand Up @@ -460,6 +521,7 @@ def invoke(
advisory_timeout_sec=advisory_timeout,
deadline_monotonic=deadline_monotonic,
log_tag=self._log_tag,
late_completion_observer=observed_late_completion,
)
finally:
_PREEXISTING_ACTOR_CONTEXT.reset(preexisting_actor_token)
Expand Down
Loading