diff --git a/src/backend/db/mongo_client.py b/src/backend/db/mongo_client.py index 8d8ff16f..9ab0931f 100644 --- a/src/backend/db/mongo_client.py +++ b/src/backend/db/mongo_client.py @@ -772,6 +772,9 @@ def assert_destructive_db_operation_allowed(operation: str) -> None: # Server-side persistence for prompts waiting behind active chat turns. CHAT_PROMPT_QUEUE_COLLECTION_NAME = "chat_prompt_queue" +# Restart-safe, actor-bound per-window organisation selections. +WINDOW_SESSION_BINDINGS_COLLECTION_NAME = "window_session_bindings" + # --- Client Initialization --- # REFACTORING_NOTE: We maintain separate clients for real MongoDB vs mongomock. # This prevents test suites from “poisoning” the process by enabling VON_USE_MOCK_DB @@ -2107,6 +2110,23 @@ def _ensure_chat_prompt_queue_indexes(coll: Collection) -> None: "active_conversation_key": {"$exists": True, "$type": "string"} }, ) + if "active_legacy_submission_key_unique" not in existing_indexes: + coll.create_index( + [("active_legacy_submission_key", ASCENDING)], + name="active_legacy_submission_key_unique", + unique=True, + partialFilterExpression={ + "active_legacy_submission_key": { + "$exists": True, + "$type": "string", + } + }, + ) + if "legacy_submission_expires_at" not in existing_indexes: + coll.create_index( + [("legacy_submission_expires_at", ASCENDING)], + name="legacy_submission_expires_at", + ) if "queued_global_slot_unique" not in existing_indexes: coll.create_index( [("queued_global_slot", ASCENDING)], @@ -2153,6 +2173,18 @@ def _ensure_chat_prompt_queue_indexes(coll: Collection) -> None: coll.create_index([("updated_at", DESCENDING)], name="updated_at_-1") +def _ensure_window_session_binding_indexes(coll: Collection) -> None: + existing_indexes = {idx["name"] for idx in coll.list_indexes()} + if "user_id_1" not in existing_indexes: + coll.create_index([("user_id", ASCENDING)], name="user_id_1") + if "expires_at_ttl" not in existing_indexes: + coll.create_index( + [("expires_at", ASCENDING)], + name="expires_at_ttl", + expireAfterSeconds=0, + ) + + def _ensure_gmail_outbound_quota_indexes(coll: Collection) -> None: existing_indexes = {idx["name"] for idx in coll.list_indexes()} if "expires_at_ttl" not in existing_indexes: @@ -2453,6 +2485,19 @@ def get_chat_prompt_queue_collection() -> Collection | None: return None +def get_window_session_binding_collection() -> Collection | None: + """Return durable actor-owned browser-window organisation selections.""" + + db = get_db() + if db is not None: + return _ensure_collection_indexes_once( + db, + WINDOW_SESSION_BINDINGS_COLLECTION_NAME, + _ensure_window_session_binding_indexes, + ) + return None + + def get_gmail_outbound_quota_collection() -> Collection | None: """Return atomic outbound Gmail counters with bounded TTL retention.""" diff --git a/src/backend/security/access_control.py b/src/backend/security/access_control.py index 2bb6de23..06595362 100644 --- a/src/backend/security/access_control.py +++ b/src/backend/security/access_control.py @@ -457,7 +457,10 @@ def get_effective_organisation_concept_id() -> Optional[str]: try: window_session_id = request.headers.get("X-Von-Window-Session") if window_session_id: - from ..services.window_session_context_service import get_effective_context + from ..services.window_session_context_service import ( + WindowSessionContextUnavailable, + get_effective_context, + ) user_for_context = get_effective_user_concept_id() or session.get( "user_concept_id" @@ -466,10 +469,15 @@ def get_effective_organisation_concept_id() -> Optional[str]: window_session_id, dict(session), user_for_context, + require_known_window=True, ) org_id = _normalise_stored_org_concept_id(effective.get("organisation_id")) if org_id: return org_id + except WindowSessionContextUnavailable: + # An explicit unknown/expired/other-actor selector must never inherit + # the browser-wide organisation selected by another tab. + return None except Exception: pass for key in ("organisation_concept_id", "org_concept_id", "org_id"): diff --git a/src/backend/server/routes/auth_routes.py b/src/backend/server/routes/auth_routes.py index 6f912be4..1d5c417a 100644 --- a/src/backend/server/routes/auth_routes.py +++ b/src/backend/server/routes/auth_routes.py @@ -1,4 +1,4 @@ -from flask import Blueprint, request, redirect, session, url_for, jsonify +from flask import Blueprint, current_app, request, redirect, session, url_for, jsonify from ...auth_service import GoogleAuthService from ...services.settings_service import ( @@ -539,18 +539,39 @@ def logout(): from ...services.window_session_context_service import ( delete_window_context_if_owned, ) - - delete_window_context_if_owned( - request.headers.get("X-Von-Window-Session"), - user_id, + from ...services.window_session_binding_store_service import ( + WindowSessionBindingStoreUnavailable, ) + cleanup_deferred = False + try: + delete_window_context_if_owned( + request.headers.get("X-Von-Window-Session"), + user_id, + ) + except WindowSessionBindingStoreUnavailable: + # Authentication must still end if derived binding storage is down. + # The selector carries no authority without a newly authenticated + # matching actor and expires independently. + cleanup_deferred = True + current_app.logger.warning( + "Logout completed while window-session binding cleanup was unavailable" + ) + # Clear the entire session to ensure a clean logout session.clear() print(f"[auth_logout] User {user_email} logged out successfully") - return jsonify({"success": True, "message": "Logged out successfully"}) + return jsonify( + { + "success": True, + "message": "Logged out successfully", + "window_context_cleanup": ( + "deferred" if cleanup_deferred else "completed" + ), + } + ) @auth_bp.route("/test-popup") diff --git a/src/backend/server/routes/von_routes.py b/src/backend/server/routes/von_routes.py index 6ef65b27..7dedf2ed 100644 --- a/src/backend/server/routes/von_routes.py +++ b/src/backend/server/routes/von_routes.py @@ -88,10 +88,18 @@ ) from ...services.window_session_context_service import ( WindowSessionContextUnavailable, + WindowSessionContextRecoveryUnavailable, WindowSessionOwnershipError, set_window_organisation, clear_window_organisation, get_effective_context, + set_window_chat_session, +) +from ...services.window_session_binding_store_service import ( + WindowSessionBindingStoreUnavailable, +) +from ...services.ontology_publication_authority_service import ( + OntologyMutationResourceBusy, ) from ...services.settings_service import ( get_show_tool_use_during_thinking, @@ -540,6 +548,7 @@ def _safe_app_log(level: str, message: str, *args: object) -> None: _TOOL_PROGRESS_WINDOW_SCOPE_PREFIX = "anon:window:" _TOOL_PROGRESS_SESSION_SCOPE_PREFIX = "anon:session:" _WINDOW_SESSION_HEADER_NAME = "X-Von-Window-Session" +_LEGACY_SUBMISSION_WINDOW_SESSION_KEY = "_von_legacy_submission_window_session_id" def _now_utc_iso() -> str: @@ -757,6 +766,147 @@ def _append_selected_workflow_execution_event( return payload +def _recover_window_context_from_owned_conversation( + *, + window_session_id: str | None, + user_concept_id: str | None, + conversation_session_id: str | None, +) -> bool: + """Recover an old tab from its exact actor-owned conversation carrier. + + This compatibility path never accepts a client-supplied organisation. The + session id selects only metadata owned by the authenticated actor, current + membership is rechecked, and the resulting binding is persisted before it + becomes usable. + """ + + clean_window_id = _normalise_non_empty_text(window_session_id) + clean_user_id = _normalise_concept_id(user_concept_id) + clean_session_id = _normalise_non_empty_text(conversation_session_id) + if not clean_window_id or not clean_user_id or not clean_session_id: + return False + try: + summary = chat_history_service.get_chat_history_session_summary( + clean_user_id, + clean_session_id, + namespace=None, + include_legacy=True, + summary_mode="light", + ) + except Exception as exc: + current_app.logger.warning( + "Owned-conversation window recovery metadata read failed: %s", + type(exc).__name__, + ) + return False + if not isinstance(summary, Mapping): + return False + + represented_namespace = _progress_str(summary.get("namespace")) + represented_org_id = _normalise_concept_id(summary.get("organisation_concept_id")) + expected_namespace = _derive_namespace_for_user_org( + clean_user_id, + represented_org_id, + ) + if represented_namespace and represented_namespace != expected_namespace: + current_app.logger.warning( + "Owned-conversation window recovery rejected inconsistent namespace" + ) + return False + if not represented_org_id and represented_namespace != expected_namespace: + # Absence of both org and namespace in legacy metadata is ambiguous, + # not evidence that this was a Personal conversation. + return False + + try: + from ...services.ontology_authority_membership_coordination_service import ( + organisation_membership_scope_barrier, + ) + + if represented_org_id: + scope_barrier = organisation_membership_scope_barrier( + clean_user_id, + represented_org_id, + ) + else: + from contextlib import nullcontext + + scope_barrier = nullcontext() + with scope_barrier: + if represented_org_id: + from ...services.organisation_membership_service import ( + resolve_user_organisation_membership, + ) + + membership = resolve_user_organisation_membership( + clean_user_id, + represented_org_id, + ) + if not isinstance(membership, Mapping): + return False + role = str(membership.get("role") or "member").strip() or "member" + set_window_organisation( + window_session_id=clean_window_id, + organisation_concept_id=represented_org_id, + role_in_org=role, + namespace=expected_namespace, + user_id=clean_user_id, + ) + else: + clear_window_organisation( + clean_window_id, + expected_namespace, + clean_user_id, + ) + set_window_chat_session( + clean_window_id, + clean_session_id, + clean_user_id, + ) + except WindowSessionBindingStoreUnavailable as exc: + raise WindowSessionContextRecoveryUnavailable( + "window_session_context_recovery_unavailable" + ) from exc + except OntologyMutationResourceBusy as exc: + raise WindowSessionContextRecoveryUnavailable( + "window_session_scope_coordination_busy" + ) from exc + except WindowSessionOwnershipError: + return False + return True + + +def _get_effective_context_with_owned_conversation_recovery( + *, + window_session_id: str | None, + flask_session_snapshot: Mapping[str, Any], + user_concept_id: str | None, + conversation_session_id: str | None, +) -> dict[str, Any]: + try: + return get_effective_context( + window_session_id, + dict(flask_session_snapshot), + user_concept_id, + require_known_window=bool(window_session_id), + ) + except WindowSessionContextRecoveryUnavailable: + raise + except WindowSessionContextUnavailable: + if not _recover_window_context_from_owned_conversation( + window_session_id=window_session_id, + user_concept_id=user_concept_id, + conversation_session_id=conversation_session_id, + ): + raise + return get_effective_context( + window_session_id, + dict(flask_session_snapshot), + user_concept_id, + require_known_window=True, + ) + + def _get_current_chat_prompt_queue_scope() -> dict[str, str | None] | tuple[Any, int]: try: from ...security.access_control import ( @@ -784,11 +934,16 @@ def _get_current_chat_prompt_queue_scope() -> dict[str, str | None] | tuple[Any, window_session_id = request.headers.get(_WINDOW_SESSION_HEADER_NAME) try: - effective_context = get_effective_context( - window_session_id, - dict(session), - user_concept_id.strip(), - require_known_window=bool(window_session_id), + queue_payload = _chat_prompt_queue_payload() + effective_context = _get_effective_context_with_owned_conversation_recovery( + window_session_id=window_session_id, + flask_session_snapshot=dict(session), + user_concept_id=user_concept_id.strip(), + conversation_session_id=( + queue_payload.get("session_id") + if isinstance(queue_payload.get("session_id"), str) + else None + ), ) except WindowSessionContextUnavailable: return ( @@ -993,6 +1148,17 @@ def create_chat_prompt_queue_route(): client_request_id=payload.get("client_request_id"), attempt_id=payload.get("attempt_id"), conversation_key=conversation_key, + legacy_submission_role=( + chat_prompt_queue_service.LEGACY_SUBMISSION_ROLE_QUEUE + if ( + _normalise_non_empty_text(payload.get("status")) + == chat_prompt_queue_service.STATUS_IN_PROGRESS + and not _normalise_non_empty_text(payload.get("client_request_id")) + and not _normalise_non_empty_text(payload.get("attempt_id")) + ) + else None + ), + window_session_id=request.headers.get(_WINDOW_SESSION_HEADER_NAME), ) return jsonify({"success": True, "item": record}), 201 except Exception as exc: @@ -10472,6 +10638,9 @@ def _submit_generate_background_request( # Do not replay the mutable browser-window selector as task authority. The # exact authenticated scope resolved at submission is frozen into the # server-side session snapshot below. + background_window_session_id = _normalise_tool_progress_window_session_id( + request_headers.get(_WINDOW_SESSION_HEADER_NAME) + ) del request_headers background_session_id = conversation_session_id @@ -10500,6 +10669,12 @@ def _submit_generate_background_request( frozen_session_snapshot["role_in_org"] = background_role else: frozen_session_snapshot.pop("role_in_org", None) + if background_window_session_id: + frozen_session_snapshot[_LEGACY_SUBMISSION_WINDOW_SESSION_KEY] = ( + background_window_session_id + ) + else: + frozen_session_snapshot.pop(_LEGACY_SUBMISSION_WINDOW_SESSION_KEY, None) def _run_generate_request_in_background() -> dict[str, Any]: with app.test_request_context( @@ -10878,6 +11053,15 @@ def generate(): # pyright: ignore[reportGeneralTypeIssues] progress_scope_key = "" progress_mirror_scope_keys: list[str] = [] request_window_session_id = request.headers.get(_WINDOW_SESSION_HEADER_NAME) + legacy_submission_window_session_id = _normalise_tool_progress_window_session_id( + request_window_session_id + ) or ( + _normalise_tool_progress_window_session_id( + session.get(_LEGACY_SUBMISSION_WINDOW_SESSION_KEY) + ) + if background_task_id is not None + else None + ) show_tool_use_progress = False try: show_tool_use_progress = bool(get_show_tool_use_during_thinking()) @@ -11062,11 +11246,11 @@ def _check_background_cancellation(subtask: str) -> None: ) # JVNAUTOSCI-1011: Use window session context if available - effective = get_effective_context( - request_window_session_id, - dict(session), - user_concept_id, - require_known_window=bool(request_window_session_id), + effective = _get_effective_context_with_owned_conversation_recovery( + window_session_id=request_window_session_id, + flask_session_snapshot=dict(session), + user_concept_id=user_concept_id, + conversation_session_id=request_conversation_session_id, ) # Window-session storage uses the organisation slug in some paths. # Canonicalise it at the authenticated request boundary so downstream @@ -11391,6 +11575,7 @@ def _check_background_cancellation(subtask: str) -> None: ), queue_id=prompt_queue_id, attempt_id=attempt_id, + window_session_id=legacy_submission_window_session_id, ) else: anonymous_admission_id = session.get("_von_anonymous_admission_id") @@ -15389,11 +15574,48 @@ def history_turn_telemetry_access(): requested_history_index = request.args.get("history_index", type=int) window_session_id = request.headers.get("X-Von-Window-Session") - effective = get_effective_context(window_session_id, dict(session), user_concept_id) + try: + effective = _get_effective_context_with_owned_conversation_recovery( + window_session_id=window_session_id, + flask_session_snapshot=dict(session), + user_concept_id=user_concept_id, + conversation_session_id=requested_session_id, + ) + except WindowSessionContextRecoveryUnavailable: + return ( + jsonify( + { + "error": "window_context_recovery_unavailable", + "error_code": "window_context_recovery_unavailable", + "retryable": True, + } + ), + 503, + ) + except WindowSessionContextUnavailable: + return ( + jsonify( + { + "error": "window_context_unavailable", + "error_code": "window_context_unavailable", + "retryable": True, + } + ), + 409, + ) actor_namespace, organisation_concept_id = _resolve_history_request_scope_hints( user_concept_id=user_concept_id, effective_context=effective, ) + actor_progress_scope_key = ( + _build_authenticated_tool_progress_scope_key( + user_concept_id=user_concept_id, + organisation_concept_id=organisation_concept_id, + namespace=actor_namespace, + ) + if actor_namespace + else None + ) owner_user_id = user_concept_id owner_namespace = actor_namespace @@ -15567,6 +15789,7 @@ def history_turn_telemetry_access(): namespace=actor_namespace, user_concept_id=user_concept_id, window_session_id=window_session_id, + scope_key=actor_progress_scope_key, ) if not isinstance(live_progress, Mapping) and not diagnostics_available: return jsonify({"error": "Turn telemetry not found"}), 404 @@ -16861,6 +17084,17 @@ def set_user_concept(): ), 200, ) + except WindowSessionBindingStoreUnavailable: + return ( + jsonify( + { + "error": "window_session_binding_store_unavailable", + "error_code": "window_session_binding_store_unavailable", + "retryable": True, + } + ), + 503, + ) except WindowSessionOwnershipError: return ( jsonify( @@ -16917,6 +17151,11 @@ def set_organisation(): if "+" in user_slug: user_slug = user_slug.split("+", 1)[0] user_slug = re.sub(r"[^a-z0-9]+", "_", user_slug.strip().lower()).strip("_") + user_concept_id = ( + str(user_id).strip() + if str(user_id).strip().startswith("#") + else f"#V#{user_slug}" + ) # Check if this is a clear request (empty dict or explicit null/empty string) is_clear_request = "organisation_concept_id" in data and not org_id @@ -16927,7 +17166,11 @@ def set_organisation(): # JVNAUTOSCI-1011: Use window session if header present if window_session_id: - clear_window_organisation(window_session_id, namespace, user_id) + clear_window_organisation( + window_session_id, + namespace, + user_concept_id, + ) else: # Fallback: update Flask session session.pop("organisation_concept_id", None) @@ -16964,68 +17207,74 @@ def set_organisation(): from ...services.organisation_membership_service import ( resolve_user_organisation_membership, ) - - user_concept_id = ( - str(user_id).strip() - if str(user_id).strip().startswith("#") - else f"#V#{user_slug}" + from ...services.ontology_authority_membership_coordination_service import ( + organisation_membership_scope_barrier, ) + organisation_concept_id = f"#V#{org_slug}" - try: - membership = resolve_user_organisation_membership( - user_concept_id, - organisation_concept_id, - ) - except ValueError: - membership = None - except Exception as exc: - current_app.logger.warning( - "Organisation membership resolution failed for session scope", - extra={"exception_type": type(exc).__name__}, - ) - return ( - jsonify( - { - "error": "organisation_membership_unavailable", - "error_code": "organisation_membership_unavailable", - } - ), - 503, - ) - if membership is None: - return ( - jsonify( - { - "error": "organisation_membership_required", - "error_code": "organisation_membership_required", - } - ), - 403, - ) + # Make the membership read and derived browser-scope publication one + # operation with respect to canonical membership mutations. This + # prevents a restart recovery or explicit selection from recreating a + # binding between revocation invalidation and the represented write. + with organisation_membership_scope_barrier( + user_concept_id, + organisation_concept_id, + ): + try: + membership = resolve_user_organisation_membership( + user_concept_id, + organisation_concept_id, + ) + except ValueError: + membership = None + except Exception as exc: + current_app.logger.warning( + "Organisation membership resolution failed for session scope", + extra={"exception_type": type(exc).__name__}, + ) + return ( + jsonify( + { + "error": "organisation_membership_unavailable", + "error_code": "organisation_membership_unavailable", + } + ), + 503, + ) + if membership is None: + return ( + jsonify( + { + "error": "organisation_membership_required", + "error_code": "organisation_membership_required", + } + ), + 403, + ) - role_in_org = str(membership.get("role") or "member").strip() or "member" + role_in_org = str(membership.get("role") or "member").strip() or "member" - # Derive composite namespace using slug values - namespace = derive_namespace(user_slug, org_slug) + # Derive composite namespace using slug values + namespace = derive_namespace(user_slug, org_slug) - # JVNAUTOSCI-1011: Use window session if header present - if window_session_id: - set_window_organisation( - window_session_id=window_session_id, - organisation_concept_id=org_slug, - role_in_org=role_in_org, - namespace=namespace, - user_id=user_id, - ) - else: - # Fallback: update Flask session (for clients without window session support) - session["organisation_concept_id"] = org_slug - session["role_in_org"] = role_in_org - session["namespace"] = namespace - # JVNAUTOSCI-1004: Clear chat session_id when org changes to avoid - # showing conversation from previous org context - session.pop("session_id", None) - session.modified = True + # JVNAUTOSCI-1011: Use window session if header present + if window_session_id: + set_window_organisation( + window_session_id=window_session_id, + organisation_concept_id=org_slug, + role_in_org=role_in_org, + namespace=namespace, + user_id=user_concept_id, + ) + else: + # Fallback: update Flask session (for clients without window session support) + session["organisation_concept_id"] = org_slug + session["role_in_org"] = role_in_org + session["namespace"] = namespace + # JVNAUTOSCI-1004: Clear chat session_id when org changes to avoid + # showing conversation from previous org context + session.pop("session_id", None) + session.modified = True # Return concept ID form in API response (with #V# prefix) concept_id_response = ( @@ -17046,6 +17295,28 @@ def set_organisation(): 200, ) + except WindowSessionBindingStoreUnavailable: + return ( + jsonify( + { + "error": "window_session_binding_store_unavailable", + "error_code": "window_session_binding_store_unavailable", + "retryable": True, + } + ), + 503, + ) + except OntologyMutationResourceBusy: + return ( + jsonify( + { + "error": "window_session_scope_coordination_busy", + "error_code": "window_session_scope_coordination_busy", + "retryable": True, + } + ), + 409, + ) except WindowSessionOwnershipError: return ( jsonify( @@ -17096,11 +17367,29 @@ def get_session_context(): 200, ) + user_slug = str(user_id) + if user_slug.startswith("#V#"): + user_slug = user_slug[3:] + if "@" in user_slug: + user_slug = user_slug.split("@", 1)[0] + if "+" in user_slug: + user_slug = user_slug.split("+", 1)[0] + user_slug = re.sub(r"[^a-z0-9]+", "_", user_slug.strip().lower()).strip("_") + user_concept_id = ( + str(user_id).strip() + if str(user_id).strip().startswith("#") + else f"#V#{user_slug}" + ) + # JVNAUTOSCI-1011: Check for window session header window_session_id = request.headers.get("X-Von-Window-Session") # Get effective context from window session or Flask session - effective = get_effective_context(window_session_id, dict(session), user_id) + effective = get_effective_context( + window_session_id, + dict(session), + user_concept_id, + ) org_id = effective.get("organisation_id") role_in_org = effective.get("role") @@ -17108,14 +17397,6 @@ def get_session_context(): # If no namespace resolved, derive it if not namespace: - user_slug = str(user_id) - if user_slug.startswith("#V#"): - user_slug = user_slug[3:] - if "@" in user_slug: - user_slug = user_slug.split("@", 1)[0] - if "+" in user_slug: - user_slug = user_slug.split("+", 1)[0] - user_slug = re.sub(r"[^a-z0-9]+", "_", user_slug.strip().lower()).strip("_") if org_id: # Get role if not in session if not role_in_org: diff --git a/src/backend/services/chat_prompt_queue_service.py b/src/backend/services/chat_prompt_queue_service.py index fde4a464..dbeb641d 100644 --- a/src/backend/services/chat_prompt_queue_service.py +++ b/src/backend/services/chat_prompt_queue_service.py @@ -8,6 +8,7 @@ import hashlib import os +import re import threading from datetime import datetime, timedelta, timezone from typing import Any, Mapping, Sequence @@ -47,6 +48,14 @@ STALE_IN_PROGRESS_LAST_ERROR = ( "Prompt queue record expired after being in progress for more than 24 hours." ) +LEGACY_SUBMISSION_ROLE_QUEUE = "queue" +LEGACY_SUBMISSION_ROLE_GENERATE = "generate" +LEGACY_SUBMISSION_ROLES = { + LEGACY_SUBMISSION_ROLE_QUEUE, + LEGACY_SUBMISSION_ROLE_GENERATE, +} +LEGACY_SUBMISSION_RENDEZVOUS_SECONDS = 5 * 60 +_LEGACY_VONTOLOGY_NON_TRIGGER_PREFIX_RE = re.compile(r"#([Vv])\u200B#") class ChatPromptQueueError(RuntimeError): @@ -273,6 +282,193 @@ def build_queue_scope( ) +def _build_active_legacy_submission_key( + *, + scope: Mapping[str, Any], + conversation_key: str | None, + prompt_raw: str, + window_session_id: str | None, +) -> str | None: + """Build a content-free idempotency key for one old-client active send.""" + + conversation_key_clean = _coerce_scope_value( + conversation_key, + field="conversation_key", + required=False, + ) + window_session_id_clean = _coerce_scope_value( + window_session_id, + field="window_session_id", + required=False, + ) + if not conversation_key_clean or not window_session_id_clean: + return None + canonical_scope = _scope_query(scope) + canonical_prompt = _LEGACY_VONTOLOGY_NON_TRIGGER_PREFIX_RE.sub( + r"#\1#", + prompt_raw, + ).strip() + prompt_digest = hashlib.sha256(canonical_prompt.encode("utf-8")).hexdigest() + window_session_digest = hashlib.sha256( + window_session_id_clean.encode("utf-8") + ).hexdigest() + material = "\x1f".join( + ( + "active_legacy_submission.v1", + str(canonical_scope.get("user_concept_id") or ""), + str(canonical_scope.get("organisation_concept_id") or ""), + str(canonical_scope.get("namespace") or ""), + conversation_key_clean, + window_session_digest, + prompt_digest, + ) + ) + return hashlib.sha256(material.encode("utf-8")).hexdigest() + + +def _expire_unpaired_legacy_submissions( + *, + coll: Collection, + now: datetime, +) -> int: + """Remove bounded rendezvous state without changing queue lifecycle truth.""" + + result = coll.update_many( + { + "active_legacy_submission_key": {"$exists": True}, + "legacy_submission_expires_at": {"$lte": now}, + }, + { + "$unset": { + "active_legacy_submission_key": "", + "legacy_submission_queue_seen": "", + "legacy_submission_generate_seen": "", + "legacy_submission_expires_at": "", + } + }, + ) + return int(getattr(result, "modified_count", 0) or 0) + + +def _find_or_join_legacy_submission( + *, + coll: Collection, + scope: Mapping[str, Any], + active_legacy_submission_key: str, + role: str, + client_request_id: str | None, + now: datetime, +) -> dict[str, Any] | None: + """Reuse the same half or atomically join the complementary old-client half.""" + + existing = coll.find_one( + { + **_scope_query(scope), + "active_legacy_submission_key": active_legacy_submission_key, + "legacy_submission_expires_at": {"$gt": now}, + } + ) + if not isinstance(existing, Mapping): + return None + + role_field = f"legacy_submission_{role}_seen" + other_role = ( + LEGACY_SUBMISSION_ROLE_GENERATE + if role == LEGACY_SUBMISSION_ROLE_QUEUE + else LEGACY_SUBMISSION_ROLE_QUEUE + ) + other_role_field = f"legacy_submission_{other_role}_seen" + if existing.get(role_field) is True: + if role == LEGACY_SUBMISSION_ROLE_GENERATE: + existing_request_id = _coerce_scope_value( + existing.get("client_request_id"), + field="client_request_id", + required=False, + ) + if ( + existing_request_id + and client_request_id + and existing_request_id != client_request_id + ): + if ( + existing.get("status") in TERMINAL_STATUSES + and existing.get(other_role_field) is not True + ): + retired = coll.find_one_and_update( + { + **_scope_query(scope), + "queue_id": str(existing.get("queue_id") or ""), + "active_legacy_submission_key": ( + active_legacy_submission_key + ), + "status": {"$in": list(TERMINAL_STATUSES)}, + role_field: True, + other_role_field: {"$ne": True}, + }, + { + "$set": {"legacy_submission_closed_at": now}, + "$unset": { + "active_legacy_submission_key": "", + "legacy_submission_queue_seen": "", + "legacy_submission_generate_seen": "", + "legacy_submission_expires_at": "", + }, + }, + return_document=ReturnDocument.AFTER, + ) + if retired is not None: + return None + if ( + coll.find_one( + { + **_scope_query(scope), + "active_legacy_submission_key": ( + active_legacy_submission_key + ), + }, + {"_id": 1}, + ) + is None + ): + return None + raise ConversationTurnAlreadyActive( + "another matching turn is active for this conversation", + queue_id=str(existing.get("queue_id") or "") or None, + ) + return serialise_queue_record(existing) + + set_fields: dict[str, Any] = {role_field: True} + update: dict[str, Any] = {"$set": set_fields} + if existing.get(other_role_field) is True: + set_fields["legacy_submission_closed_at"] = now + update["$unset"] = { + "active_legacy_submission_key": "", + "legacy_submission_expires_at": "", + } + queue_id = str(existing.get("queue_id") or "") + joined = coll.find_one_and_update( + { + **_scope_query(scope), + "queue_id": queue_id, + "active_legacy_submission_key": active_legacy_submission_key, + role_field: {"$ne": True}, + }, + update, + return_document=ReturnDocument.AFTER, + ) + if joined is None: + # A same-role retry can race the complementary join that closed the + # unique key. Re-read only the exact row already selected above. + joined = coll.find_one( + { + **_scope_query(scope), + "queue_id": queue_id, + role_field: True, + } + ) + return serialise_queue_record(joined) + + def _scope_query(scope: Mapping[str, Any]) -> dict[str, Any]: return _queue_scope_values(scope) @@ -471,6 +667,8 @@ def create_queue_record( client_request_id: Any = None, attempt_id: Any = None, conversation_key: Any = None, + legacy_submission_role: str | None = None, + window_session_id: Any = None, ) -> dict[str, Any]: if status not in {STATUS_QUEUED, STATUS_IN_PROGRESS}: raise InvalidChatPromptQueueInput("status must be queued or in_progress") @@ -506,6 +704,25 @@ def create_queue_record( field="conversation_key", required=False, ) + legacy_role_clean = _coerce_scope_value( + legacy_submission_role, + field="legacy_submission_role", + required=False, + ) + if legacy_role_clean and legacy_role_clean not in LEGACY_SUBMISSION_ROLES: + raise InvalidChatPromptQueueInput( + "legacy_submission_role must be queue or generate" + ) + active_legacy_submission_key = ( + _build_active_legacy_submission_key( + scope=scope, + conversation_key=conversation_key_clean, + prompt_raw=prompt, + window_session_id=window_session_id, + ) + if legacy_role_clean + else None + ) now = _now() doc = { "queue_id": str(uuid4()), @@ -526,7 +743,30 @@ def create_queue_record( "completed_at": None, "last_error": None, } + if active_legacy_submission_key: + doc["active_legacy_submission_key"] = active_legacy_submission_key + doc["legacy_submission_queue_seen"] = ( + legacy_role_clean == LEGACY_SUBMISSION_ROLE_QUEUE + ) + doc["legacy_submission_generate_seen"] = ( + legacy_role_clean == LEGACY_SUBMISSION_ROLE_GENERATE + ) + doc["legacy_submission_expires_at"] = now + timedelta( + seconds=LEGACY_SUBMISSION_RENDEZVOUS_SECONDS + ) coll = _collection() + if active_legacy_submission_key: + _expire_unpaired_legacy_submissions(coll=coll, now=now) + reusable = _find_or_join_legacy_submission( + coll=coll, + scope=scope, + active_legacy_submission_key=active_legacy_submission_key, + role=str(legacy_role_clean), + client_request_id=client_request_id_clean, + now=now, + ) + if reusable is not None: + return reusable if status == STATUS_QUEUED: global_limit, per_user_limit = _foreground_queue_limits() user_id = str(doc["user_concept_id"]) @@ -571,6 +811,17 @@ def create_queue_record( try: coll.insert_one(candidate) except DuplicateKeyError: + if active_legacy_submission_key: + reusable = _find_or_join_legacy_submission( + coll=coll, + scope=scope, + active_legacy_submission_key=active_legacy_submission_key, + role=str(legacy_role_clean), + client_request_id=client_request_id_clean, + now=now, + ) + if reusable is not None: + return reusable continue doc = candidate inserted = True @@ -591,7 +842,21 @@ def create_queue_record( limit_kind="global", ) else: - coll.insert_one(doc) + try: + coll.insert_one(doc) + except DuplicateKeyError: + if active_legacy_submission_key: + reusable = _find_or_join_legacy_submission( + coll=coll, + scope=scope, + active_legacy_submission_key=active_legacy_submission_key, + role=str(legacy_role_clean), + client_request_id=client_request_id_clean, + now=now, + ) + if reusable is not None: + return reusable + raise record = serialise_queue_record(doc) if record is None: # pragma: no cover - defensive raise ChatPromptQueueUnavailable("created queue record could not be serialised") @@ -1163,6 +1428,21 @@ def finish_prompt_record( } ) ) + if record is None and not attempt_id_clean: + # Pre-#473 clients terminalise the record in a finally block after the + # server-owned generate lifecycle has already done so. Limit this + # idempotent acknowledgement to an exact, fully joined legacy pair. + record = serialise_queue_record( + coll.find_one( + { + **_compatible_scope_query(scope), + "queue_id": queue_id_clean, + "status": status, + "legacy_submission_queue_seen": True, + "legacy_submission_generate_seen": True, + } + ) + ) if record is None: raise _transition_not_found_error( scope=scope, diff --git a/src/backend/services/conversation_turn_admission_service.py b/src/backend/services/conversation_turn_admission_service.py index aa51bc58..a784fd5c 100644 --- a/src/backend/services/conversation_turn_admission_service.py +++ b/src/backend/services/conversation_turn_admission_service.py @@ -307,6 +307,7 @@ def acquire( conversation_key: str, queue_id: str | None = None, attempt_id: str | None = None, + window_session_id: str | None = None, ) -> ConversationTurnAdmissionToken: """Acquire capacity and the durable conversation fence without waiting.""" @@ -331,9 +332,20 @@ def acquire( client_request_id=client_request_id, attempt_id=attempt_id, conversation_key=conversation_key, + legacy_submission_role=( + chat_prompt_queue_service.LEGACY_SUBMISSION_ROLE_GENERATE + if not isinstance(attempt_id, str) or not attempt_id.strip() + else None + ), + window_session_id=window_session_id, ) except chat_prompt_queue_service.ChatPromptQueueCapacityReached as exc: raise ConversationTurnCapacityReached(str(exc)) from exc + except chat_prompt_queue_service.ConversationTurnAlreadyActive as exc: + raise ConversationTurnActive( + str(exc), + queue_id=exc.queue_id, + ) from exc bound_queue_id = str(record["queue_id"]) try: bound = chat_prompt_queue_service.bind_queue_record_to_turn( diff --git a/src/backend/services/ontology_authority_membership_coordination_service.py b/src/backend/services/ontology_authority_membership_coordination_service.py index f8ec6f89..c6b6f7ef 100644 --- a/src/backend/services/ontology_authority_membership_coordination_service.py +++ b/src/backend/services/ontology_authority_membership_coordination_service.py @@ -9,11 +9,13 @@ from __future__ import annotations +import time from collections.abc import Iterator from contextlib import contextmanager from contextvars import ContextVar from .ontology_publication_authority_service import ( + OntologyMutationResourceBusy, ontology_mutation_resource_lock, ) @@ -22,6 +24,21 @@ "ontology_authority_membership_coordination_depth", default=0, ) +_SCOPE_COORDINATION_STACK: ContextVar[tuple[str, ...]] = ContextVar( + "organisation_membership_scope_coordination_stack", + default=(), +) +_SCOPE_RESOURCE_PREFIX = "organisation-membership-scope:v1:" +_SCOPE_LOCK_LEASE_SECONDS = 15 +_SCOPE_LOCK_WAIT_SECONDS = 3.0 +_SCOPE_LOCK_RETRY_SECONDS = 0.025 + + +def _normalise_scope_component(value: str, *, field: str) -> str: + cleaned = str(value or "").strip() + if not cleaned: + raise ValueError(f"{field} is required") + return cleaned if cleaned.startswith("#V#") else f"#V#{cleaned}" @contextmanager @@ -40,4 +57,60 @@ def ontology_authority_membership_mutation_barrier() -> Iterator[None]: _COORDINATION_DEPTH.reset(token) -__all__ = ["ontology_authority_membership_mutation_barrier"] +@contextmanager +def organisation_membership_scope_barrier( + user_concept_id: str, + organisation_concept_id: str, +) -> Iterator[None]: + """Coordinate one actor/org membership mutation with scope recovery. + + Unlike the global authority lifecycle barrier, unrelated actor/org pairs + use different leases and never contend. A short bounded wait lets + simultaneous cold-tab readers follow an in-flight mutation or one another + instead of failing on the lock's first non-blocking acquisition attempt. + """ + + user_id = _normalise_scope_component(user_concept_id, field="user_concept_id") + org_id = _normalise_scope_component( + organisation_concept_id, + field="organisation_concept_id", + ) + resource_key = f"{_SCOPE_RESOURCE_PREFIX}{user_id}\x1f{org_id}" + stack = _SCOPE_COORDINATION_STACK.get() + if resource_key in stack: + yield + return + + deadline = time.monotonic() + _SCOPE_LOCK_WAIT_SECONDS + lock_manager = None + while lock_manager is None: + candidate = ontology_mutation_resource_lock( + resource_key, + lease_seconds=_SCOPE_LOCK_LEASE_SECONDS, + ) + try: + candidate.__enter__() + except OntologyMutationResourceBusy: + if time.monotonic() >= deadline: + raise + time.sleep(_SCOPE_LOCK_RETRY_SECONDS) + else: + lock_manager = candidate + + token = _SCOPE_COORDINATION_STACK.set((*stack, resource_key)) + try: + yield + except BaseException as exc: + suppress = lock_manager.__exit__(type(exc), exc, exc.__traceback__) + if not suppress: + raise + else: + lock_manager.__exit__(None, None, None) + finally: + _SCOPE_COORDINATION_STACK.reset(token) + + +__all__ = [ + "ontology_authority_membership_mutation_barrier", + "organisation_membership_scope_barrier", +] diff --git a/src/backend/services/organisation_membership_service.py b/src/backend/services/organisation_membership_service.py index 86b7ce20..9f299fb2 100644 --- a/src/backend/services/organisation_membership_service.py +++ b/src/backend/services/organisation_membership_service.py @@ -14,6 +14,7 @@ from ..security.access_control import bypass_access_control, can_access_concept from ..services.text_value_service import upsert_text_for_concept from .ontology_authority_membership_coordination_service import ( + organisation_membership_scope_barrier, ontology_authority_membership_mutation_barrier, ) @@ -31,12 +32,60 @@ def _coordinated_membership_mutation(function): @wraps(function) def coordinated(*args, **kwargs): - with ontology_authority_membership_mutation_barrier(): + missing = object() + user_concept_id = ( + kwargs.get("user_concept_id") + if "user_concept_id" in kwargs + else (args[0] if len(args) > 0 else missing) + ) + organisation_concept_id = ( + kwargs.get("organisation_concept_id") + if "organisation_concept_id" in kwargs + else (args[1] if len(args) > 1 else missing) + ) + if ( + user_concept_id is missing + or organisation_concept_id is missing + or not isinstance(user_concept_id, str) + or not user_concept_id + or not isinstance(organisation_concept_id, str) + or not organisation_concept_id + ): + # Preserve each public function's established argument-validation + # contract (including Python's missing-argument TypeError) before + # deriving an internal coordination key. return function(*args, **kwargs) + with ontology_authority_membership_mutation_barrier(): + with organisation_membership_scope_barrier( + user_concept_id, + organisation_concept_id, + ): + return function(*args, **kwargs) return coordinated +def _invalidate_user_window_authority( + user_concept_id: str, + organisation_concept_id: str, +) -> None: + """Remove derived tab authority before a role-reducing mutation. + + The durable deletion is shared by every web worker. A still-authorised + tab can reconstruct its selection from its actor-owned conversation on the + next request; a revoked tab cannot. + """ + + from .window_session_context_service import ( + delete_window_contexts_owned_by_user_for_organisation, + ) + + delete_window_contexts_owned_by_user_for_organisation( + user_concept_id, + organisation_concept_id, + ) + + def _normalise_concept_id(value: Any) -> str | None: if not isinstance(value, str): return None @@ -204,6 +253,14 @@ def create_organisation_membership( f"memberOf relationship already exists: {user_concept_id} --memberOf--> {organisation_concept_id}" ) + if relationship_exists: + # This API also acts as an idempotent role upsert for an existing + # membership, so invalidate the exact org's derived role first. + _invalidate_user_window_authority( + user_concept_id, + organisation_concept_id, + ) + # Store the role via text relation (hasRole predicate with context) role_result = upsert_text_for_concept( subject_concept_id=user_concept_id, @@ -487,7 +544,14 @@ def update_user_role( tv = TextValuesRepository.find_one_by_id(text_value_id) if tv: - old_role = tv.get("text", "member") + parsed_role, _stored_org = parse_organisation_role_storage_text( + tv.get("text"), + current_role_rel.get("context") + if isinstance(current_role_rel.get("context"), dict) + else {}, + ) + if parsed_role: + old_role = parsed_role # If role is the same, no update needed if old_role == new_role: @@ -498,6 +562,10 @@ def update_user_role( "role_updated": False, } + # Invalidate before the represented write so a durable-store failure cannot + # leave a completed role reduction hidden behind another worker's cache. + _invalidate_user_window_authority(user_concept_id, organisation_concept_id) + # Update role via upsert upsert_text_for_concept( subject_concept_id=user_concept_id, @@ -573,6 +641,10 @@ def remove_organisation_membership( if semantic_authority_revoked and authority_read_back.get("active") is not False: raise RuntimeError("organisation_ontology_authority_revocation_not_verified") + # See update_user_role: clearing the shared binding before the authoritative + # mutation makes cross-worker revocation fail closed. + _invalidate_user_window_authority(user_concept_id, organisation_concept_id) + # Remove the memberOf relationship ConceptsRepository.mutate_relationship_edge( source_id=user_concept_id, diff --git a/src/backend/services/window_session_binding_store_service.py b/src/backend/services/window_session_binding_store_service.py new file mode 100644 index 00000000..87825659 --- /dev/null +++ b/src/backend/services/window_session_binding_store_service.py @@ -0,0 +1,338 @@ +"""Durable, actor-bound organisation selections for browser window sessions. + +The window-session identifier is only a selector. Authentication and current +organisation membership remain the authority for every recovered scope. +""" + +from __future__ import annotations + +import hashlib +from collections.abc import Callable +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from typing import Protocol + +from pymongo.errors import DuplicateKeyError, PyMongoError + +from ..db.mongo_client import get_window_session_binding_collection + +WINDOW_SESSION_BINDING_SCHEMA_VERSION = "window_session_binding.v1" +WINDOW_SESSION_SCOPE_ORGANISATION = "organisation" +WINDOW_SESSION_SCOPE_PERSONAL = "personal" + + +class WindowSessionBindingStoreError(RuntimeError): + """Base class for durable window-binding failures.""" + + +class WindowSessionBindingStoreUnavailable(WindowSessionBindingStoreError): + """Raised when a binding cannot be read or written durably.""" + + +class WindowSessionBindingOwnershipError(WindowSessionBindingStoreError): + """Raised when an existing live selector belongs to another actor.""" + + +@dataclass(frozen=True) +class PersistedWindowSessionBinding: + """The minimum durable selection needed to reconstruct a trusted scope.""" + + window_session_key: str + user_id: str + scope_kind: str + organisation_concept_id: str | None + created_at: datetime + updated_at: datetime + expires_at: datetime + + +class WindowSessionBindingRepository(Protocol): + """Persistence seam used by the in-memory window-session cache.""" + + def load_owned( + self, window_session_id: str, user_id: str + ) -> PersistedWindowSessionBinding | None: ... + + def load_for_mutation( + self, window_session_id: str + ) -> PersistedWindowSessionBinding | None: ... + + def save( + self, + *, + window_session_id: str, + user_id: str, + organisation_concept_id: str | None, + scope_kind: str, + ) -> PersistedWindowSessionBinding: ... + + def touch_owned(self, window_session_id: str, user_id: str) -> bool: ... + + def delete_owned(self, window_session_id: str, user_id: str) -> bool: ... + + def delete_owned_for_organisation( + self, user_id: str, organisation_concept_id: str + ) -> int: ... + + def delete_all_owned(self, user_id: str) -> int: ... + + +def _clean_required(value: str, *, field: str) -> str: + cleaned = value.strip() if isinstance(value, str) else "" + if not cleaned: + raise ValueError(f"{field} is required") + return cleaned + + +def _window_session_key(window_session_id: str) -> str: + """Return a content-free stable key; never persist the browser selector.""" + + cleaned = _clean_required(window_session_id, field="window_session_id") + return "sha256:" + hashlib.sha256(cleaned.encode("utf-8")).hexdigest() + + +def _normalise_org_id(value: str | None) -> str | None: + if not isinstance(value, str) or not value.strip(): + return None + cleaned = value.strip() + return cleaned if cleaned.startswith("#V#") else f"#V#{cleaned}" + + +def _coerce_utc_datetime(value: object) -> datetime | None: + if not isinstance(value, datetime): + return None + if value.tzinfo is None: + return value.replace(tzinfo=UTC) + return value.astimezone(UTC) + + +class MongoWindowSessionBindingRepository: + """Mongo-backed L2 for the small per-window organisation selection.""" + + def __init__( + self, + *, + ttl_seconds: int, + collection_getter: Callable[[], object | None] = ( + get_window_session_binding_collection + ), + ) -> None: + self._ttl_seconds = max(1, int(ttl_seconds)) + self._collection_getter = collection_getter + + def _collection(self): + try: + collection = self._collection_getter() + except Exception as exc: + raise WindowSessionBindingStoreUnavailable( + "window session binding store is unavailable" + ) from exc + if collection is None: + raise WindowSessionBindingStoreUnavailable( + "window session binding store is unavailable" + ) + return collection + + @staticmethod + def _from_document(document: object) -> PersistedWindowSessionBinding | None: + if not isinstance(document, dict): + return None + key = document.get("_id") + user_id = document.get("user_id") + scope_kind = document.get("scope_kind") + created_at = _coerce_utc_datetime(document.get("created_at")) + updated_at = _coerce_utc_datetime(document.get("updated_at")) + expires_at = _coerce_utc_datetime(document.get("expires_at")) + if not all( + ( + isinstance(key, str) and key, + isinstance(user_id, str) and user_id, + scope_kind + in { + WINDOW_SESSION_SCOPE_ORGANISATION, + WINDOW_SESSION_SCOPE_PERSONAL, + }, + created_at is not None, + updated_at is not None, + expires_at is not None, + ) + ): + return None + organisation_concept_id = _normalise_org_id( + document.get("organisation_concept_id") + if isinstance(document.get("organisation_concept_id"), str) + else None + ) + if ( + scope_kind == WINDOW_SESSION_SCOPE_ORGANISATION + and organisation_concept_id is None + ): + return None + if scope_kind == WINDOW_SESSION_SCOPE_PERSONAL: + organisation_concept_id = None + return PersistedWindowSessionBinding( + window_session_key=key, + user_id=user_id, + scope_kind=scope_kind, + organisation_concept_id=organisation_concept_id, + created_at=created_at, + updated_at=updated_at, + expires_at=expires_at, + ) + + def _load(self, query: dict[str, object]) -> PersistedWindowSessionBinding | None: + now = datetime.now(UTC) + try: + document = self._collection().find_one( + {**query, "expires_at": {"$gt": now}} + ) + except PyMongoError as exc: + raise WindowSessionBindingStoreUnavailable( + "window session binding read failed" + ) from exc + return self._from_document(document) + + def load_owned( + self, window_session_id: str, user_id: str + ) -> PersistedWindowSessionBinding | None: + return self._load( + { + "_id": _window_session_key(window_session_id), + "user_id": _clean_required(user_id, field="user_id"), + } + ) + + def load_for_mutation( + self, window_session_id: str + ) -> PersistedWindowSessionBinding | None: + return self._load({"_id": _window_session_key(window_session_id)}) + + def save( + self, + *, + window_session_id: str, + user_id: str, + organisation_concept_id: str | None, + scope_kind: str, + ) -> PersistedWindowSessionBinding: + clean_user_id = _clean_required(user_id, field="user_id") + if scope_kind not in { + WINDOW_SESSION_SCOPE_ORGANISATION, + WINDOW_SESSION_SCOPE_PERSONAL, + }: + raise ValueError("scope_kind must be organisation or personal") + clean_org_id = _normalise_org_id(organisation_concept_id) + if scope_kind == WINDOW_SESSION_SCOPE_ORGANISATION and not clean_org_id: + raise ValueError("organisation_concept_id is required") + if scope_kind == WINDOW_SESSION_SCOPE_PERSONAL: + clean_org_id = None + + key = _window_session_key(window_session_id) + now = datetime.now(UTC) + expires_at = now + timedelta(seconds=self._ttl_seconds) + collection = self._collection() + try: + # An expired selector may be reclaimed, matching the L1 semantics. + collection.delete_one({"_id": key, "expires_at": {"$lte": now}}) + result = collection.update_one( + {"_id": key, "user_id": clean_user_id}, + { + "$setOnInsert": { + "schema_version": WINDOW_SESSION_BINDING_SCHEMA_VERSION, + "created_at": now, + }, + "$set": { + "user_id": clean_user_id, + "scope_kind": scope_kind, + "organisation_concept_id": clean_org_id, + "updated_at": now, + "expires_at": expires_at, + }, + }, + upsert=True, + ) + except DuplicateKeyError as exc: + raise WindowSessionBindingOwnershipError( + "window_session_owned_by_different_actor" + ) from exc + except PyMongoError as exc: + raise WindowSessionBindingStoreUnavailable( + "window session binding write failed" + ) from exc + if not getattr(result, "acknowledged", True): + raise WindowSessionBindingStoreUnavailable( + "window session binding write was not acknowledged" + ) + binding = self.load_owned(window_session_id, clean_user_id) + if binding is None: + raise WindowSessionBindingStoreUnavailable( + "window session binding write could not be read back" + ) + return binding + + def touch_owned(self, window_session_id: str, user_id: str) -> bool: + now = datetime.now(UTC) + try: + result = self._collection().update_one( + { + "_id": _window_session_key(window_session_id), + "user_id": _clean_required(user_id, field="user_id"), + "expires_at": {"$gt": now}, + }, + { + "$set": { + "updated_at": now, + "expires_at": now + timedelta(seconds=self._ttl_seconds), + } + }, + ) + except PyMongoError as exc: + raise WindowSessionBindingStoreUnavailable( + "window session binding refresh failed" + ) from exc + return bool(getattr(result, "matched_count", 0)) + + def delete_owned(self, window_session_id: str, user_id: str) -> bool: + try: + result = self._collection().delete_one( + { + "_id": _window_session_key(window_session_id), + "user_id": _clean_required(user_id, field="user_id"), + } + ) + except PyMongoError as exc: + raise WindowSessionBindingStoreUnavailable( + "window session binding deletion failed" + ) from exc + return bool(getattr(result, "deleted_count", 0)) + + def delete_all_owned(self, user_id: str) -> int: + try: + result = self._collection().delete_many( + {"user_id": _clean_required(user_id, field="user_id")} + ) + except PyMongoError as exc: + raise WindowSessionBindingStoreUnavailable( + "window session binding deletion failed" + ) from exc + return int(getattr(result, "deleted_count", 0) or 0) + + def delete_owned_for_organisation( + self, user_id: str, organisation_concept_id: str + ) -> int: + clean_org_id = _normalise_org_id(organisation_concept_id) + if not clean_org_id: + raise ValueError("organisation_concept_id is required") + try: + result = self._collection().delete_many( + { + "user_id": _clean_required(user_id, field="user_id"), + "scope_kind": WINDOW_SESSION_SCOPE_ORGANISATION, + "organisation_concept_id": clean_org_id, + } + ) + except PyMongoError as exc: + raise WindowSessionBindingStoreUnavailable( + "window session binding deletion failed" + ) from exc + return int(getattr(result, "deleted_count", 0) or 0) diff --git a/src/backend/services/window_session_context_service.py b/src/backend/services/window_session_context_service.py index 596f324b..6d5d1b9f 100644 --- a/src/backend/services/window_session_context_service.py +++ b/src/backend/services/window_session_context_service.py @@ -5,8 +5,8 @@ to have independent organisation contexts without interfering with each other. The browser-wide Flask session cookie is used for authentication (user identity), -while window-specific context (organisation, namespace, role) is stored in -an in-memory dict keyed by window_session_id. +while window-specific context is cached in memory and its actor-owned +organisation selection is durably persisted for restart recovery. This enables use cases like: - Window A: #V#the_lu_witbrock_household @@ -18,11 +18,21 @@ import logging import threading import time -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from datetime import datetime, timezone from typing import Any, Dict, Optional from uuid import uuid4 +from .window_session_binding_store_service import ( + MongoWindowSessionBindingRepository, + PersistedWindowSessionBinding, + WindowSessionBindingOwnershipError, + WindowSessionBindingRepository, + WindowSessionBindingStoreUnavailable, + WINDOW_SESSION_SCOPE_ORGANISATION, + WINDOW_SESSION_SCOPE_PERSONAL, +) + logger = logging.getLogger(__name__) # TTL for window sessions in seconds (1 hour) @@ -31,6 +41,9 @@ # Cleanup interval in seconds (every 5 minutes) CLEANUP_INTERVAL_SECONDS = 300 +# Refresh the durable sliding expiry at most once per cleanup interval. +DURABLE_REFRESH_INTERVAL_SECONDS = CLEANUP_INTERVAL_SECONDS + class WindowSessionOwnershipError(PermissionError): """Raised when a live window-session ID is used by a different actor.""" @@ -40,6 +53,10 @@ class WindowSessionContextUnavailable(PermissionError): """Raised when an actor-bound tab selector has no usable server binding.""" +class WindowSessionContextRecoveryUnavailable(WindowSessionContextUnavailable): + """Raised when an explicit selector cannot be recovered from durable state.""" + + def _normalise_owner_id(value: Optional[str]) -> Optional[str]: if not isinstance(value, str): return None @@ -69,6 +86,10 @@ class WindowSessionContext: last_accessed_at: datetime = field( default_factory=lambda: datetime.now(timezone.utc) ) + durable_scope_kind: Optional[str] = field(default=None, repr=False) + durable_recovered: bool = field(default=False, repr=False) + durable_refreshed_at: Optional[datetime] = field(default=None, repr=False) + durably_persisted: bool = field(default=False, repr=False) def touch(self) -> None: """Update last accessed timestamp.""" @@ -95,17 +116,21 @@ def to_dict(self) -> Dict[str, Any]: class WindowSessionStore: """ - Thread-safe in-memory store for window session contexts. + Thread-safe in-memory L1 for window session contexts. - In production, this could be backed by Redis or a similar store - for multi-process/multi-server deployments. + An optional durable L2 stores only the actor-owned organisation selection. + Roles and namespaces are always re-derived after recovery. """ - def __init__(self) -> None: + def __init__( + self, + binding_repository: WindowSessionBindingRepository | None = None, + ) -> None: self._sessions: Dict[str, WindowSessionContext] = {} self._lock = threading.RLock() self._cleanup_thread: Optional[threading.Thread] = None self._shutdown = False + self._binding_repository = binding_repository def start_cleanup_thread(self) -> None: """Start background thread that periodically cleans up expired sessions.""" @@ -184,6 +209,156 @@ def get_or_create( self._sessions[window_session_id] = ctx return ctx + @staticmethod + def _context_from_binding( + window_session_id: str, + binding: PersistedWindowSessionBinding, + ) -> WindowSessionContext: + return WindowSessionContext( + window_session_id=window_session_id, + user_id=binding.user_id, + organisation_concept_id=binding.organisation_concept_id, + role_in_org=None, + namespace=None, + created_at=binding.created_at, + last_accessed_at=datetime.now(timezone.utc), + durable_scope_kind=binding.scope_kind, + durable_recovered=True, + durable_refreshed_at=binding.updated_at, + durably_persisted=True, + ) + + def get_owned( + self, + window_session_id: str, + user_id: Optional[str], + ) -> Optional[WindowSessionContext]: + """Return one exact actor-owned context from the shared binding. + + The durable row is deliberately consulted even when this process has a + warm L1 entry. Otherwise two web workers can keep different + organisation selections for the same tab after a switch, or continue + using a binding invalidated by a membership change. + """ + + requested_owner = _normalise_owner_id(user_id) + if not requested_owner: + return None + ctx = self.get(window_session_id) + repository = self._binding_repository + if repository is None: + if ctx is not None and _normalise_owner_id(ctx.user_id) == requested_owner: + return ctx + return None + try: + binding = repository.load_owned(window_session_id, requested_owner) + except WindowSessionBindingStoreUnavailable as exc: + raise WindowSessionContextRecoveryUnavailable( + "window_session_context_recovery_unavailable" + ) from exc + if binding is None: + # A shared deletion or expiry outranks this process's cache. Do + # not reveal or remove an entry owned by a different actor. + with self._lock: + cached = self._sessions.get(window_session_id) + if ( + cached is not None + and _normalise_owner_id(cached.user_id) == requested_owner + and not cached.durably_persisted + and not _has_authoritative_scope(cached) + ): + # Chat-only legacy state is not an authority carrier and + # may still enrich a non-strict Flask compatibility read. + return cached + if ( + cached is not None + and _normalise_owner_id(cached.user_id) == requested_owner + ): + del self._sessions[window_session_id] + return None + + binding_org = _normalise_owner_id(binding.organisation_concept_id) + cached_org = _normalise_owner_id(ctx.organisation_concept_id) if ctx else None + cached_scope = ctx.durable_scope_kind if ctx else None + cached_owner = _normalise_owner_id(ctx.user_id) if ctx else None + if ( + ctx is not None + and cached_owner == requested_owner + and cached_scope == binding.scope_kind + and cached_org == binding_org + ): + ctx.durably_persisted = True + ctx.durable_refreshed_at = binding.updated_at + self._refresh_durable_binding_if_due(ctx) + return ctx + + recovered = self._context_from_binding(window_session_id, binding) + if ctx is not None and cached_owner == requested_owner: + # The conversation choice is process-local and is not an authority + # carrier. Preserve it while replacing stale organisation data. + recovered.chat_session_id = ctx.chat_session_id + # The exact actor-owned durable row outranks an unowned or other-actor + # partial L1 entry. It never reveals that row to the other actor. + with self._lock: + self._sessions[window_session_id] = recovered + return recovered + + def persist_authoritative_binding( + self, + ctx: WindowSessionContext, + *, + scope_kind: str, + ) -> bool: + """Persist an explicit Personal/organisation choice before L1 publish.""" + + repository = self._binding_repository + owner = _normalise_owner_id(ctx.user_id) + if repository is None: + return False + if not owner: + raise WindowSessionBindingStoreUnavailable( + "authenticated window session owner is required" + ) + try: + binding = repository.save( + window_session_id=ctx.window_session_id, + user_id=owner, + organisation_concept_id=ctx.organisation_concept_id, + scope_kind=scope_kind, + ) + except WindowSessionBindingOwnershipError as exc: + raise WindowSessionOwnershipError( + "window_session_owned_by_different_actor" + ) from exc + ctx.durable_scope_kind = binding.scope_kind + ctx.durable_recovered = False + ctx.durable_refreshed_at = binding.updated_at + ctx.durably_persisted = True + return True + + def _refresh_durable_binding_if_due(self, ctx: WindowSessionContext) -> None: + repository = self._binding_repository + owner = _normalise_owner_id(ctx.user_id) + if repository is None or not owner or not ctx.durably_persisted: + return + refreshed_at = ctx.durable_refreshed_at + if isinstance(refreshed_at, datetime): + if refreshed_at.tzinfo is None: + refreshed_at = refreshed_at.replace(tzinfo=timezone.utc) + age_seconds = (datetime.now(timezone.utc) - refreshed_at).total_seconds() + if age_seconds < DURABLE_REFRESH_INTERVAL_SECONDS: + return + try: + refreshed = repository.touch_owned(ctx.window_session_id, owner) + except WindowSessionBindingStoreUnavailable: + logger.warning( + "Unable to refresh durable window-session binding expiry", + extra={"window_session_owner_present": True}, + ) + return + if refreshed: + ctx.durable_refreshed_at = datetime.now(timezone.utc) + def set(self, ctx: WindowSessionContext) -> None: """Store a context without replacing another live actor's entry.""" with self._lock: @@ -220,19 +395,23 @@ def delete_if_owned( user_id: Optional[str], ) -> bool: """Delete a live context only when the authenticated owner matches.""" + deleted_from_memory = False with self._lock: ctx = self._sessions.get(window_session_id) - if ctx is None: - return False - if ctx.is_expired(): + if ctx is not None and ctx.is_expired(): del self._sessions[window_session_id] - return False - stored_owner = _normalise_owner_id(ctx.user_id) + ctx = None + stored_owner = _normalise_owner_id(ctx.user_id) if ctx else None requested_owner = _normalise_owner_id(user_id) - if not stored_owner or stored_owner != requested_owner: - return False - del self._sessions[window_session_id] - return True + if ctx is not None and stored_owner == requested_owner: + del self._sessions[window_session_id] + deleted_from_memory = True + deleted_from_durable = False + if self._binding_repository is not None and requested_owner: + deleted_from_durable = self._binding_repository.delete_owned( + window_session_id, requested_owner + ) + return deleted_from_memory or deleted_from_durable def delete_all_owned(self, user_id: Optional[str]) -> int: """Delete every live window context owned by one exact actor.""" @@ -248,7 +427,48 @@ def delete_all_owned(self, user_id: Optional[str]) -> int: ] for window_session_id in matching_ids: del self._sessions[window_session_id] - return len(matching_ids) + durable_count = 0 + if self._binding_repository is not None: + durable_count = self._binding_repository.delete_all_owned(requested_owner) + return max(len(matching_ids), durable_count) + + def delete_owned_for_organisation( + self, + user_id: Optional[str], + organisation_concept_id: Optional[str], + ) -> int: + """Delete one actor's tabs bound to one organisation only.""" + + requested_owner = _normalise_owner_id(user_id) + requested_org = _normalise_owner_id(organisation_concept_id) + if not requested_owner or not requested_org: + return 0 + canonical_org = ( + requested_org + if requested_org.startswith("#V#") + else f"#V#{requested_org}" + ) + with self._lock: + matching_ids = [ + window_session_id + for window_session_id, ctx in self._sessions.items() + if _normalise_owner_id(ctx.user_id) == requested_owner + and ( + _normalise_owner_id(ctx.organisation_concept_id) + in {requested_org, canonical_org} + ) + ] + for window_session_id in matching_ids: + del self._sessions[window_session_id] + durable_count = 0 + if self._binding_repository is not None: + durable_count = ( + self._binding_repository.delete_owned_for_organisation( + requested_owner, + canonical_org, + ) + ) + return max(len(matching_ids), durable_count) def cleanup_expired(self) -> int: """Remove all expired sessions. Returns count removed.""" @@ -270,13 +490,27 @@ def count(self) -> int: # Global singleton store _window_session_store: Optional[WindowSessionStore] = None +_window_session_binding_repository: WindowSessionBindingRepository | None = None + + +def get_window_session_binding_repository() -> WindowSessionBindingRepository: + """Return the process repository used beneath the in-memory L1.""" + + global _window_session_binding_repository + if _window_session_binding_repository is None: + _window_session_binding_repository = MongoWindowSessionBindingRepository( + ttl_seconds=WINDOW_SESSION_TTL_SECONDS + ) + return _window_session_binding_repository def get_window_session_store() -> WindowSessionStore: """Get or create the global window session store.""" global _window_session_store if _window_session_store is None: - _window_session_store = WindowSessionStore() + _window_session_store = WindowSessionStore( + binding_repository=get_window_session_binding_repository() + ) _window_session_store.start_cleanup_thread() return _window_session_store @@ -314,14 +548,27 @@ def set_window_organisation( ) -> WindowSessionContext: """Set organisation context for a window session.""" store = get_window_session_store() - ctx = store.get_or_create(window_session_id, user_id) - ctx.organisation_concept_id = organisation_concept_id - ctx.role_in_org = role_in_org - ctx.namespace = namespace - # Clear chat session when org changes (JVNAUTOSCI-1004 pattern) - ctx.chat_session_id = None - store.set(ctx) - return ctx + current = store.get_or_create(window_session_id, user_id) + clean_org_id = _normalise_owner_id(organisation_concept_id) + canonical_org_id = ( + clean_org_id + if clean_org_id is None or clean_org_id.startswith("#V#") + else f"#V#{clean_org_id}" + ) + updated = replace( + current, + organisation_concept_id=canonical_org_id, + role_in_org=role_in_org, + namespace=namespace, + # Clear chat session when org changes (JVNAUTOSCI-1004 pattern). + chat_session_id=None, + ) + store.persist_authoritative_binding( + updated, + scope_kind=WINDOW_SESSION_SCOPE_ORGANISATION, + ) + store.set(updated) + return updated def clear_window_organisation( @@ -329,13 +576,116 @@ def clear_window_organisation( ) -> WindowSessionContext: """Clear organisation context for a window session (switch to personal).""" store = get_window_session_store() - ctx = store.get_or_create(window_session_id, user_id) - ctx.organisation_concept_id = None - ctx.role_in_org = None - ctx.namespace = namespace - ctx.chat_session_id = None - store.set(ctx) - return ctx + current = store.get_or_create(window_session_id, user_id) + updated = replace( + current, + organisation_concept_id=None, + role_in_org=None, + namespace=namespace, + chat_session_id=None, + ) + store.persist_authoritative_binding( + updated, + scope_kind=WINDOW_SESSION_SCOPE_PERSONAL, + ) + store.set(updated) + return updated + + +def _user_slug(user_id: str) -> str: + cleaned = user_id[3:] if user_id.startswith("#V#") else user_id + if "@" in cleaned: + cleaned = cleaned.split("@", 1)[0] + if "+" in cleaned: + cleaned = cleaned.split("+", 1)[0] + import re + + return re.sub(r"[^a-z0-9]+", "_", cleaned.strip().lower()).strip("_") + + +def _recover_authoritative_scope( + ctx: WindowSessionContext, + *, + user_id: str, +) -> WindowSessionContext | None: + """Revalidate durable selection against current represented membership.""" + + from .namespace_service import derive_namespace + + user_slug = _user_slug(user_id) + if not user_slug: + raise WindowSessionContextUnavailable("window_session_context_unavailable") + if ctx.durable_scope_kind == WINDOW_SESSION_SCOPE_PERSONAL: + recovered = replace( + ctx, + organisation_concept_id=None, + role_in_org=None, + namespace=derive_namespace(user_slug), + durable_recovered=False, + ) + else: + org_id = _normalise_owner_id(ctx.organisation_concept_id) + if not org_id: + raise WindowSessionContextUnavailable( + "window_session_context_unavailable" + ) + canonical_org_id = org_id if org_id.startswith("#V#") else f"#V#{org_id}" + canonical_user_id = user_id if user_id.startswith("#V#") else f"#V#{user_slug}" + try: + from .organisation_membership_service import ( + resolve_user_organisation_membership, + ) + from .ontology_authority_membership_coordination_service import ( + organisation_membership_scope_barrier, + ) + + # Keep membership read and derived-scope publication atomic with + # canonical role/membership mutations. Without the shared barrier, + # a recovery could recreate a binding between mutation invalidation + # and the represented write. + with organisation_membership_scope_barrier( + canonical_user_id, + canonical_org_id, + ): + membership = resolve_user_organisation_membership( + canonical_user_id, + canonical_org_id, + ) + if isinstance(membership, dict): + role = str(membership.get("role") or "member").strip() or "member" + recovered = replace( + ctx, + organisation_concept_id=canonical_org_id, + role_in_org=role, + namespace=derive_namespace(user_slug, canonical_org_id[3:]), + durable_recovered=False, + ) + get_window_session_store().set(recovered) + except Exception as exc: + if isinstance(exc, WindowSessionContextUnavailable): + raise + logger.warning( + "Unable to revalidate recovered window organisation membership: %s", + type(exc).__name__, + ) + raise WindowSessionContextRecoveryUnavailable( + "window_session_membership_recovery_unavailable" + ) from exc + if not isinstance(membership, dict): + try: + get_window_session_store().delete_if_owned( + ctx.window_session_id, user_id + ) + except WindowSessionBindingStoreUnavailable: + logger.warning( + "Unable to invalidate revoked durable window-session binding" + ) + raise WindowSessionContextUnavailable( + "window_session_context_unavailable" + ) + return recovered + get_window_session_store().set(recovered) + return recovered def get_effective_context( @@ -370,7 +720,19 @@ def _window_context_has_authoritative_scope(ctx: WindowSessionContext) -> bool: # Check window session first if window_session_id: - window_ctx = get_window_context(window_session_id) + try: + window_ctx = get_window_session_store().get_owned( + window_session_id, user_id + ) + except WindowSessionContextRecoveryUnavailable: + # Never let a durable-store outage fall through to another tab's + # browser-wide organisation scope. + raise + if window_ctx is not None and window_ctx.durable_recovered: + window_ctx = _recover_authoritative_scope( + window_ctx, + user_id=_normalise_owner_id(user_id) or "", + ) if window_ctx is not None: stored_owner = _normalise_owner_id(window_ctx.user_id) authenticated_owner = _normalise_owner_id(user_id) @@ -455,6 +817,18 @@ def delete_all_window_contexts_owned_by(user_id: Optional[str]) -> int: return get_window_session_store().delete_all_owned(user_id) +def delete_window_contexts_owned_by_user_for_organisation( + user_id: Optional[str], + organisation_concept_id: Optional[str], +) -> int: + """Invalidate derived tab authority for one exact actor/org pair.""" + + return get_window_session_store().delete_owned_for_organisation( + user_id, + organisation_concept_id, + ) + + def set_window_chat_session( window_session_id: str, chat_session_id: Optional[str], @@ -463,6 +837,6 @@ def set_window_chat_session( """Set the active chat session for a window.""" store = get_window_session_store() ctx = store.get_or_create(window_session_id, user_id) - ctx.chat_session_id = chat_session_id - store.set(ctx) - return ctx + updated = replace(ctx, chat_session_id=chat_session_id) + store.set(updated) + return updated diff --git a/src/frontend/web/von_interface/static/js/chatTab.js b/src/frontend/web/von_interface/static/js/chatTab.js index 6cbd3f8a..ecb3df69 100644 --- a/src/frontend/web/von_interface/static/js/chatTab.js +++ b/src/frontend/web/von_interface/static/js/chatTab.js @@ -33031,7 +33031,8 @@ async function copyActiveThinkingDiagnostics(button = null, requestOverride = nu includeLiveProgress: true }); const payload = buildThinkingDiagnosticsLocatorPayload(request, { - mcpAccess: turnAccess?.mcp_access || null + mcpAccess: turnAccess?.mcp_access || null, + retrievalStatus: turnAccess?.retrieval_status || null }); if (!payload) { showToast('No diagnostic reference is available yet.', 'info'); @@ -36655,7 +36656,11 @@ async function buildLlmDebugDeepInspectionJsonForTurn(turnId) { : null, includeLiveProgress: false }); - if (turnAccess?.mcp_access && typeof turnAccess.mcp_access === 'object') { + if ( + turnAccess?.mcp_access + && typeof turnAccess.mcp_access === 'object' + && Object.keys(turnAccess.mcp_access).length > 0 + ) { debugData = { ...debugData, telemetry_locator_mcp_access: turnAccess.mcp_access, @@ -37334,7 +37339,10 @@ function buildChatHistoryAccessArgs({ sessionId, historyIndex = undefined } = {} }; } -function buildThinkingDiagnosticsLocatorPayload(request, { mcpAccess = null } = {}) { +function buildThinkingDiagnosticsLocatorPayload( + request, + { mcpAccess = null, retrievalStatus = null } = {} +) { if (!request || typeof request !== 'object') { return null; } @@ -37366,6 +37374,17 @@ function buildThinkingDiagnosticsLocatorPayload(request, { mcpAccess = null } = ) : {}; + const cleanRetrievalStatus = typeof retrievalStatus === 'string' + ? retrievalStatus.trim() + : ''; + const resolvedRetrievalStatus = Object.keys(executableMcpAccess).length > 0 + ? 'server_delegation_available' + : ( + cleanRetrievalStatus === 'server_delegation_available' + ? 'server_delegation_empty' + : cleanRetrievalStatus || 'server_delegation_unavailable' + ); + return { schema_version: TURN_LIVE_PROGRESS_LOCATOR_SCHEMA_VERSION, generated_at_utc: new Date().toISOString(), @@ -37381,9 +37400,7 @@ function buildThinkingDiagnosticsLocatorPayload(request, { mcpAccess = null } = : null } : null, mcp_access: executableMcpAccess, - retrieval_status: Object.keys(executableMcpAccess).length > 0 - ? 'server_delegation_available' - : 'server_delegation_unavailable' + retrieval_status: resolvedRetrievalStatus }; } @@ -37776,17 +37793,47 @@ async function fetchTurnTelemetryMcpAccess({ timeoutMs: CONVERSATION_TELEMETRY_LOCATOR_FETCH_TIMEOUT_MS } ); - const body = await response.json(); - if (!response.ok || !body || typeof body !== 'object') { - return null; + if (!response || response.ok !== true) { + const status = Number(response?.status); + let retrievalStatus = 'server_request_rejected'; + if (status === 401) { + retrievalStatus = 'authentication_required'; + } else if (status === 403 || status === 404) { + // Deliberately keep absence and non-authorisation + // indistinguishable while explaining what the server actually + // reported. "Unavailable" previously implied a delegation + // service outage even when this exact turn simply had no row. + retrievalStatus = 'not_found_or_not_authorised'; + } else if (status === 409) { + retrievalStatus = 'window_scope_unavailable'; + } else if (status === 429) { + retrievalStatus = 'server_rate_limited'; + } else if (status >= 500) { + retrievalStatus = 'server_temporarily_unavailable'; + } + return { mcp_access: {}, retrieval_status: retrievalStatus }; + } + let body; + try { + body = await response.json(); + } catch (_error) { + return { mcp_access: {}, retrieval_status: 'invalid_server_response' }; + } + if (!body || typeof body !== 'object') { + return { mcp_access: {}, retrieval_status: 'invalid_server_response' }; } if (body.schema_version !== TURN_TELEMETRY_MCP_ACCESS_SCHEMA_VERSION) { - return null; + return { mcp_access: {}, retrieval_status: 'invalid_server_response' }; } - return body; + return { ...body, retrieval_status: 'server_delegation_available' }; } catch (error) { console.warn('[chatTab] Failed to fetch server turn telemetry delegation:', error); - return null; + return { + mcp_access: {}, + retrieval_status: error?.vonTimeout === true + ? 'server_timeout' + : 'server_unreachable' + }; } } diff --git a/src/frontend/web/von_interface/static/js/test/chatTab.test.js b/src/frontend/web/von_interface/static/js/test/chatTab.test.js index 93edc55a..420841d8 100644 --- a/src/frontend/web/von_interface/static/js/test/chatTab.test.js +++ b/src/frontend/web/von_interface/static/js/test/chatTab.test.js @@ -8245,7 +8245,7 @@ describe('thinking card toggle accessibility', () => { schema_version: 'turn_live_progress_locator.v1', request_id: expect.any(String), mcp_access: {}, - retrieval_status: 'server_delegation_unavailable' + retrieval_status: 'invalid_server_response' })); retained.toggleButton.click(); diff --git a/tests/backend/test_access_control_identity_resolution.py b/tests/backend/test_access_control_identity_resolution.py index 66ac56bf..50bbc525 100644 --- a/tests/backend/test_access_control_identity_resolution.py +++ b/tests/backend/test_access_control_identity_resolution.py @@ -427,21 +427,17 @@ def test_window_session_organisation_context_controls_visibility(monkeypatch) -> import src.backend.security.access_control as access_control import src.backend.services.window_session_context_service as window_context from src.backend.services.window_session_context_service import ( - WindowSessionContext, - get_window_session_store, + set_window_organisation, ) from flask import session window_context._window_session_store = None - store = get_window_session_store() - store.set( - WindowSessionContext( - window_session_id="ws_sail", - user_id="#V#michael_witbrock", - organisation_concept_id="university_of_auckland_strong_ai_lab", - namespace="#V#michael_witbrock@university_of_auckland_strong_ai_lab", - role_in_org="member", - ) + set_window_organisation( + "ws_sail", + "university_of_auckland_strong_ai_lab", + "member", + "#V#michael_witbrock@university_of_auckland_strong_ai_lab", + "#V#michael_witbrock", ) client = mongomock.MongoClient() diff --git a/tests/backend/test_browser_test_auth_routes.py b/tests/backend/test_browser_test_auth_routes.py index d2fd72c7..379a936c 100644 --- a/tests/backend/test_browser_test_auth_routes.py +++ b/tests/backend/test_browser_test_auth_routes.py @@ -211,3 +211,45 @@ def test_logout_invalidates_owned_window_context(monkeypatch, app_client): assert response.status_code == 200 assert store.get("owned_window") is None + + +def test_logout_clears_auth_when_durable_window_cleanup_is_unavailable( + monkeypatch, app_client +): + _, client = app_client + from src.backend.services import window_session_context_service as window_service + from src.backend.services.window_session_binding_store_service import ( + WindowSessionBindingStoreUnavailable, + ) + + class UnavailableDeleteRepository: + def delete_owned(self, *_args, **_kwargs): + raise WindowSessionBindingStoreUnavailable("unavailable") + + store = window_service.WindowSessionStore( + binding_repository=UnavailableDeleteRepository() # type: ignore[arg-type] + ) + monkeypatch.setattr(window_service, "_window_session_store", store) + store.set( + window_service.WindowSessionContext( + window_session_id="owned_window_store_down", + user_id="#V#user_a", + organisation_concept_id="#V#secret_org_a", + namespace="#V#user_a@secret_org_a", + ) + ) + with client.session_transaction() as sess: + sess["user_concept_id"] = "#V#user_a" + sess["user_email"] = "user-a@example.test" + + response = client.post( + "/von/api/auth/logout", + headers={"X-Von-Window-Session": "owned_window_store_down"}, + ) + + assert response.status_code == 200 + assert response.get_json()["window_context_cleanup"] == "deferred" + with client.session_transaction() as sess: + assert "user_concept_id" not in sess + assert "user_email" not in sess + assert store.get("owned_window_store_down") is None diff --git a/tests/backend/test_chat_prompt_queue_routes.py b/tests/backend/test_chat_prompt_queue_routes.py index 4065ade6..192ef44d 100644 --- a/tests/backend/test_chat_prompt_queue_routes.py +++ b/tests/backend/test_chat_prompt_queue_routes.py @@ -117,6 +117,108 @@ def test_chat_prompt_queue_create_returns_typed_backpressure( } +def test_legacy_active_create_is_idempotent_but_explicit_attempts_are_not( + client, +) -> None: + from src.backend.services.window_session_context_service import ( + set_window_organisation, + ) + + set_window_organisation( + "legacy-window", + "#V#test_org", + "member", + "#V#test_user@test_org", + user_id="#V#test_user", + ) + headers = {"X-Von-Window-Session": "legacy-window"} + legacy_payload = { + "prompt_raw": "legacy active prompt", + "session_id": "legacy-session", + "status": "in_progress", + } + legacy_first = client.post( + "/von/api/chat_prompt_queue", + json=legacy_payload, + headers=headers, + ) + legacy_second = client.post( + "/von/api/chat_prompt_queue", + json=legacy_payload, + headers=headers, + ) + + assert legacy_first.status_code == 201 + assert legacy_second.status_code == 201 + legacy_item = legacy_first.get_json()["item"] + legacy_queue_id = legacy_item["queue_id"] + assert legacy_second.get_json()["item"]["queue_id"] == legacy_queue_id + assert "active_legacy_submission_key" not in legacy_item + assert "legacy_submission_queue_seen" not in legacy_item + assert "legacy_submission_generate_seen" not in legacy_item + assert "legacy_submission_expires_at" not in legacy_item + + explicit_queue_ids = [] + for index in range(2): + response = client.post( + "/von/api/chat_prompt_queue", + json={ + **legacy_payload, + "client_request_id": f"request-{index}", + "attempt_id": f"attempt-{index}", + }, + headers=headers, + ) + assert response.status_code == 201 + explicit_queue_ids.append(response.get_json()["item"]["queue_id"]) + + assert explicit_queue_ids[0] != explicit_queue_ids[1] + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + assert coll.count_documents({}) == 3 + assert ( + coll.count_documents({"active_legacy_submission_key": {"$exists": True}}) == 1 + ) + + +def test_legacy_active_create_isolated_by_trusted_window_session(client) -> None: + from src.backend.services.window_session_context_service import ( + set_window_organisation, + ) + + for window_session_id in ("legacy-window-a", "legacy-window-b"): + set_window_organisation( + window_session_id, + "#V#test_org", + "member", + "#V#test_user@test_org", + user_id="#V#test_user", + ) + + payload = { + "prompt_raw": "same prompt in two tabs", + "session_id": "same-conversation", + "status": "in_progress", + } + first = client.post( + "/von/api/chat_prompt_queue", + json=payload, + headers={"X-Von-Window-Session": "legacy-window-a"}, + ) + second = client.post( + "/von/api/chat_prompt_queue", + json=payload, + headers={"X-Von-Window-Session": "legacy-window-b"}, + ) + + assert first.status_code == 201 + assert second.status_code == 201 + assert first.get_json()["item"]["queue_id"] != second.get_json()["item"]["queue_id"] + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + assert coll.count_documents({}) == 2 + + def test_queue_session_update_recomputes_and_clears_canonical_conversation_key( client, ) -> None: diff --git a/tests/backend/test_chat_prompt_queue_service.py b/tests/backend/test_chat_prompt_queue_service.py index 0875723d..61a152a0 100644 --- a/tests/backend/test_chat_prompt_queue_service.py +++ b/tests/backend/test_chat_prompt_queue_service.py @@ -1,5 +1,7 @@ from __future__ import annotations +import threading +from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timedelta, timezone import pytest @@ -332,6 +334,334 @@ def test_queue_scope_isolation() -> None: assert queue_service.list_active_queue_records(scope=second_scope) == [] +def test_active_legacy_submission_reuses_same_half_and_joins_after_terminal() -> None: + scope = queue_service.build_queue_scope( + user_concept_id="#V#user", + organisation_concept_id="#V#org", + namespace="#V#user@org", + ) + create_kwargs = { + "scope": scope, + "prompt_raw": "Run the same active prompt", + "session_id": "session-a", + "conversation_key": "canonical-conversation-a", + "window_session_id": "window-a", + } + + queue_first = queue_service.create_queue_record( + **create_kwargs, + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + ) + queue_retry = queue_service.create_queue_record( + **create_kwargs, + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + ) + + assert queue_retry["queue_id"] == queue_first["queue_id"] + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + assert coll.count_documents({}) == 1 + persisted = coll.find_one({"queue_id": queue_first["queue_id"]}) + assert persisted is not None + active_key = persisted["active_legacy_submission_key"] + assert len(active_key) == 64 + assert "Run the same active prompt" not in active_key + + queue_service.finish_prompt_record( + scope=scope, + queue_id=queue_first["queue_id"], + status=queue_service.STATUS_COMPLETED, + ) + terminal = coll.find_one({"queue_id": queue_first["queue_id"]}) + assert terminal is not None + assert "active_legacy_submission_key" in terminal + assert terminal["legacy_submission_queue_seen"] is True + assert terminal["legacy_submission_generate_seen"] is False + + generate_second = queue_service.create_queue_record( + **create_kwargs, + client_request_id="request-a", + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_GENERATE, + ) + assert generate_second["queue_id"] == queue_first["queue_id"] + joined = coll.find_one({"queue_id": queue_first["queue_id"]}) + assert joined is not None + assert "active_legacy_submission_key" not in joined + assert "legacy_submission_expires_at" not in joined + assert joined["legacy_submission_queue_seen"] is True + assert joined["legacy_submission_generate_seen"] is True + + repeated_finish = queue_service.finish_prompt_record( + scope=scope, + queue_id=queue_first["queue_id"], + status=queue_service.STATUS_COMPLETED, + ) + assert repeated_finish["queue_id"] == queue_first["queue_id"] + with pytest.raises(queue_service.ChatPromptQueueRecordNotFound): + queue_service.finish_prompt_record( + scope=scope, + queue_id=queue_first["queue_id"], + status=queue_service.STATUS_FAILED, + ) + + next_submission = queue_service.create_queue_record( + **create_kwargs, + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + ) + assert next_submission["queue_id"] != queue_first["queue_id"] + assert coll.count_documents({}) == 2 + + +def test_legacy_queue_and_generate_duplicate_key_race_converges_atomically( + monkeypatch: pytest.MonkeyPatch, +) -> None: + scope = queue_service.build_queue_scope( + user_concept_id="#V#user", + organisation_concept_id="#V#org", + namespace="#V#user@org", + ) + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + original_insert_one = coll.insert_one + insert_barrier = threading.Barrier(2) + + def racing_insert(document, *args, **kwargs): + insert_barrier.wait(timeout=2) + return original_insert_one(document, *args, **kwargs) + + monkeypatch.setattr(coll, "insert_one", racing_insert) + common = { + "scope": scope, + "session_id": "legacy-race-session", + "conversation_key": "canonical-legacy-race-conversation", + "window_session_id": "legacy-race-window", + } + + with ThreadPoolExecutor(max_workers=2) as executor: + queue_future = executor.submit( + queue_service.create_queue_record, + **common, + prompt_raw=" Race #V\u200b#prompt ", + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + ) + generate_future = executor.submit( + queue_service.create_queue_record, + **common, + prompt_raw="Race #V#prompt", + status=queue_service.STATUS_IN_PROGRESS, + client_request_id="legacy-race-request", + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_GENERATE, + ) + queue_record = queue_future.result(timeout=3) + generate_record = generate_future.result(timeout=3) + + assert queue_record["queue_id"] == generate_record["queue_id"] + assert coll.count_documents({}) == 1 + persisted = coll.find_one({"queue_id": queue_record["queue_id"]}) + assert persisted is not None + assert persisted["legacy_submission_queue_seen"] is True + assert persisted["legacy_submission_generate_seen"] is True + assert "active_legacy_submission_key" not in persisted + + +def test_unpaired_legacy_submission_rendezvous_expires_without_deleting_record( + monkeypatch: pytest.MonkeyPatch, +) -> None: + scope = queue_service.build_queue_scope( + user_concept_id="#V#user", + organisation_concept_id="#V#org", + namespace="#V#user@org", + ) + clock = [datetime(2026, 8, 27, 12, 0, tzinfo=timezone.utc)] + monkeypatch.setattr(queue_service, "_now", lambda: clock[0]) + create_kwargs = { + "scope": scope, + "prompt_raw": "Retry after a lost generate half", + "session_id": "session-expiry", + "conversation_key": "canonical-conversation-expiry", + "legacy_submission_role": queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + "window_session_id": "window-expiry", + } + + expired = queue_service.create_queue_record(**create_kwargs) + clock[0] += timedelta( + seconds=queue_service.LEGACY_SUBMISSION_RENDEZVOUS_SECONDS + 1 + ) + replacement = queue_service.create_queue_record(**create_kwargs) + + assert replacement["queue_id"] != expired["queue_id"] + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + expired_doc = coll.find_one({"queue_id": expired["queue_id"]}) + replacement_doc = coll.find_one({"queue_id": replacement["queue_id"]}) + assert expired_doc is not None + assert expired_doc["status"] == queue_service.STATUS_QUEUED + assert "active_legacy_submission_key" not in expired_doc + assert "legacy_submission_queue_seen" not in expired_doc + assert "legacy_submission_generate_seen" not in expired_doc + assert "legacy_submission_expires_at" not in expired_doc + assert replacement_doc is not None + assert "active_legacy_submission_key" in replacement_doc + + +def test_terminal_unpaired_generate_allows_new_client_request() -> None: + scope = queue_service.build_queue_scope( + user_concept_id="#V#user", + organisation_concept_id="#V#org", + namespace="#V#user@org", + ) + create_kwargs = { + "scope": scope, + "prompt_raw": "Legitimately send this prompt again", + "session_id": "session-new-generate", + "conversation_key": "canonical-conversation-new-generate", + "legacy_submission_role": queue_service.LEGACY_SUBMISSION_ROLE_GENERATE, + "window_session_id": "window-new-generate", + } + first = queue_service.create_queue_record( + **create_kwargs, + client_request_id="request-a", + ) + + with pytest.raises(queue_service.ConversationTurnAlreadyActive): + queue_service.create_queue_record( + **create_kwargs, + client_request_id="request-b", + ) + + queue_service.finish_prompt_record( + scope=scope, + queue_id=first["queue_id"], + status=queue_service.STATUS_COMPLETED, + ) + second = queue_service.create_queue_record( + **create_kwargs, + client_request_id="request-b", + ) + + assert second["queue_id"] != first["queue_id"] + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + old_doc = coll.find_one({"queue_id": first["queue_id"]}) + new_doc = coll.find_one({"queue_id": second["queue_id"]}) + assert old_doc is not None + assert "active_legacy_submission_key" not in old_doc + assert "legacy_submission_queue_seen" not in old_doc + assert "legacy_submission_generate_seen" not in old_doc + assert new_doc is not None + assert new_doc["legacy_submission_generate_seen"] is True + assert new_doc["legacy_submission_queue_seen"] is False + assert "active_legacy_submission_key" in new_doc + + +def test_active_legacy_submission_key_isolates_actor_org_conversation_and_prompt( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setenv("VON_MAX_QUEUED_TURNS_PER_USER", "20") + cases = [ + ( + queue_service.build_queue_scope( + user_concept_id="#V#actor_a", + organisation_concept_id="#V#org_a", + namespace="#V#actor_a@org_a", + ), + "conversation-a", + "same prompt", + "window-a", + ), + ( + queue_service.build_queue_scope( + user_concept_id="#V#actor_b", + organisation_concept_id="#V#org_a", + namespace="#V#actor_b@org_a", + ), + "conversation-a", + "same prompt", + "window-a", + ), + ( + queue_service.build_queue_scope( + user_concept_id="#V#actor_a", + organisation_concept_id="#V#org_b", + namespace="#V#actor_a@org_b", + ), + "conversation-a", + "same prompt", + "window-a", + ), + ( + queue_service.build_queue_scope( + user_concept_id="#V#actor_a", + organisation_concept_id="#V#org_a", + namespace="#V#actor_a@org_a", + ), + "conversation-b", + "same prompt", + "window-a", + ), + ( + queue_service.build_queue_scope( + user_concept_id="#V#actor_a", + organisation_concept_id="#V#org_a", + namespace="#V#actor_a@org_a", + ), + "conversation-a", + "different prompt", + "window-a", + ), + ( + queue_service.build_queue_scope( + user_concept_id="#V#actor_a", + organisation_concept_id="#V#org_a", + namespace="#V#actor_a@org_a", + ), + "conversation-a", + "same prompt", + "window-b", + ), + ] + + records = [ + queue_service.create_queue_record( + scope=scope, + prompt_raw=prompt, + session_id=conversation_key, + conversation_key=conversation_key, + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + window_session_id=window_session_id, + ) + for scope, conversation_key, prompt, window_session_id in cases + ] + + assert len({record["queue_id"] for record in records}) == len(cases) + + +def test_explicit_attempts_do_not_use_legacy_submission_correlation() -> None: + scope = queue_service.build_queue_scope( + user_concept_id="#V#user", + organisation_concept_id="#V#org", + namespace="#V#user@org", + ) + records = [ + queue_service.create_queue_record( + scope=scope, + prompt_raw="Repeat explicitly", + session_id="session-a", + conversation_key="canonical-conversation-a", + client_request_id=f"request-{index}", + attempt_id=f"attempt-{index}", + ) + for index in range(2) + ] + + assert records[0]["queue_id"] != records[1]["queue_id"] + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + assert ( + coll.count_documents({"active_legacy_submission_key": {"$exists": True}}) == 0 + ) + + def test_queue_scope_canonicalises_org_and_namespace_components() -> None: scope = queue_service.build_queue_scope( user_concept_id="user", diff --git a/tests/backend/test_conversation_turn_admission_service.py b/tests/backend/test_conversation_turn_admission_service.py index 6f6e4992..5a4eb22e 100644 --- a/tests/backend/test_conversation_turn_admission_service.py +++ b/tests/backend/test_conversation_turn_admission_service.py @@ -106,6 +106,107 @@ def test_distinct_conversations_run_concurrently_and_release_exactly() -> None: assert service.snapshot()["active_global"] == 0 +def test_legacy_queue_first_converges_with_generate_admission() -> None: + scope = _scope() + conversation_key = _key("legacy-queue-first") + queue_first = queue_service.create_queue_record( + scope=scope, + prompt_raw="queue first", + session_id="legacy-queue-first", + conversation_key=conversation_key, + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + window_session_id="window-queue-first", + ) + service = ConversationTurnAdmissionService(per_user_limit=2, global_limit=2) + + token = service.acquire( + scope=scope, + prompt_raw="queue first", + session_id="legacy-queue-first", + session_name=None, + client_request_id="request-queue-first", + conversation_key=conversation_key, + window_session_id="window-queue-first", + ) + + assert token.queue_id == queue_first["queue_id"] + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + assert coll.count_documents({}) == 1 + service.release(token, status=queue_service.STATUS_COMPLETED) + persisted = coll.find_one({"queue_id": token.queue_id}) + assert persisted is not None + assert "active_legacy_submission_key" not in persisted + + +def test_legacy_generate_first_converges_with_queue_create() -> None: + scope = _scope() + conversation_key = _key("legacy-generate-first") + service = ConversationTurnAdmissionService(per_user_limit=2, global_limit=2) + + token = service.acquire( + scope=scope, + prompt_raw="generate first", + session_id="legacy-generate-first", + session_name=None, + client_request_id="request-generate-first", + conversation_key=conversation_key, + window_session_id="window-generate-first", + ) + service.release(token, status=queue_service.STATUS_COMPLETED) + queue_second = queue_service.create_queue_record( + scope=scope, + prompt_raw="generate first", + session_id="legacy-generate-first", + conversation_key=conversation_key, + legacy_submission_role=queue_service.LEGACY_SUBMISSION_ROLE_QUEUE, + window_session_id="window-generate-first", + ) + + assert queue_second["queue_id"] == token.queue_id + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + assert coll.count_documents({}) == 1 + persisted = coll.find_one({"queue_id": token.queue_id}) + assert persisted is not None + assert persisted["status"] == queue_service.STATUS_COMPLETED + assert "active_legacy_submission_key" not in persisted + + +def test_terminal_unpaired_generate_does_not_block_new_request_id() -> None: + scope = _scope() + conversation_key = _key("legacy-generate-repeat") + service = ConversationTurnAdmissionService(per_user_limit=2, global_limit=2) + acquire_kwargs = { + "scope": scope, + "prompt_raw": "send the same prompt again", + "session_id": "legacy-generate-repeat", + "session_name": None, + "conversation_key": conversation_key, + "window_session_id": "window-generate-repeat", + } + + first = service.acquire( + **acquire_kwargs, + client_request_id="request-generate-first", + ) + service.release(first, status=queue_service.STATUS_COMPLETED) + second = service.acquire( + **acquire_kwargs, + client_request_id="request-generate-second", + ) + + assert second.queue_id != first.queue_id + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + assert coll.count_documents({}) == 2 + old_record = coll.find_one({"queue_id": first.queue_id}) + assert old_record is not None + assert old_record["status"] == queue_service.STATUS_COMPLETED + assert "active_legacy_submission_key" not in old_record + service.release(second, status=queue_service.STATUS_COMPLETED) + + def test_release_retries_durable_terminalisation_before_freeing_capacity( monkeypatch: pytest.MonkeyPatch, ) -> None: diff --git a/tests/backend/test_mongo_collection_index_guards.py b/tests/backend/test_mongo_collection_index_guards.py index 94e30490..db8670a8 100644 --- a/tests/backend/test_mongo_collection_index_guards.py +++ b/tests/backend/test_mongo_collection_index_guards.py @@ -476,6 +476,43 @@ def test_relationship_extent_index_ensure_does_not_replace_named_unique_drift(): assert index.get("unique") is not True +def test_chat_prompt_queue_indexes_include_active_legacy_submission_fence() -> None: + coll = mongomock.MongoClient().db.chat_prompt_queue_indexes + + mc._ensure_chat_prompt_queue_indexes(coll) + + indexes = {index["name"]: index for index in coll.list_indexes()} + active_legacy = indexes["active_legacy_submission_key_unique"] + assert list(active_legacy["key"].items()) == [ + ("active_legacy_submission_key", mc.ASCENDING) + ] + assert active_legacy["unique"] is True + assert active_legacy["partialFilterExpression"] == { + "active_legacy_submission_key": { + "$exists": True, + "$type": "string", + } + } + assert list(indexes["legacy_submission_expires_at"]["key"].items()) == [ + ("legacy_submission_expires_at", mc.ASCENDING) + ] + + +def test_window_session_binding_indexes_support_owner_cleanup_and_expiry() -> None: + coll = mongomock.MongoClient().db.window_session_binding_indexes + + mc._ensure_window_session_binding_indexes(coll) + + indexes = {index["name"]: index for index in coll.list_indexes()} + assert list(indexes["user_id_1"]["key"].items()) == [ + ("user_id", mc.ASCENDING) + ] + assert list(indexes["expires_at_ttl"]["key"].items()) == [ + ("expires_at", mc.ASCENDING) + ] + assert indexes["expires_at_ttl"]["expireAfterSeconds"] == 0 + + def test_workflow_instance_indexes_include_atlas_claim_and_lookup_indexes(monkeypatch): from src.backend.workflows.durable import instance_manager diff --git a/tests/backend/test_organisation_membership_scope_coordination.py b/tests/backend/test_organisation_membership_scope_coordination.py new file mode 100644 index 00000000..1ceadf56 --- /dev/null +++ b/tests/backend/test_organisation_membership_scope_coordination.py @@ -0,0 +1,127 @@ +from __future__ import annotations + +import threading +from collections import defaultdict +from contextlib import contextmanager + +from src.backend.services import ( + ontology_authority_membership_coordination_service as coordination, +) +from src.backend.services.ontology_publication_authority_service import ( + OntologyMutationResourceBusy, +) + + +def _install_fake_resource_locks(monkeypatch): + locks: defaultdict[str, threading.Lock] = defaultdict(threading.Lock) + + @contextmanager + def fake_resource_lock(resource_key: str, **_kwargs): + lock = locks[resource_key] + if not lock.acquire(blocking=False): + raise OntologyMutationResourceBusy("ontology_mutation_resource_busy") + try: + yield + finally: + lock.release() + + monkeypatch.setattr( + coordination, + "ontology_mutation_resource_lock", + fake_resource_lock, + ) + return locks + + +def test_unrelated_actor_org_scope_recoveries_do_not_contend(monkeypatch) -> None: + _install_fake_resource_locks(monkeypatch) + all_entered = threading.Event() + release = threading.Event() + state_lock = threading.Lock() + entered: set[str] = set() + errors: list[BaseException] = [] + + def recover(label: str, user_id: str, org_id: str) -> None: + try: + with coordination.organisation_membership_scope_barrier( + user_id, + org_id, + ): + with state_lock: + entered.add(label) + if len(entered) == 3: + all_entered.set() + assert release.wait(timeout=2) + except Exception as exc: # noqa: BLE001 - asserted in parent thread + errors.append(exc) + + threads = [ + threading.Thread( + target=recover, + args=("first", "#V#actor_a", "#V#org_a"), + ), + threading.Thread( + target=recover, + args=("second", "#V#actor_a", "#V#org_b"), + ), + threading.Thread( + target=recover, + args=("third", "#V#actor_b", "#V#org_a"), + ), + ] + for thread in threads: + thread.start() + + assert all_entered.wait(timeout=2) + release.set() + for thread in threads: + thread.join(timeout=2) + + assert errors == [] + assert entered == {"first", "second", "third"} + assert all(not thread.is_alive() for thread in threads) + + +def test_same_actor_org_scope_waits_for_inflight_mutation(monkeypatch) -> None: + _install_fake_resource_locks(monkeypatch) + first_entered = threading.Event() + release_first = threading.Event() + second_entered = threading.Event() + errors: list[BaseException] = [] + + def first() -> None: + try: + with coordination.organisation_membership_scope_barrier( + "#V#actor", + "#V#org", + ): + first_entered.set() + assert release_first.wait(timeout=2) + except Exception as exc: # noqa: BLE001 - asserted in parent thread + errors.append(exc) + + def second() -> None: + try: + assert first_entered.wait(timeout=2) + with coordination.organisation_membership_scope_barrier( + "actor", + "org", + ): + second_entered.set() + except Exception as exc: # noqa: BLE001 - asserted in parent thread + errors.append(exc) + + first_thread = threading.Thread(target=first) + second_thread = threading.Thread(target=second) + first_thread.start() + second_thread.start() + assert first_entered.wait(timeout=2) + assert not second_entered.wait(timeout=0.1) + release_first.set() + assert second_entered.wait(timeout=2) + first_thread.join(timeout=2) + second_thread.join(timeout=2) + + assert errors == [] + assert not first_thread.is_alive() + assert not second_thread.is_alive() diff --git a/tests/backend/test_organisation_membership_service.py b/tests/backend/test_organisation_membership_service.py index aeac4140..a1c8c5e4 100644 --- a/tests/backend/test_organisation_membership_service.py +++ b/tests/backend/test_organisation_membership_service.py @@ -35,6 +35,24 @@ def mock_membership_authority_barrier(monkeypatch): "ontology_authority_membership_mutation_barrier", nullcontext, ) + monkeypatch.setattr( + "src.backend.services.organisation_membership_service." + "organisation_membership_scope_barrier", + lambda *_args, **_kwargs: nullcontext(), + ) + + +@pytest.fixture(autouse=True) +def mock_window_authority_invalidation(monkeypatch): + """Keep membership unit tests off window-session persistence.""" + + invalidated_users = [] + monkeypatch.setattr( + "src.backend.services.organisation_membership_service." + "_invalidate_user_window_authority", + lambda user_id, org_id: invalidated_users.append((user_id, org_id)), + ) + return invalidated_users @pytest.fixture @@ -98,6 +116,7 @@ def test_create_membership_success( mock_text_value_service, mock_access_control, mock_text_repos, + mock_window_authority_invalidation, ): """Test successful creation of a membership relationship.""" # Setup @@ -147,9 +166,14 @@ def test_create_membership_success( # Predicate is stored as namespaced concept identifier assert call_args[1]["predicate"] == "#V#hasRole" assert call_args[1]["text"] == f"{role}::{org_id}" + assert mock_window_authority_invalidation == [] def test_create_membership_already_exists( - self, mock_concepts_repo, mock_text_value_service, mock_access_control + self, + mock_concepts_repo, + mock_text_value_service, + mock_access_control, + mock_window_authority_invalidation, ): """Test creating a membership that already exists.""" user_id = "#V#michael_witbrock" @@ -179,6 +203,7 @@ def test_create_membership_already_exists( # Assert assert result["relationship_created"] is False mock_concepts_repo.mutate_relationship_edge.assert_not_called() + assert mock_window_authority_invalidation == [(user_id, org_id)] def test_create_membership_invalid_user_id(self, mock_access_control): """Test creation with invalid user_id.""" @@ -501,6 +526,7 @@ def test_update_role_success( mock_access_control, mock_text_value_service, mock_text_repos, + mock_window_authority_invalidation, ): """Test successful role update.""" user_id = "#V#michael_witbrock" @@ -534,6 +560,7 @@ def test_update_role_success( assert result["new_role"] == "admin" assert result["role_updated"] is True assert result["previous_role"] == "member" + assert mock_window_authority_invalidation == [(user_id, org_id)] def test_update_role_not_member( self, mock_concepts_repo, mock_access_control, mock_text_repos @@ -555,7 +582,11 @@ def test_update_role_not_member( update_user_role(user_id, org_id, "admin") def test_update_role_no_change( - self, mock_concepts_repo, mock_access_control, mock_text_repos + self, + mock_concepts_repo, + mock_access_control, + mock_text_repos, + mock_window_authority_invalidation, ): """Test role update when role hasn't changed.""" user_id = "#V#michael_witbrock" @@ -572,13 +603,14 @@ def test_update_role_no_change( "object_text_id": "tv_1", "context": {"organisation_id": org_id}, } - text_vals.find_one.return_value = {"text": "admin"} + text_vals.find_one.return_value = {"text": f"admin::{org_id}"} # Execute result = update_user_role(user_id, org_id, "admin") # Assert assert result["role_updated"] is False + assert mock_window_authority_invalidation == [] # --- Tests for remove_organisation_membership --- @@ -588,7 +620,12 @@ class TestRemoveOrganisationMembership: """Tests for removing memberships.""" def test_remove_membership_success( - self, mock_concepts_repo, mock_access_control, mock_text_repos, monkeypatch + self, + mock_concepts_repo, + mock_access_control, + mock_text_repos, + mock_window_authority_invalidation, + monkeypatch, ): """Test successful removal of membership.""" user_id = "#V#michael_witbrock" @@ -618,6 +655,7 @@ def test_remove_membership_success( # Verify mutate_relationship_edge was called mock_concepts_repo.mutate_relationship_edge.assert_called_once() + assert mock_window_authority_invalidation == [(user_id, org_id)] def test_remove_membership_requires_semantic_authority_revocation_first( self, mock_concepts_repo, mock_access_control, mock_text_repos, monkeypatch diff --git a/tests/backend/test_session_org_routes.py b/tests/backend/test_session_org_routes.py index 1aded058..f5003c17 100644 --- a/tests/backend/test_session_org_routes.py +++ b/tests/backend/test_session_org_routes.py @@ -7,6 +7,8 @@ from __future__ import annotations +from contextlib import contextmanager + import pytest @@ -177,6 +179,47 @@ def test_set_organisation_updates_session_and_namespace(app_client): ) +def test_legacy_session_user_id_owns_durable_window_binding_canonically( + monkeypatch, + app_client, +): + from src.backend.services import window_session_context_service as window_context + + _, client = app_client + window_session_id = "legacy-session-owner-window" + with client.session_transaction() as sess: + sess["user_id"] = "michael_witbrock" + + selected = client.post( + "/von/api/session/set_organisation", + json={"organisation_concept_id": "university_of_auckland_strong_ai_lab"}, + headers={"X-Von-Window-Session": window_session_id}, + ) + assert selected.status_code == 200 + + repository = window_context.get_window_session_binding_repository() + binding = repository.load_owned(window_session_id, "#V#michael_witbrock") + assert binding is not None + assert binding.user_id == "#V#michael_witbrock" + assert repository.load_owned(window_session_id, "michael_witbrock") is None + + monkeypatch.setattr( + window_context, + "_window_session_store", + window_context.WindowSessionStore(binding_repository=repository), + ) + recovered = client.get( + "/von/api/session/context", + headers={"X-Von-Window-Session": window_session_id}, + ) + + assert recovered.status_code == 200 + assert recovered.get_json()["context_source"] == "window_session" + assert recovered.get_json()["organisation_id"] == ( + "#V#university_of_auckland_strong_ai_lab" + ) + + def test_set_organisation_rejects_non_member_without_changing_session( monkeypatch, app_client ): @@ -216,6 +259,44 @@ def test_set_organisation_validates_body(app_client): assert resp.get_json()["error"] == "organisation_concept_id required" +def test_set_organisation_reports_scope_coordination_contention_as_retryable( + monkeypatch, + app_client, +): + from src.backend.services import ( + ontology_authority_membership_coordination_service as coordination, + ) + from src.backend.services.ontology_publication_authority_service import ( + OntologyMutationResourceBusy, + ) + + @contextmanager + def busy_scope(*_args, **_kwargs): + raise OntologyMutationResourceBusy("ontology_mutation_resource_busy") + yield # pragma: no cover + + monkeypatch.setattr( + coordination, + "organisation_membership_scope_barrier", + busy_scope, + ) + _, client = app_client + with client.session_transaction() as sess: + sess["user_id"] = "michael_witbrock" + + response = client.post( + "/von/api/session/set_organisation", + json={"organisation_concept_id": "university_of_auckland_strong_ai_lab"}, + ) + + assert response.status_code == 409 + assert response.get_json() == { + "error": "window_session_scope_coordination_busy", + "error_code": "window_session_scope_coordination_busy", + "retryable": True, + } + + def test_get_session_context_derives_namespace_without_org(app_client): _, client = app_client diff --git a/tests/backend/test_von_generate_background_submission.py b/tests/backend/test_von_generate_background_submission.py index 285402f3..694fc30f 100644 --- a/tests/backend/test_von_generate_background_submission.py +++ b/tests/backend/test_von_generate_background_submission.py @@ -1,11 +1,13 @@ from __future__ import annotations import json +from typing import Any, Mapping, cast import pytest from flask import Flask, jsonify, request, session -from typing import Any, Mapping, cast +from src.backend.db import mongo_client +from src.backend.services import chat_prompt_queue_service from src.backend.services.adaptive_turn_service import AdaptiveTurnResult from src.backend.services.background_task_service import BackgroundTaskCapacityReached from src.backend.services.tool_progress_store_service import ( @@ -589,6 +591,418 @@ def test_generate_passes_authorised_inputs_to_adaptive_turn(monkeypatch): } +def test_legacy_queue_first_then_generate_joins_normalised_prompt_record( + monkeypatch, +): + adaptive_turn = _StubAdaptiveTurn( + AdaptiveTurnResult( + response_text="ok", + extra_messages=(), + tool_invocations=(), + aux_llm_calls=(), + ) + ) + app = _make_app(monkeypatch, adaptive_turn, _CapturingTaskRegistry()) + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + coll.delete_many({}) + client = app.test_client() + with client.session_transaction() as flask_session: + flask_session["user_concept_id"] = "#V#test_user" + + conversation_id = "legacy-route-queue-first" + headers = {"X-Von-Window-Session": "window-123"} + queue_response = client.post( + "/von/api/chat_prompt_queue", + json={ + "prompt_raw": " Discuss #V\u200b#concept ", + "session_id": conversation_id, + "status": "in_progress", + }, + headers=headers, + ) + generate_response = client.post( + "/von/generate", + json={ + "prompt": "Discuss #V#concept", + "background": False, + "model": "gpt-5.4-nano", + "client_request_id": "legacy-route-queue-first-request", + "conversation_session_id": conversation_id, + }, + headers=headers, + ) + + assert queue_response.status_code == 201 + assert generate_response.status_code == 200 + queue_id = queue_response.get_json()["item"]["queue_id"] + assert coll.count_documents({}) == 1 + persisted = coll.find_one({"queue_id": queue_id}) + assert persisted is not None + assert persisted["status"] == chat_prompt_queue_service.STATUS_COMPLETED + assert persisted["client_request_id"] == "legacy-route-queue-first-request" + assert persisted["legacy_submission_queue_seen"] is True + assert persisted["legacy_submission_generate_seen"] is True + assert "active_legacy_submission_key" not in persisted + assert adaptive_turn.calls + + legacy_finish_response = client.post( + f"/von/api/chat_prompt_queue/{queue_id}/finish", + json={"status": "completed"}, + headers=headers, + ) + assert legacy_finish_response.status_code == 200 + assert legacy_finish_response.get_json()["item"]["queue_id"] == queue_id + + +def test_legacy_generate_first_stays_joinable_after_completion_and_then_reopens( + monkeypatch, +): + adaptive_turn = _StubAdaptiveTurn( + AdaptiveTurnResult( + response_text="ok", + extra_messages=(), + tool_invocations=(), + aux_llm_calls=(), + ) + ) + app = _make_app(monkeypatch, adaptive_turn, _CapturingTaskRegistry()) + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + coll.delete_many({}) + client = app.test_client() + with client.session_transaction() as flask_session: + flask_session["user_concept_id"] = "#V#test_user" + + conversation_id = "legacy-route-generate-first" + prompt = "Generate first and join later" + headers = {"X-Von-Window-Session": "window-123"} + generate_response = client.post( + "/von/generate", + json={ + "prompt": prompt, + "background": False, + "model": "gpt-5.4-nano", + "client_request_id": "legacy-route-generate-first-request", + "conversation_session_id": conversation_id, + }, + headers=headers, + ) + + assert generate_response.status_code == 200 + assert coll.count_documents({}) == 1 + generated = coll.find_one({}) + assert generated is not None + generated_queue_id = generated["queue_id"] + assert generated["status"] == chat_prompt_queue_service.STATUS_COMPLETED + assert generated["legacy_submission_generate_seen"] is True + assert generated["legacy_submission_queue_seen"] is False + assert "active_legacy_submission_key" in generated + + legacy_payload = { + "prompt_raw": prompt, + "session_id": conversation_id, + "status": "in_progress", + } + joined_response = client.post( + "/von/api/chat_prompt_queue", + json=legacy_payload, + headers=headers, + ) + + assert joined_response.status_code == 201 + assert joined_response.get_json()["item"]["queue_id"] == generated_queue_id + assert joined_response.get_json()["item"]["status"] == "completed" + assert coll.count_documents({}) == 1 + joined = coll.find_one({"queue_id": generated_queue_id}) + assert joined is not None + assert joined["legacy_submission_generate_seen"] is True + assert joined["legacy_submission_queue_seen"] is True + assert "active_legacy_submission_key" not in joined + + legacy_finish_response = client.post( + f"/von/api/chat_prompt_queue/{generated_queue_id}/finish", + json={"status": "completed"}, + headers=headers, + ) + assert legacy_finish_response.status_code == 200 + assert legacy_finish_response.get_json()["item"]["queue_id"] == generated_queue_id + + later_identical_response = client.post( + "/von/api/chat_prompt_queue", + json=legacy_payload, + headers=headers, + ) + assert later_identical_response.status_code == 201 + assert later_identical_response.get_json()["item"]["queue_id"] != generated_queue_id + assert coll.count_documents({}) == 2 + + +def test_pre_upgrade_stale_tab_recovers_scope_across_cold_process_and_joins_old_wire( + monkeypatch, +): + import mongomock + + from src.backend.services import organisation_membership_service + from src.backend.services import window_session_context_service as window_context + from src.backend.services.window_session_binding_store_service import ( + MongoWindowSessionBindingRepository, + ) + + adaptive_turn = _StubAdaptiveTurn( + AdaptiveTurnResult( + response_text="ok", + extra_messages=(), + tool_invocations=(), + aux_llm_calls=(), + ) + ) + app = _make_app(monkeypatch, adaptive_turn, _CapturingTaskRegistry()) + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + coll.delete_many({}) + + binding_collection = mongomock.MongoClient()["legacy_restart"]["bindings"] + binding_repository = MongoWindowSessionBindingRepository( + ttl_seconds=3_600, + collection_getter=lambda: binding_collection, + ) + first_cold_store = window_context.WindowSessionStore( + binding_repository=binding_repository + ) + monkeypatch.setattr(window_context, "_window_session_store", first_cold_store) + monkeypatch.setattr( + organisation_membership_service, + "resolve_user_organisation_membership", + lambda user_id, org_id: { + "user_concept_id": user_id, + "organisation_concept_id": org_id, + "role": "member", + }, + ) + + actor_user_id = "#V#test_user" + actor_org_id = "#V#test_org" + actor_namespace = "#V#test_user@test_org" + conversation_id = "pre-upgrade-stale-tab-conversation" + stale_window_id = "pre-upgrade-stale-window" + summary_calls: list[tuple[str, str]] = [] + + def owned_conversation_summary(user_id, session_id, **_kwargs): + summary_calls.append((user_id, session_id)) + return { + "session_id": session_id, + "namespace": actor_namespace, + "organisation_concept_id": actor_org_id, + } + + monkeypatch.setattr( + "src.backend.server.routes.von_routes.chat_history_service.get_chat_history_session_summary", + owned_conversation_summary, + ) + + client = app.test_client() + with client.session_transaction() as flask_session: + flask_session["user_concept_id"] = actor_user_id + flask_session["organisation_concept_id"] = "#V#wrong_flask_org" + flask_session["org_id"] = "#V#wrong_flask_org" + flask_session["namespace"] = "#V#test_user@wrong_flask_org" + flask_session["role_in_org"] = "admin" + + headers = {"X-Von-Window-Session": stale_window_id} + generate_response = client.post( + "/von/generate", + json={ + "prompt": "Recover #V#this stale tab", + "client_request_id": "pre-upgrade-stale-tab-request", + "conversation_session_id": conversation_id, + "user_id": "#V#untrusted_client_actor", + "org_id": "#V#wrong_flask_org", + "language": "en-NZ", + "model": "gpt-5.4-nano", + "presenter_mode": True, + "skip_buttonify": False, + "thinking_card_mode": "standard", + }, + headers=headers, + ) + + assert generate_response.status_code == 200 + assert summary_calls == [(actor_user_id, conversation_id)] + persisted_binding = binding_repository.load_owned(stale_window_id, actor_user_id) + assert persisted_binding is not None + assert persisted_binding.organisation_concept_id == actor_org_id + generated = coll.find_one({}) + assert generated is not None + generated_queue_id = generated["queue_id"] + assert generated["organisation_concept_id"] == actor_org_id + assert generated["namespace"] == actor_namespace + assert generated["legacy_submission_generate_seen"] is True + + second_cold_store = window_context.WindowSessionStore( + binding_repository=binding_repository + ) + monkeypatch.setattr(window_context, "_window_session_store", second_cold_store) + queue_response = client.post( + "/von/api/chat_prompt_queue", + json={ + "prompt_raw": " Recover #V\u200b#this stale tab ", + "session_id": conversation_id, + "session_name": "Recovered stale conversation", + "status": "in_progress", + }, + headers=headers, + ) + + assert queue_response.status_code == 201 + assert queue_response.get_json()["item"]["queue_id"] == generated_queue_id + assert coll.count_documents({}) == 1 + joined = coll.find_one({"queue_id": generated_queue_id}) + assert joined is not None + assert joined["organisation_concept_id"] == actor_org_id + assert joined["namespace"] == actor_namespace + assert joined["legacy_submission_queue_seen"] is True + assert joined["legacy_submission_generate_seen"] is True + assert "active_legacy_submission_key" not in joined + assert second_cold_store.count() == 1 + recovered_context = second_cold_store.get_owned(stale_window_id, actor_user_id) + assert recovered_context is not None + assert recovered_context.organisation_concept_id == actor_org_id + assert adaptive_turn.calls + + finish_response = client.post( + f"/von/api/chat_prompt_queue/{generated_queue_id}/finish", + json={"status": "completed"}, + headers=headers, + ) + assert finish_response.status_code == 200 + + +def test_durable_window_binding_scopes_old_wire_after_cold_process_restart( + monkeypatch, +): + import mongomock + + from src.backend.services import organisation_membership_service + from src.backend.services import window_session_context_service as window_context + from src.backend.services.window_session_binding_store_service import ( + MongoWindowSessionBindingRepository, + ) + + adaptive_turn = _StubAdaptiveTurn( + AdaptiveTurnResult( + response_text="ok", + extra_messages=(), + tool_invocations=(), + aux_llm_calls=(), + ) + ) + app = _make_app(monkeypatch, adaptive_turn, _CapturingTaskRegistry()) + coll = mongo_client.get_chat_prompt_queue_collection() + assert coll is not None + coll.delete_many({}) + + binding_collection = mongomock.MongoClient()["durable_restart"]["bindings"] + binding_repository = MongoWindowSessionBindingRepository( + ttl_seconds=3_600, + collection_getter=lambda: binding_collection, + ) + first_store = window_context.WindowSessionStore( + binding_repository=binding_repository + ) + monkeypatch.setattr(window_context, "_window_session_store", first_store) + monkeypatch.setattr( + organisation_membership_service, + "resolve_user_organisation_membership", + lambda user_id, org_id: { + "user_concept_id": user_id, + "organisation_concept_id": org_id, + "role": "member", + }, + ) + + actor_user_id = "#V#test_user" + actor_org_id = "#V#test_org" + actor_namespace = "#V#test_user@test_org" + window_session_id = "durably-bound-window" + conversation_id = "durably-bound-conversation" + window_context.set_window_organisation( + window_session_id, + actor_org_id, + "member", + actor_namespace, + user_id=actor_user_id, + ) + assert binding_collection.count_documents({}) == 1 + + restarted_store = window_context.WindowSessionStore( + binding_repository=binding_repository + ) + monkeypatch.setattr(window_context, "_window_session_store", restarted_store) + monkeypatch.setattr( + "src.backend.server.routes.von_routes.chat_history_service.get_chat_history_session_summary", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError("durable binding must resolve before conversation recovery") + ), + ) + + client = app.test_client() + with client.session_transaction() as flask_session: + flask_session["user_concept_id"] = actor_user_id + flask_session["organisation_concept_id"] = "#V#wrong_flask_org" + flask_session["namespace"] = "#V#test_user@wrong_flask_org" + flask_session["role_in_org"] = "admin" + + headers = {"X-Von-Window-Session": window_session_id} + queue_response = client.post( + "/von/api/chat_prompt_queue", + json={ + "prompt_raw": " Use #v\u200b#durable context ", + "session_id": conversation_id, + "session_name": "Durably bound conversation", + "status": "in_progress", + }, + headers=headers, + ) + generate_response = client.post( + "/von/generate", + json={ + "prompt": "Use #v#durable context", + "client_request_id": "durably-bound-old-wire-request", + "conversation_session_id": conversation_id, + "user_id": "#V#untrusted_client_actor", + "org_id": "#V#wrong_flask_org", + "language": "en-NZ", + "model": "gpt-5.4-nano", + "presenter_mode": True, + "skip_buttonify": False, + "thinking_card_mode": "standard", + }, + headers=headers, + ) + + assert queue_response.status_code == 201 + assert generate_response.status_code == 200 + queue_id = queue_response.get_json()["item"]["queue_id"] + assert coll.count_documents({}) == 1 + joined = coll.find_one({"queue_id": queue_id}) + assert joined is not None + assert joined["organisation_concept_id"] == actor_org_id + assert joined["namespace"] == actor_namespace + assert joined["legacy_submission_queue_seen"] is True + assert joined["legacy_submission_generate_seen"] is True + assert "active_legacy_submission_key" not in joined + assert restarted_store.count() == 1 + assert adaptive_turn.calls + + finish_response = client.post( + f"/von/api/chat_prompt_queue/{queue_id}/finish", + json={"status": "completed"}, + headers=headers, + ) + assert finish_response.status_code == 200 + + def test_generate_surfaces_adaptive_terminal_status_without_rejudging_it( monkeypatch, ): diff --git a/tests/backend/test_von_history_telemetry_locator_endpoint.py b/tests/backend/test_von_history_telemetry_locator_endpoint.py index db0b39ac..ac1e7718 100644 --- a/tests/backend/test_von_history_telemetry_locator_endpoint.py +++ b/tests/backend/test_von_history_telemetry_locator_endpoint.py @@ -183,6 +183,235 @@ def test_history_turn_telemetry_access_requires_authenticated_actor(monkeypatch) assert response.get_json()["error"] == "Not authenticated" +def test_history_turn_telemetry_access_rejects_unknown_explicit_window_before_lookup( + monkeypatch, +): + import src.backend.server.routes.von_routes as von_routes + + captured: dict[str, object] = {} + monkeypatch.setattr( + "src.backend.security.access_control.get_effective_user_concept_id", + lambda: "#V#actor_a", + ) + + def fail_closed_context(**kwargs): + captured.update(kwargs) + raise von_routes.WindowSessionContextUnavailable( + "window_session_context_unavailable" + ) + + monkeypatch.setattr( + von_routes, + "_get_effective_context_with_owned_conversation_recovery", + fail_closed_context, + ) + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_diagnostics_payload", + lambda **_kwargs: (_ for _ in ()).throw( + AssertionError("scope denial must happen before telemetry lookup") + ), + ) + + app = Flask(__name__) + app.secret_key = "test-secret" + app.register_blueprint(von_routes.von_bp, url_prefix="/von") + response = app.test_client().get( + "/von/history/turn_telemetry_access", + query_string={ + "request_id": "req-unknown-window", + "session_id": "session-owned-by-actor-a", + }, + headers={"X-Von-Window-Session": "unknown-window"}, + ) + + assert response.status_code == 409 + assert response.get_json() == { + "error": "window_context_unavailable", + "error_code": "window_context_unavailable", + "retryable": True, + } + assert captured["window_session_id"] == "unknown-window" + assert captured["user_concept_id"] == "#V#actor_a" + assert captured["conversation_session_id"] == "session-owned-by-actor-a" + + +def test_history_turn_telemetry_access_reports_transient_scope_recovery_failure( + monkeypatch, +): + import src.backend.server.routes.von_routes as von_routes + + monkeypatch.setattr( + "src.backend.security.access_control.get_effective_user_concept_id", + lambda: "#V#actor_a", + ) + monkeypatch.setattr( + von_routes, + "_get_effective_context_with_owned_conversation_recovery", + lambda **_kwargs: (_ for _ in ()).throw( + von_routes.WindowSessionContextRecoveryUnavailable( + "window_session_context_recovery_unavailable" + ) + ), + ) + + app = Flask(__name__) + app.secret_key = "test-secret" + app.register_blueprint(von_routes.von_bp, url_prefix="/von") + response = app.test_client().get( + "/von/history/turn_telemetry_access", + query_string={"request_id": "req-store-outage"}, + headers={"X-Von-Window-Session": "known-window"}, + ) + + assert response.status_code == 503 + assert response.get_json() == { + "error": "window_context_recovery_unavailable", + "error_code": "window_context_recovery_unavailable", + "retryable": True, + } + + +def test_history_turn_telemetry_access_reads_exact_actor_scope_live_progress( + monkeypatch, +): + import src.backend.server.routes.von_routes as von_routes + + actor_user_id = "#V#actor_a" + actor_org_id = "#V#org_a" + actor_namespace = "#V#actor_a@org_a" + request_id = "req-live-progress-actor-org-a" + actor_scope_key = von_routes._build_authenticated_tool_progress_scope_key( + user_concept_id=actor_user_id, + organisation_concept_id=actor_org_id, + namespace=actor_namespace, + ) + + monkeypatch.setattr( + "src.backend.security.access_control.get_effective_user_concept_id", + lambda: actor_user_id, + ) + monkeypatch.setattr( + von_routes, + "get_effective_context", + lambda *_args, **_kwargs: { + "namespace": actor_namespace, + "organisation_id": actor_org_id, + }, + ) + monkeypatch.setattr( + von_routes, + "_resolve_history_request_scope_hints", + lambda **_kwargs: (actor_namespace, actor_org_id), + ) + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_diagnostics_payload", + lambda **_kwargs: None, + ) + monkeypatch.setattr( + von_routes, + "fetch_tool_progress_state", + lambda **_kwargs: None, + ) + monkeypatch.setitem( + von_routes._TOOL_PROGRESS, + (actor_scope_key, request_id), + { + "request_id": request_id, + "session_id": "session-live-progress-org-a", + "status": "thinking", + "stage": "tool_execution", + "phase": "tool_execution", + "updated_at_epoch": 9_999_999_999.0, + "request_started_epoch": 9_999_999_998.0, + "elapsed_ms": 1_000, + }, + ) + + app = Flask(__name__) + app.secret_key = "test-secret" + app.register_blueprint(von_routes.von_bp, url_prefix="/von") + response = app.test_client().get( + "/von/history/turn_telemetry_access", + query_string={"request_id": request_id}, + headers={"X-Von-Window-Session": "window-org-a"}, + ) + + assert response.status_code == 200 + body = response.get_json() + assert body["schema_version"] == "turn_telemetry_mcp_access.v1" + assert "turn_execution_get_live_progress" in body["mcp_access"] + + +def test_history_turn_telemetry_access_does_not_cross_actor_org_scope( + monkeypatch, +): + import src.backend.server.routes.von_routes as von_routes + + actor_user_id = "#V#actor_a" + org_a_scope_key = von_routes._build_authenticated_tool_progress_scope_key( + user_concept_id=actor_user_id, + organisation_concept_id="#V#org_a", + namespace="#V#actor_a@org_a", + ) + org_b_scope_key = von_routes._build_authenticated_tool_progress_scope_key( + user_concept_id=actor_user_id, + organisation_concept_id="#V#org_b", + namespace="#V#actor_a@org_b", + ) + request_id = "req-live-progress-private-to-org-a" + assert org_a_scope_key != org_b_scope_key + + monkeypatch.setattr( + "src.backend.security.access_control.get_effective_user_concept_id", + lambda: actor_user_id, + ) + monkeypatch.setattr( + von_routes, + "get_effective_context", + lambda *_args, **_kwargs: { + "namespace": "#V#actor_a@org_b", + "organisation_id": "#V#org_b", + }, + ) + monkeypatch.setattr( + von_routes, + "_resolve_history_request_scope_hints", + lambda **_kwargs: ("#V#actor_a@org_b", "#V#org_b"), + ) + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_diagnostics_payload", + lambda **_kwargs: None, + ) + monkeypatch.setattr( + von_routes, + "fetch_tool_progress_state", + lambda **_kwargs: None, + ) + monkeypatch.setitem( + von_routes._TOOL_PROGRESS, + (org_a_scope_key, request_id), + { + "request_id": request_id, + "status": "thinking", + "stage": "tool_execution", + "phase": "tool_execution", + "updated_at_epoch": 9_999_999_999.0, + }, + ) + + app = Flask(__name__) + app.secret_key = "test-secret" + app.register_blueprint(von_routes.von_bp, url_prefix="/von") + response = app.test_client().get( + "/von/history/turn_telemetry_access", + query_string={"request_id": request_id}, + headers={"X-Von-Window-Session": "window-org-b"}, + ) + + assert response.status_code == 404 + assert response.get_json() == {"error": "Turn telemetry not found"} + + def _stored_failure_capsule(request_id: str) -> dict: return { "schema_version": "turn_failure_capsule.v1", diff --git a/tests/backend/test_window_session_context_persistence.py b/tests/backend/test_window_session_context_persistence.py new file mode 100644 index 00000000..15ed3557 --- /dev/null +++ b/tests/backend/test_window_session_context_persistence.py @@ -0,0 +1,608 @@ +from __future__ import annotations + +import threading +from contextlib import contextmanager, nullcontext +from datetime import UTC, datetime, timedelta + +import mongomock +import pytest +from flask import Flask + +from src.backend.services import organisation_membership_service +from src.backend.services import window_session_context_service as window_context +from src.backend.services.window_session_binding_store_service import ( + MongoWindowSessionBindingRepository, + WindowSessionBindingStoreUnavailable, +) + + +@pytest.fixture() +def durable_store(monkeypatch: pytest.MonkeyPatch): + from src.backend.services import ( + ontology_authority_membership_coordination_service as coordination, + ) + + monkeypatch.setattr( + coordination, + "ontology_authority_membership_mutation_barrier", + nullcontext, + ) + monkeypatch.setattr( + coordination, + "organisation_membership_scope_barrier", + lambda *_args, **_kwargs: nullcontext(), + ) + collection = mongomock.MongoClient()["window_binding_test"]["bindings"] + repository = MongoWindowSessionBindingRepository( + ttl_seconds=3600, + collection_getter=lambda: collection, + ) + store = window_context.WindowSessionStore(binding_repository=repository) + monkeypatch.setattr(window_context, "_window_session_store", store) + monkeypatch.setattr( + organisation_membership_service, + "resolve_user_organisation_membership", + lambda user_id, org_id: { + "user_concept_id": user_id, + "organisation_concept_id": org_id, + "role": "admin" if org_id == "#V#lab" else "member", + }, + ) + return collection, repository + + +def _restart_with_repository( + monkeypatch: pytest.MonkeyPatch, + repository: MongoWindowSessionBindingRepository, +) -> window_context.WindowSessionStore: + restarted = window_context.WindowSessionStore(binding_repository=repository) + monkeypatch.setattr(window_context, "_window_session_store", restarted) + return restarted + + +def test_restart_recovers_org_and_revalidates_current_role( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + collection, repository = durable_store + window_context.set_window_organisation( + "ws-restart-org", + "lab", + "member", + "#V#actor@lab", + "#V#actor", + ) + + document = collection.find_one({}) + assert document is not None + assert document["_id"] != "ws-restart-org" + assert document["organisation_concept_id"] == "#V#lab" + assert "role_in_org" not in document + assert "namespace" not in document + + restarted = _restart_with_repository(monkeypatch, repository) + effective = window_context.get_effective_context( + "ws-restart-org", + {"organisation_concept_id": "wrong_flask_org"}, + "#V#actor", + require_known_window=True, + ) + + assert effective == { + "user_id": "#V#actor", + "organisation_id": "#V#lab", + "role": "admin", + "namespace": "#V#actor@lab", + "chat_session_id": None, + "source": "window_session", + } + assert restarted.count() == 1 + + +def test_restart_preserves_two_org_tabs_and_authoritative_personal_tab( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + _, repository = durable_store + window_context.set_window_organisation( + "ws-org-a", "org_a", "member", "#V#actor@org_a", "#V#actor" + ) + window_context.set_window_organisation( + "ws-org-b", "org_b", "member", "#V#actor@org_b", "#V#actor" + ) + window_context.clear_window_organisation("ws-personal", "#V#actor", "#V#actor") + _restart_with_repository(monkeypatch, repository) + + contexts = { + window_id: window_context.get_effective_context( + window_id, + { + "organisation_concept_id": "wrong_flask_org", + "namespace": "#V#actor@wrong_flask_org", + }, + "#V#actor", + require_known_window=True, + ) + for window_id in ("ws-org-a", "ws-org-b", "ws-personal") + } + + assert contexts["ws-org-a"]["namespace"] == "#V#actor@org_a" + assert contexts["ws-org-b"]["namespace"] == "#V#actor@org_b" + assert contexts["ws-personal"]["organisation_id"] is None + assert contexts["ws-personal"]["namespace"] == "#V#actor" + assert contexts["ws-personal"]["source"] == "window_session" + + +def test_warm_worker_observes_org_switch_from_shared_binding(durable_store) -> None: + _collection, repository = durable_store + first_worker = window_context.WindowSessionStore(binding_repository=repository) + second_worker = window_context.WindowSessionStore(binding_repository=repository) + + initial = window_context.WindowSessionContext( + window_session_id="ws-cross-worker", + user_id="#V#actor", + organisation_concept_id="#V#org_a", + role_in_org="member", + namespace="#V#actor@org_a", + ) + first_worker.persist_authoritative_binding( + initial, + scope_kind=window_context.WINDOW_SESSION_SCOPE_ORGANISATION, + ) + first_worker.set(initial) + cached = second_worker.get_owned("ws-cross-worker", "#V#actor") + assert cached is not None + cached.role_in_org = "member" + cached.namespace = "#V#actor@org_a" + cached.durable_recovered = False + second_worker.set(cached) + + switched = window_context.WindowSessionContext( + window_session_id="ws-cross-worker", + user_id="#V#actor", + organisation_concept_id="#V#org_b", + role_in_org="admin", + namespace="#V#actor@org_b", + ) + first_worker.persist_authoritative_binding( + switched, + scope_kind=window_context.WINDOW_SESSION_SCOPE_ORGANISATION, + ) + first_worker.set(switched) + + observed = second_worker.get_owned("ws-cross-worker", "#V#actor") + assert observed is not None + assert observed.organisation_concept_id == "#V#org_b" + assert observed.namespace is None + assert observed.role_in_org is None + assert observed.durable_recovered is True + + +def test_warm_worker_rejects_shared_binding_invalidation(durable_store) -> None: + _collection, repository = durable_store + first_worker = window_context.WindowSessionStore(binding_repository=repository) + second_worker = window_context.WindowSessionStore(binding_repository=repository) + selected = window_context.WindowSessionContext( + window_session_id="ws-cross-worker-revoked", + user_id="#V#actor", + organisation_concept_id="#V#org", + role_in_org="member", + namespace="#V#actor@org", + ) + first_worker.persist_authoritative_binding( + selected, + scope_kind=window_context.WINDOW_SESSION_SCOPE_ORGANISATION, + ) + first_worker.set(selected) + assert second_worker.get_owned("ws-cross-worker-revoked", "#V#actor") is not None + + assert first_worker.delete_all_owned("#V#actor") == 1 + + assert second_worker.get_owned("ws-cross-worker-revoked", "#V#actor") is None + assert second_worker.count() == 0 + + +def test_membership_invalidation_preserves_other_org_and_personal_tabs( + durable_store, +) -> None: + collection, repository = durable_store + store = window_context.WindowSessionStore(binding_repository=repository) + for window_id, org_id, scope_kind in ( + ( + "ws-org-a-invalidate", + "#V#org_a", + window_context.WINDOW_SESSION_SCOPE_ORGANISATION, + ), + ( + "ws-org-b-preserve", + "#V#org_b", + window_context.WINDOW_SESSION_SCOPE_ORGANISATION, + ), + ( + "ws-personal-preserve", + None, + window_context.WINDOW_SESSION_SCOPE_PERSONAL, + ), + ): + selected = window_context.WindowSessionContext( + window_session_id=window_id, + user_id="#V#actor", + organisation_concept_id=org_id, + ) + store.persist_authoritative_binding(selected, scope_kind=scope_kind) + store.set(selected) + + assert store.delete_owned_for_organisation("#V#actor", "#V#org_a") == 1 + + assert collection.count_documents({}) == 2 + assert store.get_owned("ws-org-a-invalidate", "#V#actor") is None + assert store.get_owned("ws-org-b-preserve", "#V#actor") is not None + assert store.get_owned("ws-personal-preserve", "#V#actor") is not None + + +def test_recovery_cannot_rebind_between_membership_invalidation_and_commit( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + collection, repository = durable_store + from src.backend.services import ( + ontology_authority_membership_coordination_service as coordination, + ) + + store = window_context.WindowSessionStore(binding_repository=repository) + monkeypatch.setattr(window_context, "_window_session_store", store) + selected = window_context.WindowSessionContext( + window_session_id="ws-revocation-race", + user_id="#V#actor", + organisation_concept_id="#V#org", + ) + store.persist_authoritative_binding( + selected, + scope_kind=window_context.WINDOW_SESSION_SCOPE_ORGANISATION, + ) + stale_recovery = window_context.WindowSessionContext( + window_session_id="ws-revocation-race", + user_id="#V#actor", + organisation_concept_id="#V#org", + durable_scope_kind=window_context.WINDOW_SESSION_SCOPE_ORGANISATION, + durable_recovered=True, + durably_persisted=True, + ) + + membership_active = [True] + monkeypatch.setattr( + organisation_membership_service, + "resolve_user_organisation_membership", + lambda user_id, org_id: ( + { + "user_concept_id": user_id, + "organisation_concept_id": org_id, + "role": "admin", + } + if membership_active[0] + else None + ), + ) + shared_lock = threading.Lock() + invalidated = threading.Event() + recovery_waiting = threading.Event() + permit_commit = threading.Event() + + @contextmanager + def shared_barrier(*_args, **_kwargs): + if threading.current_thread().name == "binding-recovery": + recovery_waiting.set() + with shared_lock: + yield + + monkeypatch.setattr( + coordination, + "organisation_membership_scope_barrier", + shared_barrier, + ) + recovery_errors: list[BaseException] = [] + + def mutate_membership() -> None: + with shared_barrier(): + store.delete_owned_for_organisation("#V#actor", "#V#org") + invalidated.set() + assert permit_commit.wait(timeout=2) + membership_active[0] = False + + def recover_binding() -> None: + try: + window_context._recover_authoritative_scope( + stale_recovery, + user_id="#V#actor", + ) + except Exception as exc: # noqa: BLE001 - asserted in parent thread + recovery_errors.append(exc) + + mutation_thread = threading.Thread(target=mutate_membership) + recovery_thread = threading.Thread( + target=recover_binding, + name="binding-recovery", + ) + mutation_thread.start() + assert invalidated.wait(timeout=2) + recovery_thread.start() + assert recovery_waiting.wait(timeout=2) + assert recovery_thread.is_alive() + permit_commit.set() + mutation_thread.join(timeout=2) + recovery_thread.join(timeout=2) + + assert not mutation_thread.is_alive() + assert not recovery_thread.is_alive() + assert len(recovery_errors) == 1 + assert isinstance( + recovery_errors[0], + window_context.WindowSessionContextUnavailable, + ) + assert collection.count_documents({}) == 0 + + +def test_other_actor_and_unknown_selector_remain_indistinguishable_after_restart( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + _, repository = durable_store + window_context.set_window_organisation( + "ws-owned", "secret_org", "owner", "#V#owner@secret_org", "#V#owner" + ) + _restart_with_repository(monkeypatch, repository) + + for selector in ("ws-owned", "ws-unknown"): + with pytest.raises( + window_context.WindowSessionContextUnavailable, + match="window_session_context_unavailable", + ): + window_context.get_effective_context( + selector, + { + "organisation_concept_id": "other_org", + "namespace": "#V#other@other_org", + }, + "#V#other", + require_known_window=True, + ) + + +def test_restart_recovery_fails_closed_and_deletes_revoked_membership( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + collection, repository = durable_store + window_context.set_window_organisation( + "ws-revoked", "org", "member", "#V#actor@org", "#V#actor" + ) + monkeypatch.setattr( + organisation_membership_service, + "resolve_user_organisation_membership", + lambda *_args, **_kwargs: None, + ) + _restart_with_repository(monkeypatch, repository) + + with pytest.raises(window_context.WindowSessionContextUnavailable): + window_context.get_effective_context( + "ws-revoked", + {"namespace": "#V#actor@wrong_org"}, + "#V#actor", + require_known_window=True, + ) + assert collection.count_documents({}) == 0 + + +def test_bound_selector_never_falls_back_to_flask_after_revocation( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + _collection, repository = durable_store + window_context.set_window_organisation( + "ws-revoked-compat", "org", "member", "#V#actor@org", "#V#actor" + ) + monkeypatch.setattr( + organisation_membership_service, + "resolve_user_organisation_membership", + lambda *_args, **_kwargs: None, + ) + _restart_with_repository(monkeypatch, repository) + + with pytest.raises(window_context.WindowSessionContextUnavailable): + window_context.get_effective_context( + "ws-revoked-compat", + { + "organisation_concept_id": "wrong_flask_org", + "namespace": "#V#actor@wrong_flask_org", + }, + "#V#actor", + require_known_window=False, + ) + + +def test_membership_read_failure_is_recovery_unavailable_not_flask_fallback( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + _collection, repository = durable_store + window_context.set_window_organisation( + "ws-membership-down", "org", "member", "#V#actor@org", "#V#actor" + ) + + def unavailable(*_args, **_kwargs): + raise RuntimeError("membership store unavailable") + + monkeypatch.setattr( + organisation_membership_service, + "resolve_user_organisation_membership", + unavailable, + ) + _restart_with_repository(monkeypatch, repository) + + with pytest.raises(window_context.WindowSessionContextRecoveryUnavailable): + window_context.get_effective_context( + "ws-membership-down", + { + "organisation_concept_id": "wrong_flask_org", + "namespace": "#V#actor@wrong_flask_org", + }, + "#V#actor", + require_known_window=False, + ) + + +def test_expired_binding_is_rejected_before_ttl_cleanup( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + collection, repository = durable_store + window_context.set_window_organisation( + "ws-expired", "org", "member", "#V#actor@org", "#V#actor" + ) + collection.update_one( + {}, + {"$set": {"expires_at": datetime.now(UTC) - timedelta(seconds=1)}}, + ) + _restart_with_repository(monkeypatch, repository) + + with pytest.raises(window_context.WindowSessionContextUnavailable): + window_context.get_effective_context( + "ws-expired", + {"namespace": "#V#actor@wrong_org"}, + "#V#actor", + require_known_window=True, + ) + + +def test_durable_store_outage_cannot_fall_back_to_flask_org( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class UnavailableRepository: + def load_owned(self, *_args, **_kwargs): + raise WindowSessionBindingStoreUnavailable("unavailable") + + store = window_context.WindowSessionStore( + binding_repository=UnavailableRepository() # type: ignore[arg-type] + ) + monkeypatch.setattr(window_context, "_window_session_store", store) + + with pytest.raises(window_context.WindowSessionContextRecoveryUnavailable): + window_context.get_effective_context( + "ws-store-down", + { + "organisation_concept_id": "wrong_flask_org", + "namespace": "#V#actor@wrong_flask_org", + }, + "#V#actor", + require_known_window=True, + ) + + +def test_failed_durable_selection_does_not_replace_working_l1_binding( + monkeypatch: pytest.MonkeyPatch, +) -> None: + class UnavailableRepository: + def save(self, **_kwargs): + raise WindowSessionBindingStoreUnavailable("unavailable") + + store = window_context.WindowSessionStore( + binding_repository=UnavailableRepository() # type: ignore[arg-type] + ) + store.set( + window_context.WindowSessionContext( + window_session_id="ws-preserve", + user_id="#V#actor", + organisation_concept_id="old_org", + role_in_org="member", + namespace="#V#actor@old_org", + ) + ) + monkeypatch.setattr(window_context, "_window_session_store", store) + + with pytest.raises(WindowSessionBindingStoreUnavailable): + window_context.set_window_organisation( + "ws-preserve", + "new_org", + "member", + "#V#actor@new_org", + "#V#actor", + ) + preserved = store.get("ws-preserve") + assert preserved is not None + assert preserved.organisation_concept_id == "old_org" + + +def test_actor_owned_delete_clears_durable_binding_after_l1_restart( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + collection, repository = durable_store + window_context.set_window_organisation( + "ws-delete", "org", "member", "#V#actor@org", "#V#actor" + ) + _restart_with_repository(monkeypatch, repository) + + assert window_context.delete_window_context_if_owned("ws-delete", "#V#actor") + assert collection.count_documents({}) == 0 + + +def test_old_tab_recovers_from_exact_owned_conversation_metadata( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + collection, repository = durable_store + from src.backend.server.routes import von_routes + + monkeypatch.setattr( + von_routes.chat_history_service, + "get_chat_history_session_summary", + lambda *_args, **_kwargs: { + "session_id": "conversation-1", + "namespace": "#V#actor@lab", + "organisation_concept_id": "#V#lab", + }, + ) + app = Flask(__name__) + with app.app_context(): + assert von_routes._recover_window_context_from_owned_conversation( + window_session_id="ws-pre-upgrade", + user_concept_id="#V#actor", + conversation_session_id="conversation-1", + ) + + assert collection.count_documents({}) == 1 + _restart_with_repository(monkeypatch, repository) + effective = window_context.get_effective_context( + "ws-pre-upgrade", + {"namespace": "#V#actor@wrong_org"}, + "#V#actor", + require_known_window=True, + ) + assert effective["organisation_id"] == "#V#lab" + assert effective["namespace"] == "#V#actor@lab" + + +def test_owned_conversation_without_org_or_personal_namespace_is_not_scope_proof( + monkeypatch: pytest.MonkeyPatch, + durable_store, +) -> None: + collection, _repository = durable_store + from src.backend.server.routes import von_routes + + monkeypatch.setattr( + von_routes.chat_history_service, + "get_chat_history_session_summary", + lambda *_args, **_kwargs: { + "session_id": "ambiguous-legacy-conversation", + "namespace": None, + "organisation_concept_id": None, + }, + ) + app = Flask(__name__) + with app.app_context(): + assert not von_routes._recover_window_context_from_owned_conversation( + window_session_id="ws-ambiguous", + user_concept_id="#V#actor", + conversation_session_id="ambiguous-legacy-conversation", + ) + assert collection.count_documents({}) == 0 diff --git a/tests/backend/test_window_session_multi_org_isolation.py b/tests/backend/test_window_session_multi_org_isolation.py index e872a15f..36962946 100644 --- a/tests/backend/test_window_session_multi_org_isolation.py +++ b/tests/backend/test_window_session_multi_org_isolation.py @@ -420,7 +420,7 @@ def capture_create(*args, **kwargs): ) assert ( created_sessions[0].get("organisation_concept_id") - == "university_of_auckland_strong_ai_lab" + == "#V#university_of_auckland_strong_ai_lab" ) assert created_sessions[0].get("is_agent_created") is None window_ctx = wscs.get_window_context(window_a) @@ -728,21 +728,17 @@ def test_get_effective_context_prefers_window_session(self): """get_effective_context should prefer window session over Flask session.""" from src.backend.services.window_session_context_service import ( get_effective_context, - get_window_session_store, - WindowSessionContext, + set_window_organisation, ) - store = get_window_session_store() - # Set up window context - ctx = WindowSessionContext( - window_session_id="test_window", - user_id="user_1", - organisation_concept_id="window_org", - namespace="#V#user_1@window_org", - role_in_org="admin", + set_window_organisation( + "test_window", + "window_org", + "admin", + "#V#user_1@window_org", + "user_1", ) - store.set(ctx) # Flask session has different org flask_session = { @@ -754,7 +750,7 @@ def test_get_effective_context_prefers_window_session(self): # get_effective_context should return window context effective = get_effective_context("test_window", flask_session, "user_1") - assert effective["organisation_id"] == "window_org" + assert effective["organisation_id"] == "#V#window_org" assert effective["namespace"] == "#V#user_1@window_org" assert effective["role"] == "admin" assert effective["source"] == "window_session" @@ -846,7 +842,7 @@ def test_cross_user_window_token_cannot_read_or_replace_org_scope(app_client): preserved = wscs.get_window_context(window_session_id) assert preserved is not None assert preserved.user_id == "#V#owner" - assert preserved.organisation_concept_id == "secret_org" + assert preserved.organisation_concept_id == "#V#secret_org" def test_get_effective_context_falls_back_to_flask(): diff --git a/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js b/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js index d0af19c2..c19df7fe 100644 --- a/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js +++ b/tests/frontend/chatTabThinkingDiagnosticsCopy.test.js @@ -144,6 +144,38 @@ describe('chat thinking diagnostics copy control', () => { expect(payload.stage_diagnostics).toBeUndefined(); }); + test('reports an exact missing or unauthorised turn instead of a delegation outage', async () => { + const chatTab = require(chatTabModulePath); + const { fetchWithTimeout } = require('../../src/frontend/web/von_interface/static/js/apiService.js'); + const button = document.getElementById('copyThinkingDiagnosticsButton'); + global.fetch = jest.fn().mockResolvedValue({ + ok: false, + status: 404, + json: async () => ({ error: 'not_available_in_test' }) + }); + fetchWithTimeout.mockResolvedValue({ + ok: false, + status: 404, + json: async () => ({ error: 'turn_not_found_or_not_authorised' }) + }); + chatTab.__testOnly_setActiveChatSession('session-missing-turn', 'Missing Turn'); + + const copied = await chatTab.__testOnly_copyActiveThinkingDiagnostics(button, { + clientRequestId: 'request-with-no-server-row', + latestProgress: { + status: 'error', + stage: 'error', + phase: 'error', + elapsed_ms: 172 + } + }); + + expect(copied).toBe(true); + const payload = JSON.parse(navigator.clipboard.writeText.mock.calls[0][0]); + expect(payload.mcp_access).toEqual({}); + expect(payload.retrieval_status).toBe('not_found_or_not_authorised'); + }); + test('does not fall back to copying a snapshot when no request reference exists', async () => { const chatTab = require(chatTabModulePath); const { showToast } = require('../../src/frontend/web/von_interface/static/js/utils/toast.js');