From 31559666df240b19f8803f71a806bc98e32c7ad2 Mon Sep 17 00:00:00 2001 From: witbrock Date: Fri, 3 Jul 2026 12:41:40 -0400 Subject: [PATCH] JVNAUTOSCI-2560 surface workflow capability blockers --- ...ticated_browser_workflow_replay_harness.md | 62 +- ...n_authenticated_browser_workflow_replay.py | 905 +++++++++++++++++- src/backend/auth_service.py | 16 +- src/backend/db/connection_manager.py | 7 + src/backend/db/mongo_client.py | 128 ++- .../integrations/internal_mcp/orchestrator.py | 297 +++++- src/backend/server/routes/auth_routes.py | 8 +- .../services/browser_test_auth_service.py | 158 ++- src/backend/services/text_value_service.py | 5 +- .../services/workflow_capability_service.py | 41 +- .../workflow_discovery_memo_service.py | 70 ++ .../workflows/durable/registry_factory.py | 76 ++ .../web/von_interface/static/js/chatTab.js | 55 +- .../web/von_interface/static/js/domUtils.js | 113 +++ .../von_interface/static/js/settingsPage.js | 31 +- .../web/von_interface/static/styles.css | 30 + .../backend/test_browser_test_auth_routes.py | 18 +- .../backend/test_browser_test_auth_service.py | 71 +- tests/backend/test_google_auth_service.py | 99 ++ .../test_mongo_dns_fallback_recovery.py | 57 ++ ...test_orchestrator_selector_core_routing.py | 257 +++++ .../backend/test_pipeline_stage_coherence.py | 78 ++ ...st_text_relations_summary_and_singleton.py | 59 ++ .../test_workflow_capability_service.py | 122 ++- .../test_workflow_discovery_memo_service.py | 53 + .../test_workflow_registry_factory_parity.py | 39 + .../chatTabLlmDebugWorkflowExecution.test.js | 68 ++ .../frontend/footerLlmExecutionStatus.test.js | 39 + .../settingsRuntimeStatusPolling.test.js | 24 + ...n_authenticated_browser_workflow_replay.py | 266 +++++ 30 files changed, 3131 insertions(+), 121 deletions(-) create mode 100644 tests/backend/test_google_auth_service.py diff --git a/docs/engineering/authenticated_browser_workflow_replay_harness.md b/docs/engineering/authenticated_browser_workflow_replay_harness.md index 8bf9066ca..657e73e2c 100644 --- a/docs/engineering/authenticated_browser_workflow_replay_harness.md +++ b/docs/engineering/authenticated_browser_workflow_replay_harness.md @@ -18,6 +18,29 @@ The harness does not add workflow policy to Python. Workflow-specific expectations are replay-case data and assertions over emitted telemetry and progress facts. +Before submitting `/von/generate`, the harness now records and enforces the +operational preconditions that make the replay user-equivalent: + +- Mongo read/write health via `/admin/db/health?probe=rw`; +- effective session user/org context via `/von/api/session/context`; +- target user/org model readiness via `/api/settings/llm/info`, including + selected Ollama model availability when the target model is local; +- workflow capability-index availability via + `/api/workflows/capability-index/status`; +- Gmail OAuth/profile readiness only for replay cases that actually require + Gmail. + +If one of these preconditions is unavailable, the report is still useful +evidence, but it is classified as a typed blocker and the harness does not +submit a misleading turn by default. + +For replay runs the login request explicitly sets `refresh_fixture=false`. +Browser-test fixture preparation is real represented data and can be useful for +browser user-view checks, but it is not part of the auth precondition for a +workflow selector replay. Keeping it separate prevents login from silently +triggering fixture writes, recommendation-workflow events, or capability-index +rebuilds before the replay reaches `/von/generate`. + ## Gmail/arXiv 2421 Case To run the motivating `JVNAUTOSCI-2421` acceptance prompt: @@ -39,13 +62,42 @@ already proves that no configured profile or token set is available. Pass `--run-despite-gmail-preflight-blocker` only when deliberately gathering downstream selector or workflow evidence despite that known external blocker. +The harness also stops before `/von/generate` when the workflow capability index +status endpoint reports that represented workflow discovery is unavailable. +That is a hard replay precondition: selector evidence captured while the +authoritative discovery surface is missing is not user-equivalent evidence. The +preflight waits through a bounded transient build/reload window before blocking, +and records the sampled status snapshots in the JSON report. +Pass `--run-despite-workflow-capability-preflight-blocker` only for diagnostic +experiments that should still be classified as blocked by that precondition. + +For exact user-turn investigations, use an ad-hoc prompt plus explicit target +context instead of forcing the nearest named replay case. Example: + +```sh +./.venv/bin/python scripts/run_authenticated_browser_workflow_replay.py \ + --base-url http://localhost:5001 \ + --allow-non-agent-test-server \ + --case jvnautosci-2560-exact-arxiv \ + --prompt 'Please ingest https://arxiv.org/abs/2406.15341 into Von, including the normal download/file-copy path if available, then read back the represented paper concept, file-copy, and blob evidence so I can see what was stored.' \ + --expected-workflow-id '#V#arxiv_paper_representation_workflow' \ + --target-user-concept-id '#V#michael_witbrock' \ + --target-organisation-concept-id '#V#university_of_auckland_strong_ai_lab' \ + --thinking-card-mode debug \ + --output-json tmp/jvnautosci-2560-exact-arxiv-replay.json +``` + ## Evidence Captured The JSON report records: - local and server branch/commit/environment evidence; - browser-test auth login result and authenticated status; -- Gmail profile/OAuth preflight; +- target user/org session context; +- Mongo read/write preflight; +- effective model readiness preflight; +- workflow capability index preflight, including user-visible blocker state; +- Gmail profile/OAuth preflight when relevant; - chat session and `/von/generate` submission identifiers; - task terminal status and result; - live Thinking-card progress snapshots; @@ -57,9 +109,15 @@ The JSON report records: Typed blockers include: - `browser_test_auth_blocker`; +- `target_session_context_blocker`; +- `database_runtime_blocker`; +- `llm_runtime_blocker`; +- `workflow_capability_index_blocker`; - `gmail_oauth_or_profile_blocker`; +- `context_build_blocker`; +- `workflow_action_terminal_state_blocker`; - `task_terminal_state_blocker`; -- `selector_blocker`; +- `selector_or_dispatch_blocker`; - `thinking_card_projection_blocker`. For the Gmail/arXiv case, a full pass requires selecting diff --git a/scripts/run_authenticated_browser_workflow_replay.py b/scripts/run_authenticated_browser_workflow_replay.py index f48e8a68e..d657853a5 100644 --- a/scripts/run_authenticated_browser_workflow_replay.py +++ b/scripts/run_authenticated_browser_workflow_replay.py @@ -80,6 +80,7 @@ class ReplayCase: expected_workflow_id: str | None expected_progress_fact_ids: tuple[str, ...] expected_contract_ids: tuple[str, ...] + requires_gmail: bool = False GMAIL_ARXIV_REPLAY_CASE = ReplayCase( @@ -88,6 +89,7 @@ class ReplayCase: expected_workflow_id=GMAIL_ARXIV_WORKFLOW_ID, expected_progress_fact_ids=GMAIL_ARXIV_FACT_IDS, expected_contract_ids=GMAIL_ARXIV_CONTRACT_IDS, + requires_gmail=True, ) @@ -112,6 +114,13 @@ def _safe_text(value: Any) -> str: return str(value).strip() +def _normalise_concept_id(value: Any) -> str | None: + text = _safe_text(value) + if not text: + return None + return text if text.startswith("#V#") else f"#V#{text}" + + def _as_mapping(value: Any) -> dict[str, Any]: return dict(value) if isinstance(value, Mapping) else {} @@ -289,7 +298,10 @@ def establish_browser_test_session( f"{base_url}/von/api/auth/browser-test-login", expected_status=200, timeout_seconds=timeout_seconds, - json={"window_session_id": window_session_id}, + json={ + "window_session_id": window_session_id, + "refresh_fixture": False, + }, ) except Exception as exc: payload = exc.payload if isinstance(exc, JsonRequestError) else {} @@ -302,6 +314,114 @@ def establish_browser_test_session( } +def apply_target_session_context( + *, + session: requests.Session, + base_url: str, + user_concept_id: str | None, + organisation_concept_id: str | None, +) -> dict[str, Any]: + """Align the replay session with the requested target user/org context.""" + + result: dict[str, Any] = { + "requested_user_concept_id": _safe_text(user_concept_id) or None, + "requested_organisation_concept_id": ( + _safe_text(organisation_concept_id) or None + ), + "updates": [], + } + target_user = _safe_text(user_concept_id) + target_org = _safe_text(organisation_concept_id) + if target_user: + session.headers.update({"X-User-Concept-ID": target_user}) + try: + result["updates"].append( + { + "step": "set_user_concept", + "payload": _request_json( + session, + "POST", + f"{base_url}/von/api/session/set_user_concept", + timeout_seconds=20.0, + json={"user_concept_id": target_user}, + ), + } + ) + except JsonRequestError as exc: + result["updates"].append( + { + "step": "set_user_concept", + "error": str(exc), + "status_code": exc.status_code, + "payload": exc.payload, + } + ) + if target_org: + try: + result["updates"].append( + { + "step": "set_organisation", + "payload": _request_json( + session, + "POST", + f"{base_url}/von/api/session/set_organisation", + timeout_seconds=20.0, + json={"organisation_concept_id": target_org}, + ), + } + ) + except JsonRequestError as exc: + result["updates"].append( + { + "step": "set_organisation", + "error": str(exc), + "status_code": exc.status_code, + "payload": exc.payload, + } + ) + try: + result["session_context"] = _request_json( + session, + "GET", + f"{base_url}/von/api/session/context", + timeout_seconds=20.0, + ) + except JsonRequestError as exc: + result["session_context"] = { + "authenticated": False, + "error": str(exc), + "status_code": exc.status_code, + "payload": exc.payload, + } + context = _as_mapping(result.get("session_context")) + update_errors = [ + item + for item in _as_list(result.get("updates")) + if isinstance(item, Mapping) and item.get("error") + ] + requested_user = _normalise_concept_id(user_concept_id) + requested_org = _normalise_concept_id(organisation_concept_id) + actual_user = _normalise_concept_id(context.get("user_id")) + actual_org = _normalise_concept_id(context.get("organisation_id")) + mismatch_reasons: list[str] = [] + if requested_user and actual_user != requested_user: + mismatch_reasons.append( + f"requested user {requested_user} but effective user is {actual_user}" + ) + if requested_org and actual_org != requested_org: + mismatch_reasons.append( + f"requested organisation {requested_org} but effective organisation is {actual_org}" + ) + result["target_session_context_ready"] = bool( + context.get("authenticated") is True and not update_errors and not mismatch_reasons + ) + if mismatch_reasons: + result["mismatch_reasons"] = mismatch_reasons + if update_errors: + result["update_errors"] = [dict(item) for item in update_errors] + return result + + def create_replay_chat_session( *, session: requests.Session, @@ -370,7 +490,14 @@ def build_gmail_preflight( session: requests.Session, base_url: str, gmail_profile: str | None, + required: bool = True, ) -> dict[str, Any]: + if not required: + return { + "required": False, + "gmail_capability_ready": True, + "reason": "Replay case does not require Gmail.", + } settings = _query_settings(session=session, base_url=base_url) profiles = _as_list(settings.get("gmail_profiles")) default_profile = _safe_text(settings.get("gmail_default_profile")) or None @@ -382,6 +509,7 @@ def build_gmail_preflight( ) return { "requested_gmail_profile": requested_profile, + "required": True, "configured_gmail_profiles": profiles, "default_gmail_profile": default_profile, "gmail_profiles_configured": bool(profiles), @@ -398,6 +526,280 @@ def build_gmail_preflight( } +def _db_probe_ready(payload: Mapping[str, Any]) -> bool: + probe = _as_mapping(payload.get("probe")) + return bool( + payload.get("connected") is True + and probe.get("ok") is True + and probe.get("ping_ok") is True + and probe.get("read_ok") is True + and probe.get("write_ok") is not False + ) + + +def build_database_preflight( + *, + session: requests.Session, + base_url: str, + attempt_count: int = 3, + poll_interval_seconds: float = 0.75, +) -> dict[str, Any]: + attempts: list[dict[str, Any]] = [] + total_attempts = max(int(attempt_count or 0), 1) + for index in range(total_attempts): + started_at = time.monotonic() + try: + payload = _request_json( + session, + "GET", + f"{base_url}/admin/db/health", + expected_status=(200, 503), + timeout_seconds=60.0, + params={"probe": "rw"}, + ) + except JsonRequestError as exc: + payload = { + "connected": False, + "error": str(exc), + "status_code": exc.status_code, + "payload": exc.payload, + } + except Exception as exc: + payload = { + "connected": False, + "error": str(exc), + } + attempt = dict(payload) + attempt["attempt"] = index + 1 + attempt["elapsed_ms"] = round((time.monotonic() - started_at) * 1000.0, 3) + attempt["database_runtime_available"] = _db_probe_ready(attempt) + attempts.append(attempt) + if index + 1 < total_attempts: + time.sleep(max(float(poll_interval_seconds or 0.0), 0.0)) + + latest = dict(attempts[-1]) + latest["database_runtime_available"] = bool( + attempts and all(_db_probe_ready(item) for item in attempts) + ) + latest["preflight_attempts"] = attempts + latest["preflight_wait"] = { + "attempt_count": total_attempts, + "poll_interval_seconds": max(float(poll_interval_seconds or 0.0), 0.0), + } + if not latest["database_runtime_available"] and not _safe_text( + latest.get("error") + ): + latest["error"] = ( + "Mongo read/write health was not stable across replay preflight " + "samples." + ) + return latest + + +def _normalised_ollama_model_names(payload: Mapping[str, Any]) -> set[str]: + names: set[str] = set() + raw_models = payload.get("models") + if not isinstance(raw_models, list): + raw_models = payload if isinstance(payload, list) else [] + for item in raw_models: + if isinstance(item, Mapping): + candidates = (item.get("name"), item.get("model"), item.get("id")) + else: + candidates = (item,) + for candidate in candidates: + name = _safe_text(candidate) + if not name: + continue + names.add(name) + if name.endswith(":latest"): + names.add(name.removesuffix(":latest")) + elif ":" not in name: + names.add(f"{name}:latest") + return names + + +def build_llm_preflight( + *, + session: requests.Session, + base_url: str, + user_concept_id: str | None, + organisation_concept_id: str | None, + explicit_model: str | None, +) -> dict[str, Any]: + params: dict[str, str] = {} + target_user = _safe_text(user_concept_id) + target_org = _safe_text(organisation_concept_id) + if target_user: + params["user_concept_id"] = target_user + if target_org: + params["organisation_concept_id"] = target_org + try: + llm_info = _request_json( + session, + "GET", + f"{base_url}/api/settings/llm/info", + timeout_seconds=45.0, + params=params, + ) + except JsonRequestError as exc: + return { + "llm_runtime_available": False, + "error": str(exc), + "status_code": exc.status_code, + "payload": exc.payload, + "target_user_concept_id": target_user or None, + "target_organisation_concept_id": target_org or None, + } + except Exception as exc: + return { + "llm_runtime_available": False, + "error": str(exc), + "target_user_concept_id": target_user or None, + "target_organisation_concept_id": target_org or None, + } + + provider = _safe_text(llm_info.get("provider")).lower() + selected_model = _safe_text(llm_info.get("model")) + model_override = _safe_text(explicit_model) + model_to_check = model_override or selected_model + preflight = { + **dict(llm_info), + "target_user_concept_id": target_user or None, + "target_organisation_concept_id": target_org or None, + "model_override": model_override or None, + "model_checked": model_to_check or None, + "selected_model_available": True, + "llm_runtime_available": bool( + llm_info.get("ping_ok") is True + and _safe_text(llm_info.get("status")).lower() == "ready" + ), + } + if provider == "ollama" and model_to_check: + try: + models_payload = _request_json( + session, + "GET", + f"{base_url}/api/settings/ollama/models", + timeout_seconds=45.0, + params={"nocache": "1"}, + ) + available_model_names = _normalised_ollama_model_names(models_payload) + preflight["available_ollama_model_names"] = sorted( + available_model_names + )[:200] + preflight["selected_model_available"] = ( + model_to_check in available_model_names + ) + except JsonRequestError as exc: + preflight["ollama_models_error"] = str(exc) + preflight["ollama_models_status_code"] = exc.status_code + preflight["selected_model_available"] = False + except Exception as exc: + preflight["ollama_models_error"] = str(exc) + preflight["selected_model_available"] = False + preflight["llm_runtime_available"] = bool( + preflight["llm_runtime_available"] + and preflight.get("selected_model_available") is not False + ) + if not preflight["llm_runtime_available"] and not _safe_text( + preflight.get("error") + ): + preflight["error"] = ( + "The effective replay model is not ready or not available for the " + "target user/org context." + ) + return preflight + + +def build_workflow_capability_preflight( + *, + session: requests.Session, + base_url: str, + timeout_seconds: float = 90.0, + poll_interval_seconds: float = 1.5, +) -> dict[str, Any]: + snapshots: list[dict[str, Any]] = [] + effective_timeout = max(float(timeout_seconds or 0.0), 0.0) + deadline = time.monotonic() + effective_timeout + attempt_count = 0 + last_preflight: dict[str, Any] = {} + + while True: + attempt_count += 1 + try: + payload = _request_json( + session, + "GET", + f"{base_url}/api/workflows/capability-index/status", + timeout_seconds=20.0, + ) + except JsonRequestError as exc: + payload = { + "error": str(exc), + "status_code": exc.status_code, + "payload": exc.payload, + "ready": False, + "workflow_discovery_available": False, + } + except Exception as exc: + payload = { + "error": str(exc), + "ready": False, + "workflow_discovery_available": False, + } + + preflight = dict(payload) + if "workflow_discovery_available" not in preflight: + preflight["workflow_discovery_available"] = payload.get("ready") is True + last_preflight = preflight + namespace_state = _as_mapping(preflight.get("namespace_state")) + status = _safe_text(preflight.get("status")).lower() + snapshots.append( + { + "attempt": attempt_count, + "status": preflight.get("status"), + "ready": preflight.get("ready"), + "workflow_discovery_available": preflight.get( + "workflow_discovery_available" + ), + "build_in_progress": preflight.get("build_in_progress"), + "size": preflight.get("size"), + "last_manifest_status": preflight.get("last_manifest_status"), + "last_invalidation_reason": preflight.get("last_invalidation_reason"), + "namespace_status": namespace_state.get("status"), + "detail": preflight.get("detail"), + } + ) + if preflight.get("workflow_discovery_available") is True: + break + + transient = bool(preflight.get("build_in_progress")) or status in { + "building", + "rebuilding", + "warming", + } + if _safe_text(namespace_state.get("status")).lower() == "rebuild_in_progress": + transient = True + if not transient: + break + if time.monotonic() >= deadline: + break + time.sleep(max(float(poll_interval_seconds or 0.0), 0.2)) + + last_preflight["preflight_wait"] = { + "timeout_seconds": effective_timeout, + "poll_interval_seconds": max(float(poll_interval_seconds or 0.0), 0.2), + "attempt_count": attempt_count, + "completed": last_preflight.get("workflow_discovery_available") is True, + "timed_out": ( + last_preflight.get("workflow_discovery_available") is not True + and time.monotonic() >= deadline + ), + "snapshots": snapshots[-8:], + } + return last_preflight + + def submit_background_generate( *, session: requests.Session, @@ -708,6 +1110,67 @@ def _contains_gmail_oauth_blocker_text(value: Any) -> bool: return gmailish and authish +def _selector_route_evidence_present( + selected_workflow_ids: Sequence[str], + observed_workflow_ids: Sequence[str], + selector_diagnostics: Sequence[Mapping[str, Any]], +) -> bool: + if selected_workflow_ids or observed_workflow_ids: + return True + route_keys = { + "selected_workflow_id", + "dispatch_workflow_id", + "workflow_id", + "workflow_selection", + "workflow_routing_diagnostics", + "selected_workflow_execution", + "discovered_workflow_ids", + "candidate_workflow_ids", + "eligible_specialised_candidate_ids", + "selector_candidate_ids", + "selector_discovered_workflow_ids", + "selector_excluded_candidate_ids", + } + for item in selector_diagnostics: + if any(key in item for key in route_keys): + return True + return False + + +def _context_build_progress_blocker( + *, + terminal_status: str, + last_task_status: Mapping[str, Any], + selector_diagnostics: Sequence[Mapping[str, Any]], +) -> dict[str, Any] | None: + progress = _as_mapping(last_task_status.get("progress")) + observed_stage_values = { + _safe_text(progress.get("stage")), + _safe_text(progress.get("phase")), + _safe_text(progress.get("workflow_task")), + } + for item in selector_diagnostics: + observed_stage_values.add(_safe_text(item.get("stage"))) + observed_stage_values.add(_safe_text(item.get("phase"))) + observed_stage_values.add(_safe_text(item.get("workflow_task"))) + observed_stage_values.discard("") + if "context_build" not in observed_stage_values: + return None + + return { + "type": "context_build_blocker", + "reason": ( + "The background turn did not reach workflow selection before it " + f"ended with status {terminal_status!r}; the last visible stage " + "was context_build/request setup, so no selector candidate set was " + "presented to the selector LLM in this replay." + ), + "terminal_task_status": terminal_status, + "last_task_status": dict(last_task_status), + "selector_diagnostics": [dict(item) for item in selector_diagnostics], + } + + def _terminal_workflow_action_blocker( *, terminal_status: str, @@ -744,17 +1207,152 @@ def _terminal_workflow_action_blocker( } +def build_precondition_blockers( + *, + auth_login: Mapping[str, Any], + auth_status: Mapping[str, Any], + target_session_context: Mapping[str, Any] | None = None, + database_preflight: Mapping[str, Any] | None = None, + llm_preflight: Mapping[str, Any] | None = None, + workflow_capability_preflight: Mapping[str, Any] | None = None, + gmail_preflight: Mapping[str, Any] | None = None, + workflow_capability_forced: bool = False, + database_forced: bool = False, + llm_forced: bool = False, + gmail_forced: bool = False, +) -> list[dict[str, Any]]: + """Return explicit replay infrastructure blockers. + + These are operational preconditions for a user-equivalent replay. Keeping + them as a list prevents one infrastructure failure from hiding another. + """ + + blockers: list[dict[str, Any]] = [] + target_context = dict( + target_session_context + if isinstance(target_session_context, Mapping) + else {"target_session_context_ready": True} + ) + verified_target_context_ready = bool( + isinstance(target_session_context, Mapping) + and target_context.get("target_session_context_ready") is True + ) + auth_ready = bool( + auth_login.get("success") + or auth_status.get("authenticated") + or verified_target_context_ready + ) + if not auth_ready: + blockers.append( + { + "type": "browser_test_auth_blocker", + "reason": _safe_text(auth_login.get("error")) + or _safe_text(auth_status.get("error")) + or "No authenticated browser-test or trusted local replay context was established.", + } + ) + + if target_context.get("target_session_context_ready") is False: + blockers.append( + { + "type": "target_session_context_blocker", + "reason": _safe_text(target_context.get("error")) + or "; ".join( + _safe_text(item) + for item in _as_list(target_context.get("mismatch_reasons")) + if _safe_text(item) + ) + or "Replay session did not resolve to the requested user/org context.", + "target_session_context": target_context, + } + ) + + db_preflight = dict( + database_preflight + if isinstance(database_preflight, Mapping) + else {"database_runtime_available": True} + ) + if ( + db_preflight.get("database_runtime_available") is not True + and not database_forced + ): + blockers.append( + { + "type": "database_runtime_blocker", + "reason": _safe_text(db_preflight.get("error")) + or "Mongo read/write health is not stable enough for replay.", + "database_preflight": db_preflight, + } + ) + + llm_payload = dict( + llm_preflight + if isinstance(llm_preflight, Mapping) + else {"llm_runtime_available": True} + ) + if llm_payload.get("llm_runtime_available") is not True and not llm_forced: + blockers.append( + { + "type": "llm_runtime_blocker", + "reason": _safe_text(llm_payload.get("error")) + or "The effective replay LLM is not ready for the target user/org.", + "llm_preflight": llm_payload, + } + ) + + workflow_preflight = dict( + workflow_capability_preflight + if isinstance(workflow_capability_preflight, Mapping) + else {"workflow_discovery_available": True} + ) + if ( + workflow_preflight.get("workflow_discovery_available") is not True + and not workflow_capability_forced + ): + blockers.append( + { + "type": "workflow_capability_index_blocker", + "reason": _safe_text(workflow_preflight.get("detail")) + or _safe_text(workflow_preflight.get("summary")) + or _safe_text(workflow_preflight.get("error")) + or ( + "Workflow discovery cannot rely on the authoritative " + "capability index." + ), + "workflow_capability_preflight": workflow_preflight, + } + ) + + gmail_payload = dict(gmail_preflight if isinstance(gmail_preflight, Mapping) else {}) + if not gmail_payload.get("gmail_capability_ready") and not gmail_forced: + blockers.append( + { + "type": "gmail_oauth_or_profile_blocker", + "reason": ( + "Gmail profile/tokens are not ready for an authenticated " + "Gmail-backed replay." + ), + "gmail_preflight": gmail_payload, + } + ) + return blockers + + def classify_replay( *, case: ReplayCase, auth_login: Mapping[str, Any], auth_status: Mapping[str, Any], + target_session_context: Mapping[str, Any] | None = None, + database_preflight: Mapping[str, Any] | None = None, + llm_preflight: Mapping[str, Any] | None = None, gmail_preflight: Mapping[str, Any], task_evidence: Mapping[str, Any], selected_workflow_ids: Sequence[str], observed_workflow_ids: Sequence[str], selector_diagnostics: Sequence[Mapping[str, Any]], progress_facts: Sequence[Mapping[str, Any]], + workflow_capability_preflight: Mapping[str, Any] | None = None, ) -> dict[str, Any]: missing_fact_ids = sorted(set(case.expected_progress_fact_ids) - _fact_ids(progress_facts)) missing_contract_ids = sorted(set(case.expected_contract_ids) - _contract_ids(progress_facts)) @@ -762,26 +1360,43 @@ def classify_replay( not case.expected_workflow_id or case.expected_workflow_id in set(selected_workflow_ids) ) - route_evidence_present = bool( - selected_workflow_ids or observed_workflow_ids or selector_diagnostics + route_evidence_present = _selector_route_evidence_present( + selected_workflow_ids, + observed_workflow_ids, + selector_diagnostics, ) last_task_status = _as_mapping(task_evidence.get("last_task_status")) terminal_status = _safe_text(last_task_status.get("status")) or "unknown" + workflow_preflight = dict( + workflow_capability_preflight + if isinstance(workflow_capability_preflight, Mapping) + else {"workflow_discovery_available": True} + ) + precondition_blockers = build_precondition_blockers( + auth_login=auth_login, + auth_status=auth_status, + target_session_context=target_session_context, + database_preflight=database_preflight, + llm_preflight=llm_preflight, + workflow_capability_preflight=workflow_preflight, + gmail_preflight=gmail_preflight, + gmail_forced=not case.requires_gmail, + ) + context_build_blocker = ( + _context_build_progress_blocker( + terminal_status=terminal_status, + last_task_status=last_task_status, + selector_diagnostics=selector_diagnostics, + ) + if terminal_status != "completed" and not route_evidence_present + else None + ) blocker: dict[str, Any] | None = None - if not auth_login.get("success") or not auth_status.get("authenticated"): - blocker = { - "type": "browser_test_auth_blocker", - "reason": _safe_text(auth_login.get("error")) - or _safe_text(auth_status.get("error")) - or "Browser-test login did not establish an authenticated session.", - } - elif not gmail_preflight.get("gmail_capability_ready"): - blocker = { - "type": "gmail_oauth_or_profile_blocker", - "reason": "Gmail profile/tokens are not ready for an authenticated Gmail-backed replay.", - "gmail_preflight": dict(gmail_preflight), - } + if precondition_blockers: + blocker = dict(precondition_blockers[0]) + elif context_build_blocker is not None: + blocker = context_build_blocker elif ( not selected_expected_workflow and (route_evidence_present or terminal_status == "completed") @@ -847,6 +1462,11 @@ def classify_replay( "selected_workflow_ids": list(selected_workflow_ids), "observed_workflow_ids": list(observed_workflow_ids), "selector_diagnostics": [dict(item) for item in selector_diagnostics], + "target_session_context": dict(target_session_context or {}), + "database_preflight": dict(database_preflight or {}), + "llm_preflight": dict(llm_preflight or {}), + "workflow_capability_preflight": workflow_preflight, + "precondition_blockers": [dict(item) for item in precondition_blockers], "observed_progress_fact_ids": sorted(_fact_ids(progress_facts)), "observed_contract_ids": sorted(_contract_ids(progress_facts)), "missing_progress_fact_ids": missing_fact_ids, @@ -862,10 +1482,14 @@ def build_replay_report( window_session_id: str, auth_login: Mapping[str, Any], auth_status: Mapping[str, Any], + target_session_context: Mapping[str, Any], + database_preflight: Mapping[str, Any], + llm_preflight: Mapping[str, Any], gmail_preflight: Mapping[str, Any], chat_session: Mapping[str, Any], submission: Mapping[str, Any], task_evidence: Mapping[str, Any], + workflow_capability_preflight: Mapping[str, Any] | None = None, ) -> dict[str, Any]: task_result = _as_mapping(task_evidence.get("task_result")) task_statuses = _as_list(task_evidence.get("task_statuses")) @@ -893,12 +1517,21 @@ def build_replay_report( case=case, auth_login=auth_login, auth_status=auth_status, + target_session_context=target_session_context, + database_preflight=database_preflight, + llm_preflight=llm_preflight, gmail_preflight=gmail_preflight, task_evidence=task_evidence, selected_workflow_ids=selected_workflow_ids, observed_workflow_ids=observed_workflow_ids, selector_diagnostics=selector_diagnostics, progress_facts=progress_facts, + workflow_capability_preflight=workflow_capability_preflight, + ) + workflow_preflight = dict( + workflow_capability_preflight + if isinstance(workflow_capability_preflight, Mapping) + else {"workflow_discovery_available": True} ) return { "schema_version": "authenticated_browser_workflow_replay.v1", @@ -913,6 +1546,10 @@ def build_replay_report( "window_session_id": window_session_id, "auth_login": dict(auth_login), "auth_status": dict(auth_status), + "target_session_context": dict(target_session_context), + "database_preflight": dict(database_preflight), + "llm_preflight": dict(llm_preflight), + "workflow_capability_preflight": workflow_preflight, "gmail_preflight": dict(gmail_preflight), "chat_session": dict(chat_session), "submission": dict(submission), @@ -940,12 +1577,21 @@ def run_replay( base_url: str, gmail_profile: str | None, model: str | None, + target_user_concept_id: str | None, + target_organisation_concept_id: str | None, presenter_mode: bool, thinking_card_mode: str, timeout_seconds: float, poll_interval_seconds: float, auth_login_timeout_seconds: float, + database_preflight_attempts: int, + database_preflight_poll_interval_seconds: float, + workflow_capability_preflight_timeout_seconds: float, + workflow_capability_preflight_poll_interval_seconds: float, allow_non_agent_test_server: bool, + run_despite_database_preflight_blocker: bool, + run_despite_llm_preflight_blocker: bool, + run_despite_workflow_capability_preflight_blocker: bool, run_despite_gmail_preflight_blocker: bool, cancel_on_timeout: bool, ) -> dict[str, Any]: @@ -967,11 +1613,38 @@ def run_replay( user_concept_id = _safe_text(auth_login.get("user_concept_id")) if user_concept_id: session.headers.update({"X-User-Concept-ID": user_concept_id}) + effective_target_user = _safe_text(target_user_concept_id) or user_concept_id + target_session_context = apply_target_session_context( + session=session, + base_url=base_url, + user_concept_id=effective_target_user, + organisation_concept_id=target_organisation_concept_id, + ) auth_status = get_auth_status(session=session, base_url=base_url) + database_preflight = build_database_preflight( + session=session, + base_url=base_url, + attempt_count=database_preflight_attempts, + poll_interval_seconds=database_preflight_poll_interval_seconds, + ) + llm_preflight = build_llm_preflight( + session=session, + base_url=base_url, + user_concept_id=effective_target_user, + organisation_concept_id=target_organisation_concept_id, + explicit_model=model, + ) + workflow_capability_preflight = build_workflow_capability_preflight( + session=session, + base_url=base_url, + timeout_seconds=workflow_capability_preflight_timeout_seconds, + poll_interval_seconds=workflow_capability_preflight_poll_interval_seconds, + ) gmail_preflight = build_gmail_preflight( session=session, base_url=base_url, gmail_profile=gmail_profile, + required=case.requires_gmail, ) chat_session = {} submission = {} @@ -986,9 +1659,51 @@ def run_replay( gmail_preflight.get("gmail_capability_ready") or run_despite_gmail_preflight_blocker ) - if ( + workflow_capability_ready_or_forced = bool( + workflow_capability_preflight.get("workflow_discovery_available") + or run_despite_workflow_capability_preflight_blocker + ) + database_ready_or_forced = bool( + database_preflight.get("database_runtime_available") + or run_despite_database_preflight_blocker + ) + llm_ready_or_forced = bool( + llm_preflight.get("llm_runtime_available") + or run_despite_llm_preflight_blocker + ) + target_context_ready = bool( + target_session_context.get("target_session_context_ready") is not False + ) + auth_ready = bool( auth_login.get("success") - and auth_status.get("authenticated") + or auth_status.get("authenticated") + or target_context_ready + ) + active_precondition_blockers = build_precondition_blockers( + auth_login=auth_login, + auth_status=auth_status, + target_session_context=target_session_context, + database_preflight=database_preflight, + llm_preflight=llm_preflight, + workflow_capability_preflight=workflow_capability_preflight, + gmail_preflight=gmail_preflight, + database_forced=run_despite_database_preflight_blocker, + llm_forced=run_despite_llm_preflight_blocker, + workflow_capability_forced=run_despite_workflow_capability_preflight_blocker, + gmail_forced=run_despite_gmail_preflight_blocker, + ) + if active_precondition_blockers: + submission = { + "skipped": True, + "skip_reason": "preflight_blocker", + "precondition_blockers": active_precondition_blockers, + } + elif ( + auth_ready + and target_context_ready + and database_ready_or_forced + and llm_ready_or_forced + and workflow_capability_ready_or_forced and gmail_ready_or_forced ): chat_session = create_replay_chat_session( @@ -1028,6 +1743,10 @@ def run_replay( window_session_id=window_session_id, auth_login=auth_login, auth_status=auth_status, + target_session_context=target_session_context, + database_preflight=database_preflight, + llm_preflight=llm_preflight, + workflow_capability_preflight=workflow_capability_preflight, gmail_preflight=gmail_preflight, chat_session=chat_session, submission=submission, @@ -1046,7 +1765,33 @@ def _write_output(path: str | None, report: Mapping[str, Any]) -> None: output_path.write_text(output, encoding="utf-8") -def _resolve_case(case_id: str) -> ReplayCase: +def _resolve_case( + case_id: str, + *, + prompt: str | None = None, + expected_workflow_id: str | None = None, + expected_progress_fact_ids: Sequence[str] | None = None, + expected_contract_ids: Sequence[str] | None = None, + requires_gmail: bool = False, +) -> ReplayCase: + prompt_text = _safe_text(prompt) + if prompt_text: + return ReplayCase( + case_id=_safe_text(case_id) or f"ad_hoc_{uuid.uuid4()}", + prompt=prompt_text, + expected_workflow_id=_safe_text(expected_workflow_id) or None, + expected_progress_fact_ids=tuple( + item + for item in (_safe_text(value) for value in (expected_progress_fact_ids or ())) + if item + ), + expected_contract_ids=tuple( + item + for item in (_safe_text(value) for value in (expected_contract_ids or ())) + if item + ), + requires_gmail=requires_gmail, + ) cleaned = _safe_text(case_id) if cleaned in {"gmail-arxiv-2421", GMAIL_ARXIV_REPLAY_CASE.case_id}: return GMAIL_ARXIV_REPLAY_CASE @@ -1061,14 +1806,81 @@ def main(argv: Sequence[str] | None = None) -> int: ) ) parser.add_argument("--case", default="gmail-arxiv-2421") + parser.add_argument( + "--prompt", + default=None, + help=( + "Run an ad-hoc replay prompt instead of a named replay case. " + "Workflow expectations remain optional telemetry assertions." + ), + ) + parser.add_argument( + "--expected-workflow-id", + default=None, + help="Expected selected workflow ID for ad-hoc replay analysis.", + ) + parser.add_argument( + "--expected-progress-fact-id", + action="append", + default=[], + help="Expected progress fact ID; may be supplied multiple times.", + ) + parser.add_argument( + "--expected-contract-id", + action="append", + default=[], + help="Expected progress contract ID; may be supplied multiple times.", + ) + parser.add_argument( + "--requires-gmail", + action="store_true", + help="Require Gmail profile/OAuth preflight for an ad-hoc prompt.", + ) parser.add_argument("--base-url", default=get_default_agent_test_base_url()) parser.add_argument("--gmail-profile", default=None) parser.add_argument("--model", default=None) + parser.add_argument("--target-user-concept-id", default=None) + parser.add_argument( + "--target-organisation-concept-id", + "--target-organization-concept-id", + dest="target_organisation_concept_id", + default=None, + ) parser.add_argument("--thinking-card-mode", default="debug") parser.add_argument("--presenter-mode", action="store_true") parser.add_argument("--timeout-seconds", type=float, default=240.0) parser.add_argument("--poll-interval-seconds", type=float, default=1.0) parser.add_argument("--auth-login-timeout-seconds", type=float, default=180.0) + parser.add_argument( + "--database-preflight-attempts", + type=int, + default=3, + help=( + "Number of Mongo read/write health samples required before replay " + "submission." + ), + ) + parser.add_argument( + "--database-preflight-poll-interval-seconds", + type=float, + default=0.75, + help="Delay between Mongo read/write preflight samples.", + ) + parser.add_argument( + "--workflow-capability-preflight-timeout-seconds", + type=float, + default=90.0, + help=( + "Maximum time to wait for represented workflow discovery to become " + "available before classifying the replay as blocked." + ), + ) + parser.add_argument( + "--workflow-capability-preflight-poll-interval-seconds", + type=float, + default=1.5, + help="Polling interval for the workflow capability preflight wait.", + ) parser.add_argument("--output-json", default=None) parser.add_argument( "--skip-cancel-on-timeout", @@ -1086,6 +1898,31 @@ def main(argv: Sequence[str] | None = None) -> int: "already indicates that the replay is blocked." ), ) + parser.add_argument( + "--run-despite-database-preflight-blocker", + action="store_true", + help=( + "Submit /von/generate even when Mongo read/write preflight is " + "unhealthy. Use only for diagnostic run-throughs." + ), + ) + parser.add_argument( + "--run-despite-llm-preflight-blocker", + action="store_true", + help=( + "Submit /von/generate even when the target user's effective model " + "is unavailable. Use only for diagnostic run-throughs." + ), + ) + parser.add_argument( + "--run-despite-workflow-capability-preflight-blocker", + action="store_true", + help=( + "Submit /von/generate even when the workflow capability index " + "preflight says represented workflow discovery is unavailable. " + "Use only for diagnostic run-throughs, not acceptance evidence." + ), + ) parser.add_argument( "--allow-non-agent-test-server", action="store_true", @@ -1096,18 +1933,46 @@ def main(argv: Sequence[str] | None = None) -> int: ) args = parser.parse_args(argv) - case = _resolve_case(args.case) + case = _resolve_case( + args.case, + prompt=args.prompt, + expected_workflow_id=args.expected_workflow_id, + expected_progress_fact_ids=args.expected_progress_fact_id, + expected_contract_ids=args.expected_contract_id, + requires_gmail=bool(args.requires_gmail), + ) report = run_replay( case=case, base_url=str(args.base_url).rstrip("/"), gmail_profile=args.gmail_profile, model=args.model, + target_user_concept_id=args.target_user_concept_id, + target_organisation_concept_id=args.target_organisation_concept_id, presenter_mode=bool(args.presenter_mode), thinking_card_mode=args.thinking_card_mode, timeout_seconds=float(args.timeout_seconds), poll_interval_seconds=float(args.poll_interval_seconds), auth_login_timeout_seconds=float(args.auth_login_timeout_seconds), + database_preflight_attempts=int(args.database_preflight_attempts), + database_preflight_poll_interval_seconds=float( + args.database_preflight_poll_interval_seconds + ), + workflow_capability_preflight_timeout_seconds=float( + args.workflow_capability_preflight_timeout_seconds + ), + workflow_capability_preflight_poll_interval_seconds=float( + args.workflow_capability_preflight_poll_interval_seconds + ), allow_non_agent_test_server=bool(args.allow_non_agent_test_server), + run_despite_database_preflight_blocker=bool( + args.run_despite_database_preflight_blocker + ), + run_despite_llm_preflight_blocker=bool( + args.run_despite_llm_preflight_blocker + ), + run_despite_workflow_capability_preflight_blocker=bool( + args.run_despite_workflow_capability_preflight_blocker + ), run_despite_gmail_preflight_blocker=bool( args.run_despite_gmail_preflight_blocker ), diff --git a/src/backend/auth_service.py b/src/backend/auth_service.py index cd545d8e2..1511d2459 100644 --- a/src/backend/auth_service.py +++ b/src/backend/auth_service.py @@ -13,6 +13,9 @@ load_secret_from_env_or_file, ) +_DEFAULT_GOOGLE_OAUTH_CLOCK_SKEW_SECONDS = 10 +_MAX_GOOGLE_OAUTH_CLOCK_SKEW_SECONDS = 300 + @runtime_checkable class CredentialsProtocol(Protocol): # Minimal shape we rely on for static checking @@ -40,12 +43,17 @@ def __init__(self): ) configure_oauthlib_insecure_transport(self.redirect_uri) # Optional tolerance for minor system clock differences when verifying ID tokens. - # Keep small (e.g., 5-10s). Default 0s to preserve strict behaviour. + # Google callbacks can arrive a few seconds before local time catches token iat. try: - skew_raw = os.getenv("GOOGLE_OAUTH_CLOCK_SKEW_SECONDS", "0").strip() - self.clock_skew_seconds = max(0, min(int(skew_raw or "0"), 300)) + skew_raw = os.getenv( + "GOOGLE_OAUTH_CLOCK_SKEW_SECONDS", + str(_DEFAULT_GOOGLE_OAUTH_CLOCK_SKEW_SECONDS), + ).strip() + self.clock_skew_seconds = max( + 0, min(int(skew_raw or "0"), _MAX_GOOGLE_OAUTH_CLOCK_SKEW_SECONDS) + ) except Exception: - self.clock_skew_seconds = 0 + self.clock_skew_seconds = _DEFAULT_GOOGLE_OAUTH_CLOCK_SKEW_SECONDS parsed_redirect = urlparse(self.redirect_uri) # Derive canonical host (scheme://host:port) from configured redirect for validation if parsed_redirect.scheme and parsed_redirect.netloc: diff --git a/src/backend/db/connection_manager.py b/src/backend/db/connection_manager.py index 458f46a84..8f43f213b 100644 --- a/src/backend/db/connection_manager.py +++ b/src/backend/db/connection_manager.py @@ -83,6 +83,12 @@ def health_summary() -> Dict[str, Any]: using_fallback=using_fallback, ) effective_uri_sanitized = connection_location.get("sanitized_uri") + try: + fallback_policy = getattr( + _mc, "get_mongo_fallback_policy_state", lambda: {} + )() + except Exception: + fallback_policy = {} return { "connected": connected, "using_fallback": using_fallback, @@ -90,6 +96,7 @@ def health_summary() -> Dict[str, Any]: "effective_uri": effective_uri_sanitized, "effective_uri_sanitized": effective_uri_sanitized, "effective_mongo_location": connection_location, + "fallback_policy": fallback_policy, "uptime_seconds": uptime, "metrics": metrics, } diff --git a/src/backend/db/mongo_client.py b/src/backend/db/mongo_client.py index b7d3aecdf..c3246756b 100644 --- a/src/backend/db/mongo_client.py +++ b/src/backend/db/mongo_client.py @@ -399,6 +399,10 @@ def _mongo_auto_recovery_ping_timeout_ms() -> int: return _get_positive_int_env("VON_MONGO_AUTO_RECOVERY_PING_TIMEOUT_MS", 1500) +def _mongo_dns_fallback_sticky_seconds() -> float: + return _get_positive_float_env("VON_MONGO_DNS_FALLBACK_STICKY_SECONDS", 300.0) + + def _mongo_server_selection_timeout_ms() -> int: return _get_positive_int_env("MONGO_SERVER_SELECTION_TIMEOUT_MS", 5000) @@ -532,6 +536,8 @@ def assert_destructive_db_operation_allowed(operation: str) -> None: _using_fallback_real = False _effective_uri_real = MONGO_URI # The URI actually used to create the real client _last_auto_recovery_check_at = 0.0 +_dns_fallback_preferred_until_monotonic = 0.0 +_dns_fallback_preference_reason: str | None = None _COLLECTION_INDEXES_LOCK = threading.Lock() _COLLECTION_INDEXES_READY: set[tuple[int, str, str]] = set() @@ -568,6 +574,81 @@ def _new_mongo_client( ) +def _dns_fallback_is_preferred() -> bool: + return bool( + MONGO_DNS_FALLBACK_URI + and _dns_fallback_preferred_until_monotonic > time.monotonic() + ) + + +def _mark_dns_fallback_preferred(reason: str) -> None: + """Prefer the configured direct-host Atlas fallback for a bounded window.""" + + global _dns_fallback_preferred_until_monotonic + global _dns_fallback_preference_reason + sticky_seconds = _mongo_dns_fallback_sticky_seconds() + if not MONGO_DNS_FALLBACK_URI or sticky_seconds <= 0: + _dns_fallback_preferred_until_monotonic = 0.0 + _dns_fallback_preference_reason = None + return + _dns_fallback_preferred_until_monotonic = time.monotonic() + sticky_seconds + _dns_fallback_preference_reason = reason + + +def _clear_dns_fallback_preference() -> None: + global _dns_fallback_preferred_until_monotonic + global _dns_fallback_preference_reason + _dns_fallback_preferred_until_monotonic = 0.0 + _dns_fallback_preference_reason = None + + +def _try_connect_uri( + uri: str, + *, + fallback_label: str | None = None, + server_selection_timeout_ms: int | None = None, +) -> MongoClient: + client = _new_mongo_client( + uri, + server_selection_timeout_ms=server_selection_timeout_ms, + ) + client.admin.command("ismaster") + if ( + fallback_label == "dns fallback" + and MONGO_DNS_FALLBACK_URI + and uri == MONGO_DNS_FALLBACK_URI + ): + _mark_dns_fallback_preferred(f"connected via {fallback_label}") + elif uri == MONGO_URI: + _clear_dns_fallback_preference() + return client + + +def _try_preferred_dns_fallback_first() -> bool: + """Connect to the direct-host Atlas fallback while the sticky window is active.""" + + global _mongo_client_real, _effective_uri_real, _using_fallback_real + if not _dns_fallback_is_preferred() or not MONGO_DNS_FALLBACK_URI: + return False + try: + logger.warning( + "[mongo_fallback] Preferring configured direct-host Mongo URI during DNS fallback recovery window." + ) + _mongo_client_real = _try_connect_uri( + MONGO_DNS_FALLBACK_URI, + fallback_label="dns fallback", + server_selection_timeout_ms=3000, + ) + _effective_uri_real = MONGO_DNS_FALLBACK_URI + _using_fallback_real = True + return True + except Exception as exc: + logger.warning("[mongo_fallback] Preferred DNS fallback connection failed: %s", exc) + _mongo_client_real = None + _clear_dns_fallback_preference() + return False + + def _should_run_auto_recovery_check() -> bool: global _last_auto_recovery_check_at if not _mongo_auto_recovery_enabled(): @@ -586,8 +667,12 @@ def _invalidate_real_client_for_recovery(reason: str) -> None: global _mongo_client_real, _effective_uri_real, _using_fallback_real logger.warning("[mongo_recovery] Invalidating Mongo client: %s", reason) _mongo_client_real = None - _effective_uri_real = MONGO_URI - _using_fallback_real = False + if _dns_fallback_is_preferred() and MONGO_DNS_FALLBACK_URI: + _effective_uri_real = MONGO_DNS_FALLBACK_URI + _using_fallback_real = True + else: + _effective_uri_real = MONGO_URI + _using_fallback_real = False def _ensure_active_client_is_healthy() -> None: @@ -627,10 +712,13 @@ def get_db() -> Database | None: if _mongo_client_real is None: try: + if _try_preferred_dns_fallback_first(): + _last_auto_recovery_check_at = time.monotonic() + if _mongo_client_real: + print_connection_info() + return _mongo_client_real[db_name] # Create a fresh client when none exists, or when recovery invalidated it. - _mongo_client_real = _new_mongo_client(MONGO_URI) - # The ismaster command is cheap and does not require auth. - _mongo_client_real.admin.command("ismaster") + _mongo_client_real = _try_connect_uri(MONGO_URI) _effective_uri_real = MONGO_URI _using_fallback_real = False _last_auto_recovery_check_at = time.monotonic() @@ -696,6 +784,14 @@ def get_db() -> Database | None: fallback_uri, server_selection_timeout_ms=3000 ) _mongo_client_real.admin.command("ismaster") + if ( + fallback_label == "dns fallback" + and MONGO_DNS_FALLBACK_URI + and fallback_uri == MONGO_DNS_FALLBACK_URI + ): + _mark_dns_fallback_preferred( + "primary connection failure" + ) _effective_uri_real = fallback_uri _using_fallback_real = True _last_auto_recovery_check_at = time.monotonic() @@ -747,6 +843,12 @@ def get_db() -> Database | None: fallback_uri, server_selection_timeout_ms=3000 ) _mongo_client_real.admin.command("ismaster") + if ( + fallback_label == "dns fallback" + and MONGO_DNS_FALLBACK_URI + and fallback_uri == MONGO_DNS_FALLBACK_URI + ): + _mark_dns_fallback_preferred("primary DNS/SRV error") _effective_uri_real = fallback_uri _using_fallback_real = True _last_auto_recovery_check_at = time.monotonic() @@ -803,6 +905,21 @@ def is_using_fallback_uri() -> bool: return _using_fallback_real +def get_mongo_fallback_policy_state() -> dict[str, object]: + """Return paste-safe fallback/recovery state for operator diagnostics.""" + + remaining_seconds = max( + 0.0, _dns_fallback_preferred_until_monotonic - time.monotonic() + ) + return { + "dns_fallback_configured": bool(MONGO_DNS_FALLBACK_URI), + "dns_fallback_sticky": bool(remaining_seconds > 0), + "dns_fallback_sticky_seconds_remaining": round(remaining_seconds, 3), + "dns_fallback_sticky_seconds_configured": _mongo_dns_fallback_sticky_seconds(), + "dns_fallback_preference_reason": _dns_fallback_preference_reason, + } + + def close_connection(): """Closes the MongoDB connection.""" global _mongo_client_real, _mongo_client_mock, _last_auto_recovery_check_at @@ -812,6 +929,7 @@ def close_connection(): # mongomock doesn't require close(), but clear ref for correctness. _mongo_client_mock = None _last_auto_recovery_check_at = 0.0 + _clear_dns_fallback_preference() with _COLLECTION_INDEXES_LOCK: _COLLECTION_INDEXES_READY.clear() diff --git a/src/backend/integrations/internal_mcp/orchestrator.py b/src/backend/integrations/internal_mcp/orchestrator.py index 5ace9db1e..9c2b6e7fe 100644 --- a/src/backend/integrations/internal_mcp/orchestrator.py +++ b/src/backend/integrations/internal_mcp/orchestrator.py @@ -33381,15 +33381,8 @@ def _prepare_selector_discovered_matches( routing_profile = candidate.get("routing_profile") if not isinstance(routing_profile, Mapping): - registration = self._workflow_registry.get_registration(concept_id) - definition = getattr(registration, "definition", None) - definition_metadata = ( - getattr(definition, "metadata", None) - if definition is not None - else None - ) - if isinstance(definition_metadata, Mapping): - routing_profile = definition_metadata.get("routing_profile") + metadata = self._peek_workflow_registration_metadata(concept_id) + routing_profile = metadata.get("routing_profile") if isinstance(routing_profile, Mapping): item["routing_profile"] = dict(routing_profile) @@ -34469,6 +34462,81 @@ def _turn_discovery_query_is_enriched( and clean_discovery_input != clean_requested ) + @staticmethod + def _workflow_discovery_candidate_count( + workflow_discovery_result: Mapping[str, Any] | None, + ) -> int: + if not isinstance(workflow_discovery_result, Mapping): + return 0 + for key in ("candidate_count", "match_count"): + raw_value = workflow_discovery_result.get(key) + if raw_value is None: + continue + try: + parsed = int(raw_value) + except (TypeError, ValueError): + continue + if parsed >= 0: + return parsed + for key in ("candidates", "matches", "routing_matches"): + raw_candidates = workflow_discovery_result.get(key) + if isinstance(raw_candidates, list): + return len(raw_candidates) + return 0 + + @classmethod + def _workflow_discovery_has_zero_candidate_operational_blocker( + cls, + workflow_discovery_result: Mapping[str, Any] | None, + ) -> bool: + if not isinstance(workflow_discovery_result, Mapping): + return False + if cls._workflow_discovery_candidate_count(workflow_discovery_result) > 0: + return False + + reason_text: list[str] = [] + for key in ( + "match_absence_reason", + "blocker_reason", + "budget_exhaustion_stage", + "budget_exhaustion_detail", + ): + value = workflow_discovery_result.get(key) + if isinstance(value, str) and value.strip(): + reason_text.append(value.strip().lower()) + errors = workflow_discovery_result.get("errors") + if isinstance(errors, list): + reason_text.extend( + str(item).strip().lower() + for item in errors + if isinstance(item, str) and str(item).strip() + ) + stage_timings = workflow_discovery_result.get("stage_timings") + if isinstance(stage_timings, list): + for stage in stage_timings: + if not isinstance(stage, Mapping): + continue + for key in ("stage", "status", "error"): + value = stage.get(key) + if isinstance(value, str) and value.strip(): + reason_text.append(value.strip().lower()) + stage_errors = stage.get("errors") + if isinstance(stage_errors, list): + reason_text.extend( + str(item).strip().lower() + for item in stage_errors + if isinstance(item, str) and str(item).strip() + ) + + if bool(workflow_discovery_result.get("budget_exhausted")): + return True + return any( + "capability_index" in reason + or "workflow_discovery_budget_exhausted" in reason + or "query_surface" in reason + for reason in reason_text + ) + @classmethod def _should_refresh_turn_workflow_discovery_result( cls, @@ -34482,6 +34550,10 @@ def _should_refresh_turn_workflow_discovery_result( or not workflow_discovery_result ): return False + if cls._workflow_discovery_has_zero_candidate_operational_blocker( + workflow_discovery_result + ): + return True clean_requested_query = ( requested_query.strip() if isinstance(requested_query, str) else "" ) @@ -34738,6 +34810,59 @@ def _collect_workflow_definition_action_ids( action_ids.append(action_id) return tuple(action_ids) + def _peek_workflow_registration_metadata( + self, + workflow_id: str, + ) -> Mapping[str, Any]: + """Return eager or lazy workflow metadata without resolving lazy graphs.""" + + registration = None + try: + peek_registration = getattr(self._workflow_registry, "peek_registration") + except Exception: + peek_registration = None + if callable(peek_registration): + try: + registration = peek_registration(workflow_id) + except Exception: + registration = None + + if registration is None: + try: + eager_ids = set(self._workflow_registry.eager_workflow_ids()) + except Exception: + eager_ids = set() + if workflow_id in eager_ids: + try: + registration = self._workflow_registry.get_registration(workflow_id) + except Exception: + registration = None + + if registration is None: + return {} + + definition = getattr(registration, "definition", None) + metadata = ( + getattr(definition, "metadata", None) if definition is not None else None + ) + routing_profile = ( + metadata.get("routing_profile") + if isinstance(metadata, Mapping) + else None + ) + return { + "purpose": str( + getattr(registration, "purpose", None) + or getattr(definition, "purpose", None) + or "" + ).strip(), + "source": str(getattr(registration, "source", "") or "").strip(), + "definition_loaded": definition is not None, + "routing_profile": ( + dict(routing_profile) if isinstance(routing_profile, Mapping) else None + ), + } + def _build_agent_test_required_tool_workflow_candidates( self, *, @@ -35243,6 +35368,20 @@ def _build_agent_test_local_selector_outputs() -> dict[str, Any]: "VON_TURN_WORKFLOW_DISCOVERY_TIMEOUT_SECONDS", "0.75", ) + if self._workflow_discovery_has_zero_candidate_operational_blocker( + raw_discovery if isinstance(raw_discovery, Mapping) else None + ): + recovery_timeout_raw = os.getenv( + "VON_TURN_WORKFLOW_DISCOVERY_RECOVERY_TIMEOUT_SECONDS", + "15.0", + ) + try: + discovery_timeout_seconds = max( + float(discovery_timeout_seconds), + float(recovery_timeout_raw), + ) + except (TypeError, ValueError): + discovery_timeout_seconds = recovery_timeout_raw turn_scope = None for scope_key in ("turn_id", "request_id"): scope_value = data.get(scope_key) @@ -47215,6 +47354,47 @@ def _selector_contract_required_tools( ordered.append(tool_name) return tuple(ordered) + @staticmethod + def _selector_contract_workflow_concept_ids( + workflow_discovery_result: Mapping[str, Any] | None, + ) -> tuple[str, ...]: + if not isinstance(workflow_discovery_result, Mapping): + return () + sources: list[Any] = [ + workflow_discovery_result.get("workflow_concept_ids"), + workflow_discovery_result.get("turn_expected_workflow_concept_ids"), + ] + for key in ( + "contract_projection", + "turn_expected_outcome_contract", + "turn_expected_outcome_contract_state", + ): + raw_payload = workflow_discovery_result.get(key) + if isinstance(raw_payload, Mapping): + sources.append(raw_payload.get("workflow_concept_ids")) + sources.append(raw_payload.get("turn_expected_workflow_concept_ids")) + ordered: list[str] = [] + seen: set[str] = set() + for source in sources: + if isinstance(source, str): + raw_items: Sequence[Any] = (source,) + elif isinstance(source, Sequence): + raw_items = source + else: + continue + for raw_item in raw_items: + if not isinstance(raw_item, str): + continue + workflow_id = raw_item.strip() + if not workflow_id: + continue + dedupe_key = workflow_id.lower() + if dedupe_key in seen: + continue + seen.add(dedupe_key) + ordered.append(workflow_id) + return tuple(ordered) + def _build_selector_default_candidates( self, *, @@ -47233,12 +47413,14 @@ def _build_selector_default_candidates( if isinstance(tool_name, str) and str(tool_name).strip() ) for workflow_id in candidate_ids: - registration = self._workflow_registry.get_registration(workflow_id) - if registration is None: + registration_metadata = self._peek_workflow_registration_metadata( + workflow_id + ) + if not registration_metadata: continue name = workflow_id[3:] if workflow_id.startswith("#V#") else workflow_id name = name.replace("_", " ").strip().title() or workflow_id - description = str(getattr(registration, "purpose", "") or "").strip() + description = str(registration_metadata.get("purpose") or "").strip() item = { "concept_id": workflow_id, "name": name, @@ -47361,11 +47543,57 @@ def _selector_candidate_covers_required_tools( candidate_tool_keys.add(raw_item.strip().lower()) return required_tool_keys.issubset(candidate_tool_keys) + @staticmethod + def _selector_candidate_covers_workflow_execute_obligation( + candidate: Mapping[str, Any], + required_tools: Sequence[str], + ) -> tuple[bool, list[str], list[str]]: + required_tool_names = [ + str(tool_name).strip() + for tool_name in (required_tools or ()) + if isinstance(tool_name, str) and str(tool_name).strip() + ] + required_tool_keys = {tool_name.lower() for tool_name in required_tool_names} + if "workflow_execute" not in required_tool_keys: + return False, [], required_tool_names + if not _eligible_discovered_workflow_execute_candidates((candidate,)): + return False, [], required_tool_names + + covered = [ + tool_name + for tool_name in required_tool_names + if tool_name.lower() == "workflow_execute" + ] + remaining = [ + tool_name + for tool_name in required_tool_names + if tool_name.lower() != "workflow_execute" + ] + return True, covered, remaining + + @staticmethod + def _selector_candidate_is_contract_named_launchable_workflow( + candidate: Mapping[str, Any], + workflow_concept_ids: Sequence[str], + ) -> bool: + concept_id = str(candidate.get("concept_id") or "").strip() + if not concept_id: + return False + contract_workflow_ids = { + str(workflow_id).strip().lower() + for workflow_id in (workflow_concept_ids or ()) + if isinstance(workflow_id, str) and str(workflow_id).strip() + } + if concept_id.lower() not in contract_workflow_ids: + return False + return bool(_eligible_discovered_workflow_execute_candidates((candidate,))) + def _partition_discovered_candidates_by_required_tools( self, candidates: Sequence[Mapping[str, Any]], *, required_tools: Sequence[str], + contract_workflow_concept_ids: Sequence[str] = (), ) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]: if not required_tools: return [dict(candidate) for candidate in candidates], [] @@ -47376,6 +47604,47 @@ def _partition_discovered_candidates_by_required_tools( if self._selector_candidate_covers_required_tools(item, required_tools): included.append(item) continue + ( + covers_workflow_execute, + covered_required_tools, + remaining_required_tools, + ) = self._selector_candidate_covers_workflow_execute_obligation( + item, + required_tools, + ) + if covers_workflow_execute: + item["required_tools"] = [ + str(tool_name).strip() + for tool_name in required_tools + if isinstance(tool_name, str) and str(tool_name).strip() + ] + item["required_tool_coverage"] = { + "coverage_basis": "workflow_execute_primary_obligation", + "covered_required_tools": covered_required_tools, + "remaining_turn_level_required_tools": remaining_required_tools, + } + included.append(item) + continue + if self._selector_candidate_is_contract_named_launchable_workflow( + item, + contract_workflow_concept_ids, + ): + concept_id = str(item.get("concept_id") or "").strip() + item["candidate_reason"] = ( + "contract_named_launchable_workflow_preserved" + ) + item["required_tools"] = [ + str(tool_name).strip() + for tool_name in required_tools + if isinstance(tool_name, str) and str(tool_name).strip() + ] + item["required_tool_coverage"] = { + "coverage_basis": "contract_named_launchable_workflow", + "covered_workflow_concept_ids": [concept_id] if concept_id else [], + "remaining_turn_level_required_tools": list(item["required_tools"]), + } + included.append(item) + continue concept_id = str(item.get("concept_id") or "").strip() if self._selected_workflow_prefers_direct_response( selected_workflow_id=concept_id @@ -47421,12 +47690,16 @@ def _prepare_selector_candidates( required_tools = self._selector_contract_required_tools( workflow_discovery_result ) + contract_workflow_concept_ids = self._selector_contract_workflow_concept_ids( + workflow_discovery_result + ) ( local_discovered_matches, required_tool_excluded_matches, ) = self._partition_discovered_candidates_by_required_tools( local_discovered_matches, required_tools=required_tools, + contract_workflow_concept_ids=contract_workflow_concept_ids, ) default_candidates, excluded_default_candidates = ( self._build_selector_default_candidates(required_tools=required_tools) diff --git a/src/backend/server/routes/auth_routes.py b/src/backend/server/routes/auth_routes.py index 8837abb21..f4bc59896 100644 --- a/src/backend/server/routes/auth_routes.py +++ b/src/backend/server/routes/auth_routes.py @@ -409,8 +409,14 @@ def browser_test_login(): window_session_id = request.headers.get("X-Von-Window-Session") or data.get( "window_session_id" ) + refresh_fixture = data.get("refresh_fixture") + if not isinstance(refresh_fixture, bool): + refresh_fixture = None try: - result = login_browser_test_user(window_session_id=window_session_id) + result = login_browser_test_user( + window_session_id=window_session_id, + refresh_fixture=refresh_fixture, + ) except MultipleUsersForEmailError as exc: return ( jsonify( diff --git a/src/backend/services/browser_test_auth_service.py b/src/backend/services/browser_test_auth_service.py index 16a4c9bd1..fa76382e5 100644 --- a/src/backend/services/browser_test_auth_service.py +++ b/src/backend/services/browser_test_auth_service.py @@ -57,11 +57,13 @@ _LOOPBACK_REMOTES = {"127.0.0.1", "::1", "localhost", ""} _BROWSER_TEST_ENV_KEYS = ( "VON_BROWSER_TEST_AUTH_ENABLED", + "VON_BROWSER_TEST_REFRESH_FIXTURE_ON_LOGIN", "VON_BROWSER_TEST_PSEUDOUSER_NAME", "VON_BROWSER_TEST_PSEUDOUSER_EMAIL", "VON_BROWSER_TEST_PSEUDOUSER_CONCEPT_ID", "VON_BROWSER_TEST_ORGANISATION_CONCEPT_ID", ) +_BROWSER_TEST_FIXTURE_STABLE_SEEDED_AT = "2026-03-14T00:00:00+00:00" _BROWSER_TEST_RECOMMENDATION_PAPER_ID = ( "#V#browser_test_paper_workflow_grounded_neuro_symbolic_planning" ) @@ -105,6 +107,10 @@ def _trimmed_env(name: str) -> str: return str(os.getenv(name) or "").strip() +def _refresh_fixture_on_login_default() -> bool: + return _truthy_env("VON_BROWSER_TEST_REFRESH_FIXTURE_ON_LOGIN", default=False) + + def _browser_test_identity_source() -> str: explicit_identity = any( _trimmed_env(name) @@ -390,7 +396,7 @@ def _build_fixture_message_metadata(*, spec: Mapping[str, Any]) -> dict[str, Any { "browser_test_fixture_id": _BROWSER_TEST_FIXTURE_ID, "browser_test_message_key": str(spec["key"]), - "browser_test_seeded_at": _current_iso(), + "browser_test_seeded_at": _BROWSER_TEST_FIXTURE_STABLE_SEEDED_AT, } ) target_user_concept_id = str(spec.get("target_user_concept_id") or "").strip() @@ -481,6 +487,7 @@ def _paper_recommendation_fixture_spec( "policy_version": PAPER_RECOMMENDATION_POLICY_VERSION, "decision_mode": "browser_test_fixture", "trigger_source": "browser_test_fixture_refresh", + "updated_at": _BROWSER_TEST_FIXTURE_STABLE_SEEDED_AT, "rationale_summary": _BROWSER_TEST_RECOMMENDATION_RATIONALE, "rationale": [_BROWSER_TEST_RECOMMENDATION_RATIONALE], "evidence": [ @@ -839,9 +846,100 @@ def _build_counterpart_specs() -> Iterable[dict[str, str]]: ) +def _skipped_fixture_refresh_payload( + *, + counterpart_docs: list[dict[str, Any]], +) -> dict[str, Any]: + return { + "fixture_id": _BROWSER_TEST_FIXTURE_ID, + "status": "not_refreshed", + "refresh_requested": False, + "refresh_on_login_default": _refresh_fixture_on_login_default(), + "reason": "Browser-test login established an authenticated session without refreshing fixture data.", + "counterparts": counterpart_docs, + "messages": { + "total": 0, + "created": 0, + "reused": 0, + "concept_ids": [], + }, + "chat_sessions": { + "total": 0, + "created_entries": 0, + "sessions": [], + }, + } + + +def _ensure_browser_test_fixture_state( + *, + pseudouser_concept_id: str, + counterpart_docs: list[dict[str, Any]], + namespace: str, + organisation_concept_id: str, + role_in_org: str, +) -> tuple[dict[str, Any], Optional[str]]: + counterpart_by_fixture_concept_id = { + item["fixture_concept_id"]: item for item in counterpart_docs + } + reviewer = counterpart_by_fixture_concept_id["#V#browser_test_workflow_reviewer"] + scout = counterpart_by_fixture_concept_id["#V#browser_test_paper_scout"] + + message_results = [ + _ensure_fixture_message( + spec=spec, + ) + for spec in _message_fixture_specs( + pseudouser_concept_id=pseudouser_concept_id, + reviewer_concept_id=reviewer["concept_id"], + scout_concept_id=scout["concept_id"], + organisation_concept_id=organisation_concept_id, + ) + ] + chat_results = [ + _ensure_fixture_chat_session( + user_concept_id=pseudouser_concept_id, + namespace=namespace, + organisation_concept_id=organisation_concept_id, + role_in_org=role_in_org, + spec=spec, + ) + for spec in _chat_fixture_specs() + ] + active_chat_session_id = chat_results[0]["session_id"] if chat_results else None + created_message_count = sum(1 for item in message_results if item.get("created")) + reused_message_count = len(message_results) - created_message_count + created_history_entries = sum( + int(item.get("created_entries") or 0) for item in chat_results + ) + + return ( + { + "fixture_id": _BROWSER_TEST_FIXTURE_ID, + "status": "refreshed", + "refresh_requested": True, + "refresh_on_login_default": _refresh_fixture_on_login_default(), + "counterparts": counterpart_docs, + "messages": { + "total": len(message_results), + "created": created_message_count, + "reused": reused_message_count, + "concept_ids": [item.get("concept_id") for item in message_results], + }, + "chat_sessions": { + "total": len(chat_results), + "created_entries": created_history_entries, + "sessions": chat_results, + }, + }, + active_chat_session_id, + ) + + def login_browser_test_user( *, window_session_id: Optional[str] = None, + refresh_fixture: Optional[bool] = None, ) -> dict[str, Any]: config = get_browser_test_auth_config() organisation_doc = _ensure_org_exists(config.organisation_concept_id) @@ -914,35 +1012,25 @@ def login_browser_test_user( user_id=pseudouser_concept_id, ) - counterpart_by_fixture_concept_id = { - item["fixture_concept_id"]: item for item in counterpart_docs - } - reviewer = counterpart_by_fixture_concept_id["#V#browser_test_workflow_reviewer"] - scout = counterpart_by_fixture_concept_id["#V#browser_test_paper_scout"] - - message_results = [ - _ensure_fixture_message( - spec=spec, - ) - for spec in _message_fixture_specs( + should_refresh_fixture = ( + _refresh_fixture_on_login_default() + if refresh_fixture is None + else bool(refresh_fixture) + ) + if should_refresh_fixture: + fixture_payload, active_chat_session_id = _ensure_browser_test_fixture_state( pseudouser_concept_id=pseudouser_concept_id, - reviewer_concept_id=reviewer["concept_id"], - scout_concept_id=scout["concept_id"], - organisation_concept_id=config.organisation_concept_id, - ) - ] - chat_results = [ - _ensure_fixture_chat_session( - user_concept_id=pseudouser_concept_id, namespace=namespace, organisation_concept_id=config.organisation_concept_id, role_in_org=role_in_org, - spec=spec, + counterpart_docs=counterpart_docs, ) - for spec in _chat_fixture_specs() - ] + else: + fixture_payload = _skipped_fixture_refresh_payload( + counterpart_docs=counterpart_docs, + ) + active_chat_session_id = None - active_chat_session_id = chat_results[0]["session_id"] if chat_results else None if ( active_chat_session_id and isinstance(window_session_id, str) @@ -954,12 +1042,6 @@ def login_browser_test_user( user_id=pseudouser_concept_id, ) - created_message_count = sum(1 for item in message_results if item.get("created")) - reused_message_count = len(message_results) - created_message_count - created_history_entries = sum( - int(item.get("created_entries") or 0) for item in chat_results - ) - return { "user": { "concept_id": pseudouser_concept_id, @@ -977,21 +1059,7 @@ def login_browser_test_user( "namespace": namespace, "window_session_id": window_session_id, "active_chat_session_id": active_chat_session_id, - "fixture": { - "fixture_id": _BROWSER_TEST_FIXTURE_ID, - "counterparts": counterpart_docs, - "messages": { - "total": len(message_results), - "created": created_message_count, - "reused": reused_message_count, - "concept_ids": [item.get("concept_id") for item in message_results], - }, - "chat_sessions": { - "total": len(chat_results), - "created_entries": created_history_entries, - "sessions": chat_results, - }, - }, + "fixture": fixture_payload, } diff --git a/src/backend/services/text_value_service.py b/src/backend/services/text_value_service.py index db94109d5..4898a068c 100644 --- a/src/backend/services/text_value_service.py +++ b/src/backend/services/text_value_service.py @@ -329,7 +329,10 @@ def upsert_text_for_concept( if relation_created: _invalidate_stats_for_predicate_change(predicate) - if relation_created or context_updated: + if relation_created: + # Workflow routing projections are derived from the relation text and + # predicate, not relation provenance/context. Context-only seed + # refreshes must not churn the capability index. _invalidate_workflow_routing_projection_for_text_relation_change( subject_concept_id=subject_concept_id, predicate=predicate, diff --git a/src/backend/services/workflow_capability_service.py b/src/backend/services/workflow_capability_service.py index 6ef16b836..f0ee5f7ae 100644 --- a/src/backend/services/workflow_capability_service.py +++ b/src/backend/services/workflow_capability_service.py @@ -378,9 +378,6 @@ def _authoritative_registry_fingerprint_rows( { "workflow_id": workflow_id, "source": source, - "purpose": str( - getattr(registration, "purpose", "") or "" - ).strip(), } ) return tuple(sorted(rows_out, key=lambda item: item["workflow_id"])) @@ -1067,31 +1064,29 @@ def load_from_persisted_namespace_if_current(self, registry: Any) -> bool: ) return False - manifest_workflow_ids = tuple( - sorted(str(item) for item in manifest_entries.keys()) - ) - registry_workflow_ids = _authoritative_registry_workflow_ids(registry) current_registry_fingerprint = _authoritative_registry_fingerprint_digest( registry ) manifest_registry_fingerprint = str( manifest.get("registry_fingerprint_digest") or "" ).strip() - if registry_workflow_ids and manifest_workflow_ids != registry_workflow_ids: - _set_workflow_capability_manifest_state( - status="registry_workflow_set_mismatch", - detail=( - "Authoritative workflow registry IDs differ from the " - "materialised routing projection manifest." - ), - path=manifest_path, - digest=manifest_entry_digest, + if not manifest_registry_fingerprint: + manifest_workflow_ids = tuple( + sorted(str(item) for item in manifest_entries.keys()) ) - return False - if ( - manifest_registry_fingerprint - and manifest_registry_fingerprint != current_registry_fingerprint - ): + registry_workflow_ids = _authoritative_registry_workflow_ids(registry) + if registry_workflow_ids and manifest_workflow_ids != registry_workflow_ids: + _set_workflow_capability_manifest_state( + status="registry_workflow_set_mismatch", + detail=( + "Authoritative workflow registry IDs differ from the " + "legacy materialised routing projection manifest." + ), + path=manifest_path, + digest=manifest_entry_digest, + ) + return False + elif manifest_registry_fingerprint != current_registry_fingerprint: _set_workflow_capability_manifest_state( status="registry_fingerprint_mismatch", detail=( @@ -2367,6 +2362,10 @@ def get_workflow_capability_index_readiness_report() -> Dict[str, Any]: **runtime_state, "status": status, "warning_level": warning_level, + "workflow_discovery_available": bool(ready), + "user_visible_blocker": not bool(ready), + "user_visible_severity": "ok" if ready else "error", + "footer_red_flag": not bool(ready), "summary": summary, "detail": detail, "background_initialisation": background_initialisation_event, diff --git a/src/backend/services/workflow_discovery_memo_service.py b/src/backend/services/workflow_discovery_memo_service.py index ceb477788..0411d2314 100644 --- a/src/backend/services/workflow_discovery_memo_service.py +++ b/src/backend/services/workflow_discovery_memo_service.py @@ -147,6 +147,57 @@ def _candidate_count(payload: Mapping[str, Any]) -> int: return 0 +def _has_zero_candidate_operational_blocker(payload: Mapping[str, Any]) -> bool: + if _candidate_count(payload) > 0: + return False + + def _iter_reason_text() -> list[str]: + values: list[str] = [] + for key in ( + "match_absence_reason", + "blocker_reason", + "budget_exhaustion_stage", + "budget_exhaustion_detail", + ): + value = payload.get(key) + if isinstance(value, str) and value.strip(): + values.append(value.strip().lower()) + errors = payload.get("errors") + if isinstance(errors, list): + values.extend( + str(item).strip().lower() + for item in errors + if isinstance(item, str) and str(item).strip() + ) + stage_timings = payload.get("stage_timings") + if isinstance(stage_timings, list): + for stage in stage_timings: + if not isinstance(stage, Mapping): + continue + for key in ("stage", "status", "error"): + value = stage.get(key) + if isinstance(value, str) and value.strip(): + values.append(value.strip().lower()) + stage_errors = stage.get("errors") + if isinstance(stage_errors, list): + values.extend( + str(item).strip().lower() + for item in stage_errors + if isinstance(item, str) and str(item).strip() + ) + return values + + if bool(payload.get("budget_exhausted")): + return True + reasons = _iter_reason_text() + return any( + "capability_index" in reason + or "workflow_discovery_budget_exhausted" in reason + or "query_surface" in reason + for reason in reasons + ) + + def _discovery_ranking_input( *, user_input: str, @@ -366,6 +417,25 @@ def discover_workflows_for_turn_memoized( if clean_user_input: payload["query"] = clean_user_input candidate_count = _candidate_count(payload) + if _has_zero_candidate_operational_blocker(payload): + lookup_elapsed_ms = (time.perf_counter() - lookup_started) * 1000.0 + annotated = _annotate_payload( + payload, + cache_hit=False, + cache_key_digest=cache_key_digest, + turn_scope=turn_scope, + candidate_count=candidate_count, + lookup_elapsed_ms=lookup_elapsed_ms, + capability_index_version=capability_version, + uncached_elapsed_ms=uncached_elapsed_ms, + ) + cache_payload = annotated.setdefault("workflow_discovery_cache", {}) + if isinstance(cache_payload, dict): + cache_payload["cache_skipped_reason"] = ( + "zero_candidate_operational_blocker" + ) + return annotated + with _CACHE_LOCK: _CACHE[cache_key_digest] = _MemoEntry( created_monotonic=time.perf_counter(), diff --git a/src/backend/workflows/durable/registry_factory.py b/src/backend/workflows/durable/registry_factory.py index 9e824a906..00b31b3ec 100644 --- a/src/backend/workflows/durable/registry_factory.py +++ b/src/backend/workflows/durable/registry_factory.py @@ -16,6 +16,7 @@ import logging import os import threading +import time from dataclasses import dataclass from datetime import datetime, timezone from threading import Lock @@ -23,6 +24,7 @@ from .. import WorkflowRegistry from ..engine import WorkflowDefinition +from ..definitions import CONVERSATION_TURN_WORKFLOW_IDS from ..workflow_registry import LazyWorkflowRegistration, WorkflowRegistration from ..action_registry import ActionRegistry, WorkflowActionResult from ..mcp_tool_bridge import ( @@ -59,6 +61,73 @@ _shared_action_registry: ActionRegistry | None = None +def _core_workflow_definition_prewarm_enabled() -> bool: + raw_value = os.getenv("VON_CORE_WORKFLOW_PREWARM_ENABLED") + if raw_value is None: + return True + return _truthy_env_value(raw_value) + + +def _start_core_workflow_definition_prewarm(registry: WorkflowRegistry) -> bool: + """Warm core turn workflow definitions outside the user request path.""" + + if not _core_workflow_definition_prewarm_enabled(): + return False + + workflow_ids = tuple( + workflow_id + for workflow_id in CONVERSATION_TURN_WORKFLOW_IDS + if isinstance(workflow_id, str) and workflow_id.strip() + ) + if not workflow_ids: + return False + + def _prewarm() -> None: + start = time.monotonic() + loaded: list[str] = [] + missing: list[str] = [] + errors: dict[str, str] = {} + for workflow_id in workflow_ids: + try: + definition = registry.get(workflow_id) + if definition is None: + missing.append(workflow_id) + else: + loaded.append(workflow_id) + except Exception as exc: + errors[workflow_id] = type(exc).__name__ + logger.debug( + "[workflow_registry] Core workflow prewarm failed for %s: %s", + workflow_id, + exc, + exc_info=True, + ) + elapsed_ms = (time.monotonic() - start) * 1000.0 + logger.info( + "[workflow_registry] Core workflow prewarm completed loaded=%d " + "missing=%d errors=%d duration_ms=%.0f", + len(loaded), + len(missing), + len(errors), + elapsed_ms, + ) + if missing or errors: + logger.warning( + "[workflow_registry] Core workflow prewarm incomplete missing=%s " + "errors=%s", + missing[:10], + errors, + ) + + thread = threading.Thread( + target=_prewarm, + name="workflow-registry-core-prewarm", + daemon=True, + ) + thread.start() + return True + + def _truthy_env_value(value: str | None) -> bool: return str(value or "").strip().lower() in {"1", "true", "yes", "on", "y"} @@ -180,6 +249,13 @@ def get_shared_workflow_registry_read_only( registry = _shared_workflow_registry if created_registry and registry is not None: + try: + _start_core_workflow_definition_prewarm(registry) + except Exception: + logger.debug( + "shared workflow registry could not start core workflow prewarm", + exc_info=True, + ) try: from ...services.workflow_capability_service import ( prewarm_workflow_capability_index, diff --git a/src/frontend/web/von_interface/static/js/chatTab.js b/src/frontend/web/von_interface/static/js/chatTab.js index 98c104794..ea8f2050c 100644 --- a/src/frontend/web/von_interface/static/js/chatTab.js +++ b/src/frontend/web/von_interface/static/js/chatTab.js @@ -2341,6 +2341,47 @@ function buildWorkflowRoutingDiagnosticsLocatorSnapshot(progressLike) { }; } +function resolveLlmDebugPromptPreview({ debugData, turnExecutionDiagnostics }) { + const candidates = [ + turnExecutionDiagnostics?.prompt_preview, + debugData?.prompt_preview, + debugData?.llm_prompt_preview, + debugData?.prompt_text + ]; + for (const candidate of candidates) { + if (typeof candidate === 'string' && candidate.trim()) { + return candidate; + } + if (candidate && typeof candidate === 'object') { + const text = typeof candidate.text === 'string' ? candidate.text : ''; + if (text.trim()) { + return text; + } + const preview = typeof candidate.preview === 'string' ? candidate.preview : ''; + if (preview.trim()) { + return preview; + } + } + } + return null; +} + +function resolveLlmDebugWorkflowRoutingDiagnostics({ debugData, turnExecutionDiagnostics }) { + const sources = [ + turnExecutionDiagnostics, + debugData, + turnExecutionDiagnostics?.workflow_routing_diagnostics, + debugData?.workflow_routing_diagnostics + ]; + for (const source of sources) { + const routingDiagnostics = getWorkflowRoutingDiagnostics(source); + if (routingDiagnostics) { + return routingDiagnostics; + } + } + return null; +} + function deriveSelectedWorkflowIdFromRoutingDiagnostics(workflowRoutingDiagnostics) { if (!workflowRoutingDiagnostics || typeof workflowRoutingDiagnostics !== 'object') { return null; @@ -32448,14 +32489,16 @@ function buildLlmDebugLocatorPayload({ turnId, debugData, metadata, workflowExec chat_session_id: sessionId, namespace_context: namespaceContext, metadata: metadata || null, - prompt_preview: typeof turnExecutionDiagnostics?.prompt_preview === 'string' - ? turnExecutionDiagnostics.prompt_preview - : (typeof debugData?.prompt_text === 'string' ? debugData.prompt_text : null), + prompt_preview: resolveLlmDebugPromptPreview({ + debugData, + turnExecutionDiagnostics + }), diagnostic_summary: diagnosticSummary, workflow_routing_diagnostics: buildWorkflowRoutingDiagnosticsLocatorSnapshot( - (turnExecutionDiagnostics && typeof turnExecutionDiagnostics === 'object') - ? turnExecutionDiagnostics - : debugData + resolveLlmDebugWorkflowRoutingDiagnostics({ + debugData, + turnExecutionDiagnostics + }) ), mcp_access: {} }; diff --git a/src/frontend/web/von_interface/static/js/domUtils.js b/src/frontend/web/von_interface/static/js/domUtils.js index 3d508ab20..1028abb2b 100644 --- a/src/frontend/web/von_interface/static/js/domUtils.js +++ b/src/frontend/web/von_interface/static/js/domUtils.js @@ -353,10 +353,12 @@ let footerDbRetryTimerId = null; let footerDbLoadGeneration = 0; let footerServerReachability = null; let footerLlmExecutionRefreshScheduled = false; +let footerWorkflowCapabilityRetryTimerId = null; const FOOTER_DB_PROBE_STATS_KEY = 'von_footer_db_probe_stats_v1'; const FOOTER_DB_PROBE_SAMPLE_LIMIT = 32; const FOOTER_DB_RETRY_MIN_MS = 1500; const FOOTER_DB_RETRY_MAX_MS = 20000; +const FOOTER_WORKFLOW_CAPABILITY_RETRY_MS = 60000; function normaliseFooterServerReachability(value) { return (typeof value === 'boolean') ? value : null; @@ -466,6 +468,25 @@ function clearFooterDbRetryTimer() { } } +function clearFooterWorkflowCapabilityRetryTimer() { + if (footerWorkflowCapabilityRetryTimerId) { + clearTimeout(footerWorkflowCapabilityRetryTimerId); + footerWorkflowCapabilityRetryTimerId = null; + } +} + +function scheduleFooterWorkflowCapabilityRefresh() { + clearFooterWorkflowCapabilityRetryTimer(); + const inJest = typeof globalThis.process !== 'undefined' && globalThis.process?.env?.JEST_WORKER_ID; + if (inJest) return; + footerWorkflowCapabilityRetryTimerId = setTimeout(() => { + footerWorkflowCapabilityRetryTimerId = null; + if (document.getElementById('modelInfoFooter')) { + void setModelInfoFooterText(); + } + }, FOOTER_WORKFLOW_CAPABILITY_RETRY_MS); +} + function normaliseFooterTelemetryStrings(values) { if (!Array.isArray(values)) return []; const result = []; @@ -888,6 +909,77 @@ export function openSettingsForModelControls() { return openSettingsTabAndFocus('von:focus-model-settings'); } +export function openSettingsForWorkflowCapabilityIndexStatus() { + return openSettingsTabAndFocus('von:focus-workflow-capability-index-status'); +} + +function normaliseWorkflowCapabilityIndexFooterStatus(report) { + if (!report || typeof report !== 'object') return null; + if (report.ready === true) return { ready: true }; + + const status = (typeof report.status === 'string' && report.status.trim()) + ? report.status.trim() + : 'not_ready'; + const summary = (typeof report.summary === 'string' && report.summary.trim()) + ? report.summary.trim() + : 'Workflow capability index not ready.'; + const detail = (typeof report.detail === 'string' && report.detail.trim()) + ? report.detail.trim() + : 'Represented workflow discovery cannot rely on the authoritative capability index.'; + const namespaceState = report.namespace_state && typeof report.namespace_state === 'object' + ? report.namespace_state + : null; + const namespaceDetail = (typeof namespaceState?.detail === 'string' && namespaceState.detail.trim()) + ? namespaceState.detail.trim() + : ''; + const namespaceStatus = (typeof namespaceState?.status === 'string' && namespaceState.status.trim()) + ? namespaceState.status.trim() + : ''; + const querySurfaceReady = report.query_surface_ready === true; + const userVisibleSeverity = (typeof report.user_visible_severity === 'string' && report.user_visible_severity.trim()) + ? report.user_visible_severity.trim().toLowerCase() + : 'error'; + const severity = userVisibleSeverity === 'ok' ? 'warning' : 'fatal'; + const labelMap = { + building: 'building', + rebuilding: 'rebuilding', + warming: 'warming', + error: 'error', + rebuild_required: 'rebuild required', + not_ready: 'unavailable', + }; + const displayStatus = labelMap[status] || status.replace(/_/g, ' '); + const titleParts = [ + 'Represented workflow discovery is not fully available.', + 'Click to open RAG & workflow index status.', + `Status: ${displayStatus}`, + `Summary: ${summary}`, + `Detail: ${detail}`, + ]; + if (Number.isFinite(Number(report.size))) { + titleParts.push(`Indexed workflow entries: ${Number(report.size)}`); + } + titleParts.push(`Query surface ready: ${querySurfaceReady ? 'yes' : 'no'}`); + if (namespaceStatus) titleParts.push(`Namespace status: ${namespaceStatus}`); + if (namespaceDetail) titleParts.push(`Namespace detail: ${namespaceDetail}`); + if (typeof report.last_error === 'string' && report.last_error.trim()) { + titleParts.push(`Last error: ${report.last_error.trim()}`); + } + if (typeof report.query_surface_last_error === 'string' && report.query_surface_last_error.trim()) { + titleParts.push(`Query warm-up error: ${report.query_surface_last_error.trim()}`); + } + + return { + ready: false, + status, + displayStatus, + severity, + summary, + detail, + title: titleParts.join('\n'), + }; +} + export async function setModelInfoFooterText() { const footer = document.getElementById('modelInfoFooter'); if (!footer) return; @@ -896,6 +988,7 @@ export async function setModelInfoFooterText() { footerDbLoadGeneration += 1; const dbLoadGeneration = footerDbLoadGeneration; clearFooterDbRetryTimer(); + clearFooterWorkflowCapabilityRetryTimer(); const settings = await getSettings(); if (!settings || Object.keys(settings).length === 0) { @@ -941,6 +1034,7 @@ export async function setModelInfoFooterText() { } const executionTelemetry = readLatestLlmExecutionTelemetry(); + const workflowCapabilityStatus = normaliseWorkflowCapabilityIndexFooterStatus(settings.workflow_capability_index); // Determine LLM status styles const status = llmInfo?.status || 'unknown'; @@ -1179,6 +1273,25 @@ export async function setModelInfoFooterText() { segments.push(placeholder); } + if (workflowCapabilityStatus && workflowCapabilityStatus.ready === false) { + const workflowIndexSegment = makeActionButton( + 'Workflow Index', + workflowCapabilityStatus.displayStatus, + () => { openSettingsForWorkflowCapabilityIndexStatus(); }, + { + ariaLabel: 'Open workflow capability index status', + title: workflowCapabilityStatus.title, + } + ); + workflowIndexSegment.classList.add( + 'workflow-index-status-badge', + workflowCapabilityStatus.severity === 'fatal' ? 'fatal' : 'warning', + ); + segments.push(workflowIndexSegment); + readinessIssues.add('workflow capability index'); + scheduleFooterWorkflowCapabilityRefresh(); + } + // Admin safety signal: show a persistent badge when write-tool conservatism is enabled. try { if (Object.prototype.hasOwnProperty.call(settings, 'disable_write_tool_conservatism') && !settings.disable_write_tool_conservatism) { diff --git a/src/frontend/web/von_interface/static/js/settingsPage.js b/src/frontend/web/von_interface/static/js/settingsPage.js index 45053e328..bb5398f81 100644 --- a/src/frontend/web/von_interface/static/js/settingsPage.js +++ b/src/frontend/web/von_interface/static/js/settingsPage.js @@ -651,6 +651,24 @@ function _focusModelSettingsSection() { } catch { } } +function _focusWorkflowCapabilityIndexStatusSection() { + try { + focusSettingsSection('rag-model-settings'); + void refreshRuntimeModelStatus({ force: true }); + + setTimeout(() => { + const card = document.getElementById('workflowCapabilityIndexStatusCard'); + if (!card) return; + card.setAttribute('tabindex', '-1'); + try { + card.focus({ preventScroll: true }); + } catch (_) { + try { card.focus(); } catch { } + } + }, 250); + } catch { } +} + // Allow parent (main app) to request focus on the login area from elsewhere (e.g. chat tabs placeholder). try { window.addEventListener('message', (event) => { @@ -661,6 +679,8 @@ try { _focusCurrentUserSettingsSection(); } else if (type === 'von:focus-model-settings') { _focusModelSettingsSection(); + } else if (type === 'von:focus-workflow-capability-index-status') { + _focusWorkflowCapabilityIndexStatusSection(); } } catch { } }); @@ -1521,6 +1541,10 @@ function applyCapabilityIndexStatusCard(report) { const status = String(report?.status || 'unknown').trim().toLowerCase(); const warningLevel = String(report?.warning_level || '').trim().toLowerCase(); + const userVisibleSeverity = String(report?.user_visible_severity || '').trim().toLowerCase(); + const displayWarningLevel = report?.ready === true + ? 'ok' + : (userVisibleSeverity === 'error' ? 'error' : (warningLevel || 'warning')); const summary = String(report?.summary || 'Workflow capability index status unavailable.').trim(); let detail = String(report?.detail || '').trim(); const namespaceDetail = String(report?.namespace_state?.detail || '').trim(); @@ -1532,7 +1556,8 @@ function applyCapabilityIndexStatusCard(report) { } cardEl.dataset.status = status || 'unknown'; - cardEl.dataset.warningLevel = warningLevel || 'warning'; + cardEl.dataset.warningLevel = displayWarningLevel; + cardEl.dataset.userVisibleSeverity = userVisibleSeverity || displayWarningLevel; summaryEl.textContent = summary; detailEl.textContent = detail; } @@ -2878,6 +2903,10 @@ export async function __testOnly_refreshRuntimeModelStatus(options = {}) { return refreshRuntimeModelStatus(options); } +export function __testOnly_applyCapabilityIndexStatusCard(report) { + return applyCapabilityIndexStatusCard(report); +} + export function __testOnly_resetRuntimeModelStatusCache() { runtimeModelStatusInFlight = null; runtimeModelStatusLastFetchedAt = 0; diff --git a/src/frontend/web/von_interface/static/styles.css b/src/frontend/web/von_interface/static/styles.css index 00b96a0c4..ff1a34562 100644 --- a/src/frontend/web/von_interface/static/styles.css +++ b/src/frontend/web/von_interface/static/styles.css @@ -6134,6 +6134,36 @@ body[data-active-tab="settingsTab"] .tab-content-area { border-color: #fcd34d; } +.workflow-index-status-badge { + display: inline-flex; + align-items: center; + font-size: 12px; + line-height: 1; + padding: 4px 8px; + border-radius: 12px; + border: 1px solid rgba(0, 0, 0, 0.1); +} + +.workflow-index-status-badge.warning { + background: #fef3c7; + color: #92400e; + border-color: #fcd34d; +} + +.workflow-index-status-badge.fatal { + position: relative; + background: #7f1d1d; + color: #ffffff; + border-color: #b91c1c; + animation: dbBadgeFlash 1.2s ease-in-out infinite; +} + +.workflow-index-status-badge.fatal .concept-footer-button { + background: rgba(255, 255, 255, 0.16); + border-color: rgba(255, 255, 255, 0.45); + color: #ffffff; +} + /* Admin override badge (write-tool conservatism disabled) */ .write-policy-override-badge { display: inline-block; diff --git a/tests/backend/test_browser_test_auth_routes.py b/tests/backend/test_browser_test_auth_routes.py index f3130ef11..86714f41b 100644 --- a/tests/backend/test_browser_test_auth_routes.py +++ b/tests/backend/test_browser_test_auth_routes.py @@ -95,7 +95,15 @@ def test_browser_test_login_sets_browser_test_session(monkeypatch, app_client): }, ) - def _fake_login(window_session_id=None): + login_calls = [] + + def _fake_login(window_session_id=None, refresh_fixture=None): + login_calls.append( + { + "window_session_id": window_session_id, + "refresh_fixture": refresh_fixture, + } + ) session["user_email"] = "zhanvonwitbrock@gmail.com" session["google_user_info"] = { "name": "Zhan von Witbrock", @@ -115,6 +123,8 @@ def _fake_login(window_session_id=None): "active_chat_session_id": "browser-fixture-messages-acceptance", "fixture": { "fixture_id": "browser_user_view.v1", + "status": "not_refreshed", + "refresh_requested": False, "messages": {"total": 3, "created": 3, "reused": 0}, }, } @@ -136,3 +146,9 @@ def _fake_login(window_session_id=None): assert payload["organisation"]["concept_id"] == "#V#university_of_auckland_strong_ai_lab" assert payload["window_session_id"] == "ws_browser_test" assert payload["fixture"]["messages"]["total"] == 3 + assert login_calls == [ + { + "window_session_id": "ws_browser_test", + "refresh_fixture": None, + } + ] diff --git a/tests/backend/test_browser_test_auth_service.py b/tests/backend/test_browser_test_auth_service.py index 65d14597b..cd5f2428e 100644 --- a/tests/backend/test_browser_test_auth_service.py +++ b/tests/backend/test_browser_test_auth_service.py @@ -425,7 +425,10 @@ def _fake_ensure_person_concept(*, concept_id, name, email): ) with app.test_request_context("/von/api/auth/browser-test-login"): - result = service.login_browser_test_user(window_session_id="ws_fixture") + result = service.login_browser_test_user( + window_session_id="ws_fixture", + refresh_fixture=True, + ) assert len(message_specs_seen) == 4 sender_ids = {item["sender_id"] for item in message_specs_seen} @@ -444,6 +447,72 @@ def _fake_ensure_person_concept(*, concept_id, name, email): == "Workflow Reviewer Existing Alias" ) assert result["fixture"]["counterparts"][1]["name"] == "Paper Scout Existing Alias" + assert result["fixture"]["status"] == "refreshed" + + +def test_login_browser_test_user_skips_fixture_refresh_by_default(monkeypatch): + app = Flask(__name__) + app.secret_key = "browser-test-secret" + + config = service.BrowserTestAuthConfig( + enabled=True, + pseudouser_name="Zhan von Witbrock", + pseudouser_email="zhanvonwitbrock@gmail.com", + pseudouser_concept_id="#V#zhan_von_witbrock", + organisation_concept_id="#V#university_of_auckland_strong_ai_lab", + ) + + def _fake_ensure_person_concept(*, concept_id, name, email): + return { + "concept_id": concept_id, + "direct_concept_name": name, + } + + fixture_calls: list[str] = [] + + monkeypatch.setattr(service, "get_browser_test_auth_config", lambda: config) + monkeypatch.setattr(service, "_refresh_fixture_on_login_default", lambda: False) + monkeypatch.setattr( + service, + "_ensure_org_exists", + lambda organisation_concept_id: { + "concept_id": organisation_concept_id, + "direct_concept_name": "University Of Auckland Strong AI Lab", + }, + ) + monkeypatch.setattr(service, "_ensure_person_concept", _fake_ensure_person_concept) + monkeypatch.setattr(service, "_ensure_org_membership", lambda **kwargs: None) + monkeypatch.setattr( + service, + "derive_namespace", + lambda user_slug, org_slug: f"#V#{user_slug}@{org_slug}", + ) + monkeypatch.setattr(service, "set_window_organisation", lambda **kwargs: None) + monkeypatch.setattr( + service, + "set_window_chat_session", + lambda **kwargs: fixture_calls.append("chat_session"), + ) + monkeypatch.setattr( + service, + "_ensure_fixture_message", + lambda *, spec: fixture_calls.append("message"), + ) + monkeypatch.setattr( + service, + "_ensure_fixture_chat_session", + lambda **kwargs: fixture_calls.append("chat_history"), + ) + + with app.test_request_context("/von/api/auth/browser-test-login"): + result = service.login_browser_test_user(window_session_id="ws_fixture") + + assert fixture_calls == [] + assert result["active_chat_session_id"] is None + assert result["fixture"]["status"] == "not_refreshed" + assert result["fixture"]["refresh_requested"] is False + assert result["fixture"]["messages"]["total"] == 0 + assert result["fixture"]["chat_sessions"]["total"] == 0 def test_describe_browser_test_mode_surfaces_disabled_setup_hint_and_default_identity( diff --git a/tests/backend/test_google_auth_service.py b/tests/backend/test_google_auth_service.py new file mode 100644 index 000000000..1832b9bf3 --- /dev/null +++ b/tests/backend/test_google_auth_service.py @@ -0,0 +1,99 @@ +from __future__ import annotations + +import os +import types + +import pytest + +from src.backend import auth_service +from src.backend.auth_service import GoogleAuthService + + +class _DummyFlow: + def __init__(self) -> None: + self.credentials = types.SimpleNamespace(id_token="dummy-id-token") + self.redirect_uri = None + self.client_config = {"web": {"redirect_uris": []}} + self.fetched_authorization_response = None + + def fetch_token(self, *, authorization_response: str) -> None: + self.fetched_authorization_response = authorization_response + + +def _install_dummy_flow(monkeypatch: pytest.MonkeyPatch) -> _DummyFlow: + dummy_flow = _DummyFlow() + + def _from_client_config(*args, **kwargs): + del args, kwargs + return dummy_flow + + monkeypatch.setattr( + auth_service.Flow, + "from_client_config", + staticmethod(_from_client_config), + ) + return dummy_flow + + +def _set_required_google_oauth_env(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("GOOGLE_OAUTH_CLIENT_ID", "client-id") + monkeypatch.setenv("GOOGLE_OAUTH_CLIENT_SECRET", "client-secret") + monkeypatch.setenv( + "GOOGLE_OAUTH_REDIRECT_URI", + "http://localhost:5001/von/api/auth/google/callback", + ) + + +def test_google_oauth_verifier_uses_small_default_clock_skew( + monkeypatch: pytest.MonkeyPatch, +) -> None: + _set_required_google_oauth_env(monkeypatch) + monkeypatch.delenv("GOOGLE_OAUTH_CLOCK_SKEW_SECONDS", raising=False) + dummy_flow = _install_dummy_flow(monkeypatch) + + verify_kwargs: dict[str, object] = {} + + def _verify_oauth2_token(**kwargs): + verify_kwargs.update(kwargs) + return {"email": "user@example.test", "name": "User"} + + monkeypatch.setattr(auth_service.id_token, "verify_oauth2_token", _verify_oauth2_token) + + previous = os.environ.get("OAUTHLIB_RELAX_TOKEN_SCOPE") + os.environ.pop("OAUTHLIB_RELAX_TOKEN_SCOPE", None) + try: + result = GoogleAuthService().exchange_code_for_tokens( + "http://localhost:5001/von/api/auth/google/callback?code=abc&state=xyz" + ) + finally: + if previous is None: + os.environ.pop("OAUTHLIB_RELAX_TOKEN_SCOPE", None) + else: + os.environ["OAUTHLIB_RELAX_TOKEN_SCOPE"] = previous + + assert result["email"] == "user@example.test" + assert dummy_flow.fetched_authorization_response.endswith("code=abc&state=xyz") + assert verify_kwargs["id_token"] == "dummy-id-token" + assert verify_kwargs["audience"] == "client-id" + assert verify_kwargs["clock_skew_in_seconds"] == 10 + assert os.environ.get("OAUTHLIB_RELAX_TOKEN_SCOPE") is None + + +@pytest.mark.parametrize( + ("raw_value", "expected"), + [ + ("2", 2), + ("999", 300), + ("not-an-int", 10), + ], +) +def test_google_oauth_clock_skew_env_override_is_bounded( + monkeypatch: pytest.MonkeyPatch, + raw_value: str, + expected: int, +) -> None: + _set_required_google_oauth_env(monkeypatch) + monkeypatch.setenv("GOOGLE_OAUTH_CLOCK_SKEW_SECONDS", raw_value) + _install_dummy_flow(monkeypatch) + + assert GoogleAuthService().clock_skew_seconds == expected diff --git a/tests/backend/test_mongo_dns_fallback_recovery.py b/tests/backend/test_mongo_dns_fallback_recovery.py index fbf792819..0103d156e 100644 --- a/tests/backend/test_mongo_dns_fallback_recovery.py +++ b/tests/backend/test_mongo_dns_fallback_recovery.py @@ -1,5 +1,7 @@ from __future__ import annotations +import time + from src.backend.db import mongo_client as mc @@ -61,3 +63,58 @@ def _fake_new_client(uri: str, **kwargs): assert ( mc.get_effective_mongo_uri() == "mongodb://direct-host.example:27017/?tls=true" ) + fallback_state = mc.get_mongo_fallback_policy_state() + assert fallback_state["dns_fallback_sticky"] is True + assert fallback_state["dns_fallback_configured"] is True + + +def test_get_db_prefers_recent_successful_dns_fallback(monkeypatch) -> None: + created: list[tuple[str, dict]] = [] + fallback_client = _FakeClient(label="dns-fallback") + + def _fake_new_client(uri: str, **kwargs): + created.append((uri, dict(kwargs))) + if uri == "mongodb://direct-host.example:27017/?tls=true": + return fallback_client + if uri == "mongodb+srv://primary.example/von_db": + raise AssertionError("Primary SRV URI should not be retried during sticky fallback window.") + raise AssertionError(f"Unexpected URI: {uri}") + + monkeypatch.setattr(mc, "_new_mongo_client", _fake_new_client) + monkeypatch.setattr(mc, "is_mock_db_enabled", lambda: False) + monkeypatch.setattr(mc, "assert_safe_database_name_for_pytest", lambda _: None) + monkeypatch.setattr(mc, "MONGO_URI", "mongodb+srv://primary.example/von_db") + monkeypatch.setattr( + mc, + "MONGO_DNS_FALLBACK_URI", + "mongodb://direct-host.example:27017/?tls=true", + ) + monkeypatch.setattr(mc, "MONGO_ALLOW_LOCAL_FALLBACK", False) + monkeypatch.setattr(mc, "_mongo_client_real", None) + monkeypatch.setattr(mc, "_mongo_client_mock", None) + monkeypatch.setattr(mc, "_using_fallback_real", False) + monkeypatch.setattr(mc, "_effective_uri_real", mc.MONGO_URI) + monkeypatch.setattr( + mc, "_dns_fallback_preferred_until_monotonic", time.monotonic() + 60.0 + ) + monkeypatch.setattr( + mc, + "_dns_fallback_preference_reason", + "primary DNS/SRV error", + ) + monkeypatch.setenv("VON_DB_NAME", "von_db") + + db = mc.get_db() + + assert db is not None + assert db["label"] == "dns-fallback" + assert created == [ + ( + "mongodb://direct-host.example:27017/?tls=true", + {"server_selection_timeout_ms": 3000}, + ) + ] + assert mc.is_using_fallback_uri() is True + assert ( + mc.get_effective_mongo_uri() == "mongodb://direct-host.example:27017/?tls=true" + ) diff --git a/tests/backend/test_orchestrator_selector_core_routing.py b/tests/backend/test_orchestrator_selector_core_routing.py index 53e4afe04..7d04dba80 100644 --- a/tests/backend/test_orchestrator_selector_core_routing.py +++ b/tests/backend/test_orchestrator_selector_core_routing.py @@ -548,6 +548,263 @@ def test_workflow_execute_contract_keeps_launchable_discovered_workflow(monkeypa assert not any(candidate.get("concept_id") == workflow_id for candidate in excluded) +def test_workflow_execute_with_readback_tools_keeps_discovered_workflow_candidate( + monkeypatch, +): + """Read-back obligations should not hide launchable represented workflows.""" + + orchestrator = _build_orchestrator(monkeypatch, selector_enabled=True) + workflow_id = "#V#source_neutral_paper_reference_ingestion_workflow" + _register_terminal_custom_workflow( + orchestrator, + workflow_id=workflow_id, + purpose="Ingest a paper reference and produce represented artefacts.", + ) + + discovered, excluded, selector_candidates = ( + orchestrator._prepare_selector_candidates( + workflow_discovery_result={ + "matches": [ + { + "concept_id": workflow_id, + "name": "Source Neutral Paper Reference Ingestion Workflow", + "description": ( + "Canonical entry workflow for ingesting scholarly paper " + "references and producing represented paper artefacts." + ), + "is_executable": True, + "executability_reason": "executable_now", + "is_policy_safe": True, + "routing_eligible": True, + "turn_launchable": True, + "candidate_source": "workflow_discovery", + "match_source": "capability_index", + "routing_profile": {"role": "execution"}, + } + ], + "match_count": 1, + "contract_projection": { + "required_tools": [ + "workflow_execute", + "workflow_get_execution_trace", + "workflow_get_instance", + "rag_list_indexed", + "rag_get_item", + ], + }, + }, + prompt=( + "Please ingest https://arxiv.org/abs/2406.15341 into Von, then " + "read back the represented paper concept, file-copy, and blob evidence." + ), + ) + ) + + assert [candidate["concept_id"] for candidate in discovered] == [workflow_id] + assert [candidate["concept_id"] for candidate in selector_candidates][:1] == [ + workflow_id + ] + represented_candidate = selector_candidates[0] + assert represented_candidate["required_tool_coverage"] == { + "coverage_basis": "workflow_execute_primary_obligation", + "covered_required_tools": ["workflow_execute"], + "remaining_turn_level_required_tools": [ + "workflow_get_execution_trace", + "workflow_get_instance", + "rag_list_indexed", + "rag_get_item", + ], + } + assert not any(candidate.get("concept_id") == workflow_id for candidate in excluded) + + +def test_contract_named_launchable_workflow_survives_internal_tool_contract_shape( + monkeypatch, +): + """Expected-outcome internal-tool wording must not veto workflow initiation.""" + + orchestrator = _build_orchestrator(monkeypatch, selector_enabled=True) + workflow_id = "#V#contract_named_paper_ingestion_workflow" + _register_terminal_custom_workflow( + orchestrator, + workflow_id=workflow_id, + purpose="Ingest a paper reference through a represented workflow.", + ) + + discovered, excluded, selector_candidates = ( + orchestrator._prepare_selector_candidates( + workflow_discovery_result={ + "matches": [ + { + "concept_id": workflow_id, + "name": "Contract Named Paper Ingestion Workflow", + "description": ( + "Ingest a paper reference through a represented workflow." + ), + "is_executable": True, + "executability_reason": "executable_now", + "is_policy_safe": True, + "routing_eligible": True, + "turn_launchable": True, + "candidate_source": "workflow_discovery", + "match_source": "contract_direct_workflow_resolution", + "routing_profile": {"role": "execution"}, + } + ], + "match_count": 1, + "contract_projection": { + "required_tools": [ + "paper:download_source", + "paper:finalise_cached_source", + "vontology:read_file_copy", + ], + "conditional_required_tools": ["workflow_execute"], + "workflow_concept_ids": [workflow_id], + }, + }, + prompt=( + "Please ingest this paper into Von, then read back the represented " + "paper concept, file-copy, and blob evidence." + ), + ) + ) + + assert [candidate["concept_id"] for candidate in discovered] == [workflow_id] + assert [candidate["concept_id"] for candidate in selector_candidates][:1] == [ + workflow_id + ] + represented_candidate = selector_candidates[0] + assert represented_candidate["candidate_reason"] == ( + "contract_named_launchable_workflow_preserved" + ) + assert represented_candidate["required_tool_coverage"] == { + "coverage_basis": "contract_named_launchable_workflow", + "covered_workflow_concept_ids": [workflow_id], + "remaining_turn_level_required_tools": [ + "paper:download_source", + "paper:finalise_cached_source", + "vontology:read_file_copy", + ], + } + assert not any(candidate.get("concept_id") == workflow_id for candidate in excluded) + + +def test_contract_named_nonlaunchable_workflow_still_fails_required_tool_gate( + monkeypatch, +): + """Contract naming is not enough when the workflow cannot launch this turn.""" + + orchestrator = _build_orchestrator(monkeypatch, selector_enabled=True) + workflow_id = "#V#contract_named_unlaunchable_ingestion_workflow" + _register_terminal_custom_workflow( + orchestrator, + workflow_id=workflow_id, + purpose="Ingest a paper reference through a represented workflow.", + ) + + discovered, excluded, selector_candidates = ( + orchestrator._prepare_selector_candidates( + workflow_discovery_result={ + "matches": [ + { + "concept_id": workflow_id, + "name": "Contract Named Unlaunchable Ingestion Workflow", + "description": ( + "Ingest a paper reference through a represented workflow." + ), + "is_executable": True, + "executability_reason": "executable_now", + "is_policy_safe": True, + "routing_eligible": True, + "turn_launchable": False, + "candidate_source": "workflow_discovery", + "match_source": "contract_direct_workflow_resolution", + "routing_profile": {"role": "execution"}, + } + ], + "match_count": 1, + "contract_projection": { + "required_tools": [ + "paper:download_source", + "paper:finalise_cached_source", + ], + "conditional_required_tools": ["workflow_execute"], + "workflow_concept_ids": [workflow_id], + }, + }, + prompt="Please ingest the relevant paper if you can resolve it.", + ) + ) + + assert not any(candidate.get("concept_id") == workflow_id for candidate in discovered) + assert not any( + candidate.get("concept_id") == workflow_id for candidate in selector_candidates + ) + assert any( + candidate.get("concept_id") == workflow_id + and candidate.get("routing_exclusion_reason") + == "required_tool_contract_not_satisfied" + for candidate in excluded + ) + + +def test_zero_candidate_capability_index_blocker_forces_discovery_refresh( + monkeypatch, +): + """A transiently unavailable discovery surface must not become selector truth.""" + + orchestrator = _build_orchestrator(monkeypatch, selector_enabled=True) + stale_discovery = { + "query": "Represent this arXiv paper.", + "requested_query": "Represent this arXiv paper.", + "matches": [], + "candidates": [], + "candidate_count": 0, + "match_count": 0, + "errors": [ + "capability_index_wait_timed_out", + "capability_index_build_in_progress", + ], + "match_absence_reason": "capability_index_wait_timed_out_build_in_progress", + "stage_timings": [ + { + "stage": "capability_index_unavailable", + "errors": [ + "capability_index_wait_timed_out", + "capability_index_build_in_progress", + ], + } + ], + } + + assert orchestrator._should_refresh_turn_workflow_discovery_result( + workflow_discovery_result=stale_discovery, + requested_query="Represent this arXiv paper.", + discovery_query_input="Represent this arXiv paper.", + ) + + +def test_clean_zero_candidate_discovery_can_be_reused(monkeypatch): + """A real no-match result is different from operational unavailability.""" + + orchestrator = _build_orchestrator(monkeypatch, selector_enabled=True) + clean_no_match = { + "query": "hello", + "requested_query": "hello", + "matches": [], + "candidates": [], + "candidate_count": 0, + "match_count": 0, + "match_absence_reason": "no_relevant_workflow", + } + + assert not orchestrator._should_refresh_turn_workflow_discovery_result( + workflow_discovery_result=clean_no_match, + requested_query="hello", + discovery_query_input="hello", + ) + + def test_top_level_selector_applies_represented_fast_path_before_llm(monkeypatch): """Represented candidate policy should route the top-level path without selector LLM drift.""" diff --git a/tests/backend/test_pipeline_stage_coherence.py b/tests/backend/test_pipeline_stage_coherence.py index f3bd513d5..7eb0d8b99 100644 --- a/tests/backend/test_pipeline_stage_coherence.py +++ b/tests/backend/test_pipeline_stage_coherence.py @@ -20,6 +20,11 @@ WorkflowRegistration, WorkflowStateSpec, ) +from src.backend.workflows.definitions import ( + CHAT_ASSISTANT_WORKFLOW_ID, + CHAT_NARRATION_WORKFLOW_ID, + TOOL_CALLING_WORKFLOW_ID, +) from src.backend.workflows.workflow_registry import WorkflowRegistry from orchestrator_test_harness import build_db_independent_orchestrator @@ -165,6 +170,79 @@ def test_non_executable_candidate_excluded_even_with_jit( assert len(excluded) == 1 assert "non_executable" in excluded[0].get("routing_exclusion_reason", "") + def test_missing_routing_profile_does_not_materialise_lazy_candidate( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + orchestrator = _build_orchestrator(monkeypatch) + + def _unexpected_lazy_load(workflow_id: str) -> WorkflowDefinition: + raise AssertionError(f"selector preparation loaded {workflow_id}") + + registry = WorkflowRegistry(definition_loader=_unexpected_lazy_load) + registry.register_lazy( + LazyWorkflowRegistration( + workflow_id="#V#lazy_discovered_workflow", + purpose="Lazy discovered workflow", + source="vontology", + ) + ) + orchestrator._workflow_registry = registry + + included, excluded = orchestrator._prepare_selector_discovered_matches( + { + "candidates": [ + { + "concept_id": "#V#lazy_discovered_workflow", + "description": "Lazy discovered workflow", + "is_executable": True, + "executability_reason": "executable_now", + "routing_eligible": True, + } + ] + }, + turn_text="use the lazy discovered workflow", + ) + + assert [item["concept_id"] for item in included] == [ + "#V#lazy_discovered_workflow" + ] + assert excluded == [] + + +class TestSelectorDefaultCandidates: + def test_default_candidates_do_not_materialise_lazy_workflows( + self, monkeypatch: pytest.MonkeyPatch + ) -> None: + orchestrator = _build_orchestrator(monkeypatch) + + def _unexpected_lazy_load(workflow_id: str) -> WorkflowDefinition: + raise AssertionError(f"default candidate builder loaded {workflow_id}") + + registry = WorkflowRegistry(definition_loader=_unexpected_lazy_load) + for workflow_id in ( + CHAT_ASSISTANT_WORKFLOW_ID, + TOOL_CALLING_WORKFLOW_ID, + CHAT_NARRATION_WORKFLOW_ID, + ): + registry.register_lazy( + LazyWorkflowRegistration( + workflow_id=workflow_id, + purpose=f"Purpose for {workflow_id}", + source="vontology", + ) + ) + orchestrator._workflow_registry = registry + + candidates, excluded = orchestrator._build_selector_default_candidates() + + assert [item["concept_id"] for item in candidates] == [ + CHAT_ASSISTANT_WORKFLOW_ID, + TOOL_CALLING_WORKFLOW_ID, + CHAT_NARRATION_WORKFLOW_ID, + ] + assert excluded == [] + assert candidates[1]["description"] == f"Purpose for {TOOL_CALLING_WORKFLOW_ID}" + # --------------------------------------------------------------------------- # Bug 2: Dispatch identity divergence (JVNAUTOSCI-1809) diff --git a/tests/backend/test_text_relations_summary_and_singleton.py b/tests/backend/test_text_relations_summary_and_singleton.py index 7b5352430..e17146237 100644 --- a/tests/backend/test_text_relations_summary_and_singleton.py +++ b/tests/backend/test_text_relations_summary_and_singleton.py @@ -144,6 +144,65 @@ def test_upsert_singleton_text_relation_replaces_others_for_language(): assert len(remaining) == 1 +def test_workflow_routing_text_context_update_does_not_invalidate_projection( + monkeypatch: pytest.MonkeyPatch, +): + if TextValuesRepository.db() is None: + pytest.skip("MongoDB not configured for this test run") + + concept_id = "#V#workflow_context_update_test" + predicate = "#V#hasWorkflowDiscoveryExemplarsJson" + concepts = ConceptsRepository.collection() + if concepts is None: + pytest.skip("MongoDB not configured for this test run") + + concepts.insert_one({"concept_id": concept_id, "relationships": {}}) + invalidations: list[str | None] = [] + discovery_cache_clears: list[bool] = [] + + monkeypatch.setattr( + "src.backend.services.workflow_capability_service.invalidate_workflow_capability_index", + lambda **kwargs: ( + invalidations.append(kwargs.get("reason")) or {"success": True} + ), + ) + monkeypatch.setattr( + "src.backend.services.workflow_discovery_service.invalidate_workflow_discovery_executability_caches", + lambda: discovery_cache_clears.append(True), + ) + monkeypatch.setattr( + "src.backend.services.workflow_capability_service.is_authoritative_workflow_concept_id", + lambda subject_concept_id: subject_concept_id == concept_id, + ) + + text = '{"schema_version":"workflow_discovery_exemplars.v1","keywords":["demo"]}' + with bypass_access_control(): + created = upsert_text_for_concept( + subject_concept_id=concept_id, + predicate=predicate, + text=text, + lang="en-NZ", + context={"source": "first"}, + ) + context_only_update = upsert_text_for_concept( + subject_concept_id=concept_id, + predicate=predicate, + text=text, + lang="en-NZ", + context={"source": "second"}, + ) + + assert created["relation_created"] is True + assert created["context_updated"] is False + assert context_only_update["relation_created"] is False + assert context_only_update["context_updated"] is True + assert context_only_update["text_value_id"] == created["text_value_id"] + assert invalidations == [ + f"workflow_routing_text_relation_changed:{concept_id}:{predicate}" + ] + assert discovery_cache_clears == [True] + + def test_get_texts_for_concept_can_return_recent_rows_first(): if TextValuesRepository.db() is None: pytest.skip("MongoDB not configured for this test run") diff --git a/tests/backend/test_workflow_capability_service.py b/tests/backend/test_workflow_capability_service.py index 03e1149fd..318ae012b 100644 --- a/tests/backend/test_workflow_capability_service.py +++ b/tests/backend/test_workflow_capability_service.py @@ -827,10 +827,68 @@ def test_blocking_build_reuses_current_persisted_manifest_without_backend_reset( assert readiness["last_manifest_status"] == "loaded" assert readiness["ready"] is True - def test_blocking_build_rebuilds_when_authoritative_manifest_digest_changes( + def test_manifest_reuse_allows_intentionally_unindexed_registry_workflows( self, tmp_path: Any, _fake_retrieval_backend: _FakeWorkflowRetrievalBackend, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + import src.backend.services.workflow_capability_service as capability_service + from src.backend.workflows import WorkflowRegistry + from src.backend.workflows.workflow_registry import LazyWorkflowRegistration + + _enable_fake_backend_persistence(_fake_retrieval_backend, tmp_path) + registry = WorkflowRegistry() + registry.register_lazy( + LazyWorkflowRegistration( + workflow_id="#V#indexable_workflow", + purpose="This workflow has authoritative routing text.", + source="vontology", + ) + ) + registry.register_lazy( + LazyWorkflowRegistration( + workflow_id="#V#textless_workflow", + purpose="", + source="vontology", + ) + ) + + first_index = capability_service._perform_workflow_capability_index_build( + mode="blocking", + workflow_registry=registry, + ) + assert first_index.size == 1 + + reset_workflow_capability_index() + monkeypatch.setattr( + capability_service.WorkflowCapabilityIndex, + "_entries_from_registry", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError( + "matching current registry fingerprint should load the " + "manifest even when some registry workflows are not indexable" + ) + ), + ) + + second_index = capability_service._perform_workflow_capability_index_build( + mode="blocking", + workflow_registry=registry, + ) + + assert second_index.size == 1 + assert _fake_retrieval_backend.reset_calls == ["workflow_capabilities"] + assert len(_fake_retrieval_backend.upsert_calls) == 1 + readiness = get_workflow_capability_index_readiness_report() + assert readiness["last_manifest_status"] == "loaded" + assert readiness["ready"] is True + + def test_manifest_reuse_ignores_lazy_loaded_purpose_metadata_change( + self, + tmp_path: Any, + _fake_retrieval_backend: _FakeWorkflowRetrievalBackend, + monkeypatch: pytest.MonkeyPatch, ) -> None: import src.backend.services.workflow_capability_service as capability_service from src.backend.workflows import WorkflowRegistry @@ -856,6 +914,62 @@ def test_blocking_build_rebuilds_when_authoritative_manifest_digest_changes( second_registry.register_lazy( LazyWorkflowRegistration( workflow_id="#V#workflow_repair_or_create_workflow", + purpose="Runtime-populated purpose after lazy definition load.", + source="vontology", + ) + ) + monkeypatch.setattr( + capability_service.WorkflowCapabilityIndex, + "_entries_from_registry", + lambda *_args, **_kwargs: (_ for _ in ()).throw( + AssertionError( + "lazy-loaded purpose metadata must not make a current " + "manifest look stale" + ) + ), + ) + + second_index = capability_service._perform_workflow_capability_index_build( + mode="blocking", + workflow_registry=second_registry, + ) + + assert second_index.size == 1 + assert _fake_retrieval_backend.reset_calls == ["workflow_capabilities"] + assert len(_fake_retrieval_backend.upsert_calls) == 1 + readiness = get_workflow_capability_index_readiness_report() + assert readiness["last_manifest_status"] == "loaded" + assert readiness["ready"] is True + + def test_blocking_build_rebuilds_when_authoritative_workflow_set_changes( + self, + tmp_path: Any, + _fake_retrieval_backend: _FakeWorkflowRetrievalBackend, + ) -> None: + import src.backend.services.workflow_capability_service as capability_service + from src.backend.workflows import WorkflowRegistry + from src.backend.workflows.workflow_registry import LazyWorkflowRegistration + + _enable_fake_backend_persistence(_fake_retrieval_backend, tmp_path) + + first_registry = WorkflowRegistry() + first_registry.register_lazy( + LazyWorkflowRegistration( + workflow_id="#V#workflow_repair_or_create_workflow", + purpose="Repair workflows from requests.", + source="vontology", + ) + ) + capability_service._perform_workflow_capability_index_build( + mode="blocking", + workflow_registry=first_registry, + ) + + reset_workflow_capability_index() + second_registry = WorkflowRegistry() + second_registry.register_lazy( + LazyWorkflowRegistration( + workflow_id="#V#different_authoritative_workflow", purpose="Repair or create workflows from current user requests.", source="vontology", ) @@ -1182,6 +1296,10 @@ def test_startup_check_records_not_ready_report( readiness = get_workflow_capability_index_readiness_report() assert readiness["status"] == "not_ready" assert readiness["startup_check"]["status"] == "not_ready" + assert readiness["workflow_discovery_available"] is False + assert readiness["user_visible_blocker"] is True + assert readiness["user_visible_severity"] == "error" + assert readiness["footer_red_flag"] is True def test_readiness_report_requires_query_surface_warmth() -> None: @@ -1194,6 +1312,8 @@ def test_readiness_report_requires_query_surface_warmth() -> None: assert readiness["ready"] is False assert readiness["status"] == "warming" assert readiness["summary"] == "Workflow capability index query surface warming." + assert readiness["workflow_discovery_available"] is False + assert readiness["user_visible_severity"] == "error" def test_startup_check_warms_query_surface_for_ready_namespace( diff --git a/tests/backend/test_workflow_discovery_memo_service.py b/tests/backend/test_workflow_discovery_memo_service.py index ac121162b..636fe1946 100644 --- a/tests/backend/test_workflow_discovery_memo_service.py +++ b/tests/backend/test_workflow_discovery_memo_service.py @@ -146,6 +146,59 @@ def _discover(user_input: str, **_kwargs: Any) -> dict[str, Any]: assert calls == 2 +def test_turn_workflow_discovery_memo_does_not_cache_operational_empty_result( + monkeypatch, +) -> None: + monkeypatch.setattr( + "src.backend.services.workflow_capability_service.get_workflow_capability_index_runtime_state", + lambda *, latency_sensitive=False: _runtime_state("v1"), + ) + calls = 0 + + def _discover(user_input: str, **_kwargs: Any) -> dict[str, Any]: + nonlocal calls + calls += 1 + return { + "query": user_input, + "matches": [], + "candidates": [], + "candidate_count": 0, + "match_count": 0, + "errors": [ + "capability_index_wait_timed_out", + "capability_index_build_in_progress", + ], + "match_absence_reason": ( + "capability_index_wait_timed_out_build_in_progress" + ), + } + + first = discover_workflows_for_turn_memoized( + "Represent this arXiv paper.", + namespace="#V#user@org", + turn_scope="turn-1", + discovery_func=_discover, + ) + second = discover_workflows_for_turn_memoized( + "Represent this arXiv paper.", + namespace="#V#user@org", + turn_scope="turn-1", + discovery_func=_discover, + ) + + assert calls == 2 + assert first is not None + assert second is not None + assert first["workflow_discovery_cache"]["cache_hit"] is False + assert second["workflow_discovery_cache"]["cache_hit"] is False + assert first["workflow_discovery_cache"]["cache_skipped_reason"] == ( + "zero_candidate_operational_blocker" + ) + assert second["workflow_discovery_cache"]["cache_skipped_reason"] == ( + "zero_candidate_operational_blocker" + ) + + def test_turn_workflow_discovery_memo_ranks_with_requested_query( monkeypatch, ) -> None: diff --git a/tests/backend/test_workflow_registry_factory_parity.py b/tests/backend/test_workflow_registry_factory_parity.py index c5d82ce46..a845b61f6 100644 --- a/tests/backend/test_workflow_registry_factory_parity.py +++ b/tests/backend/test_workflow_registry_factory_parity.py @@ -53,6 +53,7 @@ def test_get_shared_workflow_registry_read_only_starts_capability_index_warmup( ): sentinel_registry = object() observed: list[tuple[bool, object | None]] = [] + core_prewarm: list[object] = [] monkeypatch.setattr(registry_factory, "_shared_workflow_registry", None) monkeypatch.setattr( @@ -62,6 +63,11 @@ def test_get_shared_workflow_registry_read_only_starts_capability_index_warmup( sentinel_registry ), ) + monkeypatch.setattr( + registry_factory, + "_start_core_workflow_definition_prewarm", + lambda registry: core_prewarm.append(registry) or True, + ) monkeypatch.setattr( "src.backend.services.workflow_capability_service.prewarm_workflow_capability_index", lambda *, force_refresh=False, workflow_registry=None: ( @@ -75,9 +81,42 @@ def test_get_shared_workflow_registry_read_only_starts_capability_index_warmup( ) assert result is sentinel_registry + assert core_prewarm == [sentinel_registry] assert observed == [(False, sentinel_registry)] +def test_core_workflow_definition_prewarm_resolves_conversation_turn_family( + monkeypatch, +): + warmed: list[str] = [] + + class _FakeRegistry: + def get(self, workflow_id: str): + warmed.append(workflow_id) + return object() + + class _SynchronousThread: + def __init__(self, *, target, name=None, daemon=None): + self._target = target + self.name = name + self.daemon = daemon + + def start(self): + self._target() + + monkeypatch.setattr( + registry_factory, + "_core_workflow_definition_prewarm_enabled", + lambda: True, + ) + monkeypatch.setattr(registry_factory.threading, "Thread", _SynchronousThread) + + started = registry_factory._start_core_workflow_definition_prewarm(_FakeRegistry()) + + assert started is True + assert warmed == list(registry_factory.CONVERSATION_TURN_WORKFLOW_IDS) + + def test_invalidate_shared_workflow_registry_read_only_resets_capability_index( monkeypatch, ): diff --git a/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js b/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js index 98e5315e1..94ac44333 100644 --- a/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js +++ b/tests/frontend/chatTabLlmDebugWorkflowExecution.test.js @@ -358,6 +358,74 @@ describe('LLM debug popup workflow execution hook', () => { expect(payload.messages).toBeUndefined(); }); + test('locator falls back to hydrated top-level prompt and routing diagnostics', async () => { + const { setLlmDebugDataForTurn, showLlmDebugPopup } = require(chatTabModulePath); + + setLlmDebugDataForTurn('assistant-top-level-diagnostics', { + model: 'gpt-5.2-test', + messages: [], + response: 'ok', + prompt_preview: 'Please ingest https://arxiv.org/abs/2406.15341 into Von.', + workflow_routing_diagnostics: { + schema_version: 'workflow_routing_diagnostics.v1', + selected_workflow_id: '#V#tool_calling_workflow', + selector_verdict: 'selected', + selector_source: 'selector', + selector: { + prompt_id: '#V#chat_turn_classifier_prompt', + candidate_list: { + text: '- #V#tool_calling_workflow: Tool Calling Workflow', + char_count: 46 + }, + response: { + text: '{"workflow_id":"#V#tool_calling_workflow"}', + char_count: 43 + } + } + }, + turn_execution_diagnostics: { + generated_at_utc: '2026-07-01T16:08:05.572Z', + request_id: 'req-top-level-diagnostics', + history_location: { + session_id: 'session-top-level-diagnostics', + history_index: 8 + }, + workflow_discovery: { + matches: [ + { + concept_id: '#V#source_neutral_paper_reference_ingestion_workflow' + } + ] + } + } + }); + + await showLlmDebugPopup('assistant-top-level-diagnostics'); + + const popup = document.getElementById('chatLlmDebugPopup'); + const payload = JSON.parse(popup.dataset.currentDebugData || '{}'); + + expect(payload.prompt_preview).toBe( + 'Please ingest https://arxiv.org/abs/2406.15341 into Von.' + ); + expect(payload.workflow_routing_diagnostics).toEqual(expect.objectContaining({ + selected_workflow_id: '#V#tool_calling_workflow', + selector_verdict: 'selected', + selector_source: 'selector' + })); + expect(payload.workflow_routing_diagnostics.selector).toEqual( + expect.objectContaining({ + prompt_id: '#V#chat_turn_classifier_prompt', + candidate_list: expect.objectContaining({ + preview: '- #V#tool_calling_workflow: Tool Calling Workflow' + }), + response: expect.objectContaining({ + preview: '{"workflow_id":"#V#tool_calling_workflow"}' + }) + }) + ); + }); + test('copy payload keeps auxiliary traces separate from the selected-workflow trace', async () => { const { setLlmDebugDataForTurn, showLlmDebugPopup } = require(chatTabModulePath); diff --git a/tests/frontend/footerLlmExecutionStatus.test.js b/tests/frontend/footerLlmExecutionStatus.test.js index f4c2f550e..a7812091c 100644 --- a/tests/frontend/footerLlmExecutionStatus.test.js +++ b/tests/frontend/footerLlmExecutionStatus.test.js @@ -317,4 +317,43 @@ describe('footer latest LLM execution status', () => { expect(footerContainer.getAttribute('data-keep-title')).toBe('true'); expect(footerContainer.title).toContain('Footer partially ready.'); }); + + test('shows fatal workflow index footer badge when capability index is unavailable', async () => { + installFetchMock({ + settingsPayload: { + resolved_llm: { provider: 'openai', model: 'gpt-5.4-mini' }, + workflow_capability_index: { + ready: false, + workflow_discovery_available: false, + user_visible_severity: 'error', + status: 'building', + summary: 'Workflow capability index still building.', + detail: 'Workflow discovery is waiting on the authoritative capability index to finish building.', + size: 0, + query_surface_ready: false, + namespace_state: { + status: 'missing_index', + detail: 'No persisted index exists for this namespace yet.' + } + } + } + }); + const { setModelInfoFooterText } = require(domUtilsPath); + + await setModelInfoFooterText(); + await flushUiTicks(); + + const badge = document.querySelector('.workflow-index-status-badge'); + const footerContainer = document.querySelector('.footer-container'); + const button = badge?.querySelector('.concept-footer-button'); + expect(badge).toBeTruthy(); + expect(badge.classList.contains('fatal')).toBe(true); + expect(badge.textContent).toContain('Workflow Index'); + expect(badge.textContent).toContain('building'); + expect(button?.getAttribute('aria-label')).toBe('Open workflow capability index status'); + expect(button?.title).toContain('Represented workflow discovery is not fully available.'); + expect(button?.title).toContain('Namespace status: missing_index'); + expect(footerContainer.classList.contains('footer-not-ready')).toBe(true); + expect(footerContainer.title).toContain('workflow capability index'); + }); }); diff --git a/tests/frontend/settingsRuntimeStatusPolling.test.js b/tests/frontend/settingsRuntimeStatusPolling.test.js index 2ebda14e1..ad71dc4e7 100644 --- a/tests/frontend/settingsRuntimeStatusPolling.test.js +++ b/tests/frontend/settingsRuntimeStatusPolling.test.js @@ -6,6 +6,7 @@ jest.mock('../../src/frontend/web/von_interface/static/js/apiService.js', () => })); const { + __testOnly_applyCapabilityIndexStatusCard, __testOnly_refreshRuntimeModelStatus, __testOnly_resetRuntimeModelStatusCache } = require('../../src/frontend/web/von_interface/static/js/settingsPage.js'); @@ -102,4 +103,27 @@ describe('settings runtime model status polling', () => { '/api/workflows/capability-index/status?nocache=1' ]); }); + + test('renders unavailable workflow capability index as user-visible error', () => { + __testOnly_applyCapabilityIndexStatusCard({ + ready: false, + status: 'building', + warning_level: 'warning', + user_visible_severity: 'error', + summary: 'Workflow capability index still building.', + detail: 'Workflow discovery is waiting on the authoritative capability index.', + namespace_state: { + detail: 'No persisted index exists for this namespace yet.' + } + }); + + const card = document.getElementById('workflowCapabilityIndexStatusCard'); + expect(card.dataset.status).toBe('building'); + expect(card.dataset.warningLevel).toBe('error'); + expect(card.dataset.userVisibleSeverity).toBe('error'); + expect(document.getElementById('workflowCapabilityIndexStatusSummary').textContent) + .toContain('still building'); + expect(document.getElementById('workflowCapabilityIndexStatusDetail').textContent) + .toContain('No persisted index exists'); + }); }); diff --git a/tests/test_run_authenticated_browser_workflow_replay.py b/tests/test_run_authenticated_browser_workflow_replay.py index 029b8f3de..3afdead81 100644 --- a/tests/test_run_authenticated_browser_workflow_replay.py +++ b/tests/test_run_authenticated_browser_workflow_replay.py @@ -3,6 +3,82 @@ import scripts.run_authenticated_browser_workflow_replay as replay +def test_establish_browser_test_session_skips_fixture_refresh(monkeypatch) -> None: + captured: dict[str, object] = {} + + class _FakeSession: + def __init__(self) -> None: + self.headers: dict[str, str] = {} + + def _fake_request_json(*args, **kwargs): + captured["args"] = args + captured["kwargs"] = kwargs + return {"success": True, "user_concept_id": "#V#zhan_von_witbrock"} + + monkeypatch.setattr(replay, "_request_json", _fake_request_json) + session = _FakeSession() + + result = replay.establish_browser_test_session( + session=session, # type: ignore[arg-type] + base_url="http://127.0.0.1:5001", + window_session_id="window-1", + timeout_seconds=12.0, + ) + + assert result["success"] is True + assert session.headers["X-Von-Window-Session"] == "window-1" + assert captured["kwargs"]["json"] == { + "window_session_id": "window-1", + "refresh_fixture": False, + } + + +def test_workflow_capability_preflight_waits_through_transient_build( + monkeypatch, +) -> None: + payloads = [ + { + "ready": False, + "workflow_discovery_available": False, + "status": "building", + "build_in_progress": True, + "size": 0, + "detail": "Workflow capability index still building.", + }, + { + "ready": True, + "workflow_discovery_available": True, + "status": "ready", + "build_in_progress": False, + "size": 83, + "detail": "Workflow capability index ready.", + }, + ] + sleeps: list[float] = [] + + def _fake_request_json(*_args, **_kwargs): + return payloads.pop(0) + + monkeypatch.setattr(replay, "_request_json", _fake_request_json) + monkeypatch.setattr(replay.time, "sleep", lambda seconds: sleeps.append(seconds)) + + preflight = replay.build_workflow_capability_preflight( + session=object(), # type: ignore[arg-type] + base_url="http://127.0.0.1:5001", + timeout_seconds=30.0, + poll_interval_seconds=0.5, + ) + + assert preflight["workflow_discovery_available"] is True + assert preflight["preflight_wait"]["completed"] is True + assert preflight["preflight_wait"]["attempt_count"] == 2 + assert [item["status"] for item in preflight["preflight_wait"]["snapshots"]] == [ + "building", + "ready", + ] + assert sleeps == [0.5] + + def test_extract_progress_facts_from_nested_progress_payloads() -> None: payload = { "progress_snapshots": [ @@ -58,6 +134,9 @@ def test_build_report_uses_task_status_snapshots_for_route_and_progress_evidence window_session_id="window-1", auth_login={"success": True}, auth_status={"authenticated": True}, + target_session_context={"target_session_context_ready": True}, + database_preflight={"database_runtime_available": True}, + llm_preflight={"llm_runtime_available": True}, gmail_preflight={"gmail_capability_ready": True}, chat_session={}, submission={"task_id": "task-1"}, @@ -123,6 +202,126 @@ def test_classify_replay_reports_auth_blocker_before_workflow_assertions() -> No assert "disabled" in analysis["blocker"]["reason"] +def test_classify_replay_keeps_secondary_preflight_blockers_visible() -> None: + analysis = replay.classify_replay( + case=replay.GMAIL_ARXIV_REPLAY_CASE, + auth_login={"success": False, "error": "Browser-test auth is disabled"}, + auth_status={"authenticated": False}, + gmail_preflight={"gmail_capability_ready": True}, + task_evidence={"last_task_status": {"status": "unknown"}}, + selected_workflow_ids=[], + observed_workflow_ids=[], + selector_diagnostics=[], + progress_facts=[], + workflow_capability_preflight={ + "ready": False, + "workflow_discovery_available": False, + "status": "error", + "summary": "Workflow capability index requires rebuild.", + }, + ) + + assert analysis["verdict"] == "blocked" + assert analysis["blocker"]["type"] == "browser_test_auth_blocker" + assert [item["type"] for item in analysis["precondition_blockers"]] == [ + "browser_test_auth_blocker", + "workflow_capability_index_blocker", + ] + workflow_blocker = analysis["precondition_blockers"][1] + assert workflow_blocker["workflow_capability_preflight"]["status"] == "error" + + +def test_classify_replay_reports_database_blocker_before_selector() -> None: + analysis = replay.classify_replay( + case=replay.GMAIL_ARXIV_REPLAY_CASE, + auth_login={"success": True}, + auth_status={"authenticated": True}, + database_preflight={ + "database_runtime_available": False, + "connected": False, + "error": "Mongo read/write health was not stable.", + }, + gmail_preflight={"gmail_capability_ready": True}, + task_evidence={"last_task_status": {"status": "completed"}}, + selected_workflow_ids=[replay.GMAIL_ARXIV_WORKFLOW_ID], + observed_workflow_ids=[replay.GMAIL_ARXIV_WORKFLOW_ID], + selector_diagnostics=[], + progress_facts=[], + ) + + assert analysis["verdict"] == "blocked" + assert analysis["blocker"]["type"] == "database_runtime_blocker" + assert analysis["precondition_blockers"][0]["type"] == "database_runtime_blocker" + + +def test_classify_replay_reports_llm_model_blocker_before_selector() -> None: + analysis = replay.classify_replay( + case=replay.GMAIL_ARXIV_REPLAY_CASE, + auth_login={"success": True}, + auth_status={"authenticated": True}, + llm_preflight={ + "llm_runtime_available": False, + "provider": "ollama", + "model_checked": "gemma4:e4b", + "selected_model_available": False, + "error": "The effective replay model is not ready.", + }, + gmail_preflight={"gmail_capability_ready": True}, + task_evidence={"last_task_status": {"status": "completed"}}, + selected_workflow_ids=[replay.GMAIL_ARXIV_WORKFLOW_ID], + observed_workflow_ids=[replay.GMAIL_ARXIV_WORKFLOW_ID], + selector_diagnostics=[], + progress_facts=[], + ) + + assert analysis["verdict"] == "blocked" + assert analysis["blocker"]["type"] == "llm_runtime_blocker" + assert analysis["blocker"]["llm_preflight"]["model_checked"] == "gemma4:e4b" + + +def test_apply_target_session_context_reports_mismatch(monkeypatch) -> None: + calls: list[tuple[str, str, dict]] = [] + + class _FakeSession: + def __init__(self) -> None: + self.headers: dict[str, str] = {} + + def _fake_request_json(_session, method, url, **kwargs): + calls.append((method, url, dict(kwargs))) + if url.endswith("/set_user_concept"): + return {"status": "updated", "user_id": "#V#michael_witbrock"} + if url.endswith("/set_organisation"): + return { + "status": "updated", + "organisation_id": "#V#university_of_auckland_strong_ai_lab", + } + if url.endswith("/session/context"): + return { + "authenticated": True, + "user_id": "#V#zhan_von_witbrock", + "organisation_id": "#V#university_of_auckland_strong_ai_lab", + "namespace": "#V#zhan_von_witbrock@university_of_auckland_strong_ai_lab", + } + raise AssertionError(url) + + monkeypatch.setattr(replay, "_request_json", _fake_request_json) + session = _FakeSession() + + result = replay.apply_target_session_context( + session=session, # type: ignore[arg-type] + base_url="http://127.0.0.1:5001", + user_concept_id="#V#michael_witbrock", + organisation_concept_id="#V#university_of_auckland_strong_ai_lab", + ) + + assert result["target_session_context_ready"] is False + assert result["mismatch_reasons"] == [ + "requested user #V#michael_witbrock but effective user is #V#zhan_von_witbrock" + ] + assert session.headers["X-User-Concept-ID"] == "#V#michael_witbrock" + assert [call[0] for call in calls] == ["POST", "POST", "GET"] + + def test_classify_replay_reports_gmail_profile_blocker_when_auth_ready() -> None: analysis = replay.classify_replay( case=replay.GMAIL_ARXIV_REPLAY_CASE, @@ -144,6 +343,37 @@ def test_classify_replay_reports_gmail_profile_blocker_when_auth_ready() -> None assert analysis["blocker"]["type"] == "gmail_oauth_or_profile_blocker" +def test_classify_replay_reports_workflow_capability_index_blocker_before_selector() -> None: + analysis = replay.classify_replay( + case=replay.GMAIL_ARXIV_REPLAY_CASE, + auth_login={"success": True}, + auth_status={"authenticated": True}, + gmail_preflight={"gmail_capability_ready": True}, + task_evidence={"last_task_status": {"status": "completed"}}, + selected_workflow_ids=["#V#tool_calling_workflow"], + observed_workflow_ids=[], + selector_diagnostics=[ + {"selected_workflow_id": "#V#tool_calling_workflow"} + ], + progress_facts=[], + workflow_capability_preflight={ + "ready": False, + "workflow_discovery_available": False, + "status": "building", + "summary": "Workflow capability index still building.", + "detail": "Workflow discovery is waiting on the authoritative capability index.", + }, + ) + + assert analysis["verdict"] == "blocked" + assert analysis["blocker"]["type"] == "workflow_capability_index_blocker" + assert analysis["blocker"]["workflow_capability_preflight"]["status"] == "building" + assert analysis["workflow_capability_preflight"]["workflow_discovery_available"] is False + assert analysis["precondition_blockers"][0]["type"] == ( + "workflow_capability_index_blocker" + ) + + def test_classify_replay_passes_when_expected_workflow_and_projection_seen() -> None: facts = [ {"fact_id": fact_id, "contract_id": contract_id} @@ -201,6 +431,42 @@ def test_classify_replay_reports_selector_blocker_before_running_timeout() -> No ] +def test_classify_replay_reports_context_build_blocker_before_selector() -> None: + analysis = replay.classify_replay( + case=replay.GMAIL_ARXIV_REPLAY_CASE, + auth_login={"success": True}, + auth_status={"authenticated": True}, + gmail_preflight={"gmail_capability_ready": True}, + task_evidence={ + "last_task_status": { + "status": "cancelled", + "progress": { + "stage": "context_build", + "phase": "context_build", + "subtask": "request setup", + }, + } + }, + selected_workflow_ids=[], + observed_workflow_ids=[], + selector_diagnostics=[ + {"status": "running"}, + {"phase": "context_build", "status": "thinking"}, + {"status": "cancelled"}, + ], + progress_facts=[], + workflow_capability_preflight={ + "ready": True, + "workflow_discovery_available": True, + "status": "ready", + }, + ) + + assert analysis["verdict"] == "blocked" + assert analysis["blocker"]["type"] == "context_build_blocker" + assert "no selector candidate set" in analysis["blocker"]["reason"] + + def test_classify_replay_reports_expected_workflow_action_blocker() -> None: analysis = replay.classify_replay( case=replay.GMAIL_ARXIV_REPLAY_CASE,