fix: round-5 bug-hunt sweep (cost gates, MCP stability, frontend a11y)

LLM cost-safety gates added to seven previously uncovered call sites
that could otherwise drive unbounded spend on long inputs:
  experts/citation_validator, gap_discovery, conflict_resolver, curriculum,
  map_reduce (map + reduce), multi_pass (extract/cross-ref/synthesise),
  synthesis (synthesize + extract), task_planner (decompose + synth),
  embedding_cache (per-doc + query).

MCP / skills:
  - executor MCPClientProxy now drains stderr so subprocess buffer
    can't fill and deadlock on chatty servers.
  - skills/definition trigger regex protected from ReDoS by capping
    pattern length and rejecting nested-quantifier backtracking risks.
  - mcp/client/pool emits a terminal ProgressEvent on tool completion
    (was stored but never triggered).

Storage:
  - findings_store guards in-memory index mutations with threading.RLock.

CLI:
  - research.py cancel merges nested asyncio.run() into a single loop
    so lookup + cancel share the same event loop.

Frontend:
  - a11y: htmlFor/id pairing for settings + research-studio inputs,
    aria-label on search/slider widgets, aria-valuemin/max/now on cost slider.
  - responsive: trace-explorer decision sidebar hidden on <lg, expert-profile
    conversation sidebar hidden on <md.
  - dead code removal: use-local-storage, use-media-query, activity-feed,
    memory-indicator.

Author: blisspixel

deepr/cli/commands/research.py

@@ -515,25 +515,7 @@ async def cancel_job():
                 return job, "Job already failed"
             return job, None
 
-        job, error = asyncio.run(cancel_job())
-        if error:
-            print_error(error)
-            raise click.Abort()
-
-        console.print("\nJob Details:")
-        console.print(f"   ID: {job.id}")
-        console.print(f"   Status: {job.status.name}")
-        console.print(f"   Prompt: {job.prompt[:80]}{'...' if len(job.prompt) > 80 else ''}")
-        if job.provider_job_id:
-            console.print(f"   Provider Job ID: {job.provider_job_id}")
-
-        if not yes and not click.confirm("\nCancel this job?"):
-            print_warning("Cancelled")
-            return
-
-        print_success("Cancelling job...")
-
-        async def do_cancel():
+        async def do_cancel(job):
             if job.provider_job_id:
                 try:
                     await provider.cancel_job(job.provider_job_id)
@@ -542,10 +524,38 @@ async def do_cancel():
                     console.print(f"   Warning: Could not cancel at provider: {e}")
             await queue.update_status(job_id=job.id, status=JobStatus.FAILED, error="Cancelled by user")
 
-        import asyncio as _asyncio
+        # Single event loop spans the lookup AND the cancel so the
+        # sqlite queue / aiohttp provider connections opened during
+        # lookup stay valid for the cancel call. The previous
+        # ``asyncio.run`` per step closed the loop between them,
+        # binding any reused stateful objects to a now-closed loop
+        # and intermittently raising "got Future attached to a
+        # different loop".
+        loop = asyncio.new_event_loop()
+        try:
+            asyncio.set_event_loop(loop)
+            job, error = loop.run_until_complete(cancel_job())
+            if error:
+                print_error(error)
+                raise click.Abort()
+
+            console.print("\nJob Details:")
+            console.print(f"   ID: {job.id}")
+            console.print(f"   Status: {job.status.name}")
+            console.print(f"   Prompt: {job.prompt[:80]}{'...' if len(job.prompt) > 80 else ''}")
+            if job.provider_job_id:
+                console.print(f"   Provider Job ID: {job.provider_job_id}")
+
+            if not yes and not click.confirm("\nCancel this job?"):
+                print_warning("Cancelled")
+                return
 
-        _asyncio.run(do_cancel())
-        print_success("Job cancelled successfully!")
+            print_success("Cancelling job...")
+            loop.run_until_complete(do_cancel(job))
+            print_success("Job cancelled successfully!")
+        finally:
+            loop.close()
+            asyncio.set_event_loop(None)
 
     except Exception as e:
         print_error(f"Error: {e}")

deepr/experts/citation_validator.py

@@ -102,6 +102,35 @@ async def _validate_batch(
         prompt_parts.append("Output ONLY the JSON array, no other text.")
         prompt = "\n".join(prompt_parts)
 
+        # Pre-flight cost-safety gate. Citation validation can fire many
+        # paid LLM calls in a long-running expert workflow; without this
+        # gate the daily / monthly budget is bypassed.
+        try:
+            from deepr.experts.cost_safety import get_cost_safety_manager
+
+            cost_safety = get_cost_safety_manager()
+            est_cost = 0.02  # gpt-5.2 batch validation; low reasoning effort
+            allowed, deny_reason, _ = cost_safety.check_operation(
+                session_id="citation_validator",
+                operation_type="citation_validation",
+                estimated_cost=est_cost,
+                require_confirmation=False,
+            )
+            if not allowed:
+                logger.warning("Citation validation blocked by cost-safety: %s", deny_reason)
+                return [
+                    SourceValidation(
+                        source_id=src.id,
+                        claim_id=claim.id,
+                        support_class=SupportClass.UNCERTAIN,
+                        explanation=f"Skipped: {deny_reason}",
+                    )
+                    for claim, src, _ in pairs
+                ]
+        except Exception:  # never let cost-safety bookkeeping mask the result
+            cost_safety = None  # type: ignore[assignment]
+            est_cost = 0.0
+
         try:
             client = await self._get_client()
             response = await client.chat.completions.create(
@@ -142,6 +171,26 @@ async def _validate_batch(
                             explanation=result.get("explanation", ""),
                         )
                     )
+
+            # Settle the cost into the canonical ledger.
+            if cost_safety is not None:
+                try:
+                    actual_cost = est_cost
+                    if response.usage:
+                        from deepr.experts.chat import _chat_token_cost as _tc
+
+                        actual_cost = _tc(response.usage, self.model)
+                    cost_safety.record_cost(
+                        session_id="citation_validator",
+                        operation_type="citation_validation",
+                        actual_cost=float(actual_cost),
+                        provider="openai",
+                        model=self.model,
+                        source="experts.citation_validator.validate_claims",
+                    )
+                except Exception:
+                    pass
+
             return validations
 
         except Exception as e:

deepr/experts/conflict_resolver.py

@@ -141,6 +141,25 @@ async def _llm_detect_contradictions(self, pairs: list[tuple[Belief, Belief]]) -
 
         prompt_parts.append("\nOutput ONLY a JSON array like [0, 3, 5] or [] if none contradict.")
 
+        # Cost-safety gate on the LLM-driven contradiction scan.
+        try:
+            from deepr.experts.cost_safety import get_cost_safety_manager
+
+            cost_safety = get_cost_safety_manager()
+            est_cost = 0.02
+            allowed, deny_reason, _ = cost_safety.check_operation(
+                session_id="conflict_resolver",
+                operation_type="conflict_detection",
+                estimated_cost=est_cost,
+                require_confirmation=False,
+            )
+            if not allowed:
+                logger.warning("Conflict detection blocked by cost-safety: %s", deny_reason)
+                return []
+        except Exception:
+            cost_safety = None  # type: ignore[assignment]
+            est_cost = 0.0
+
         try:
             client = await self._get_client()
             response = await client.chat.completions.create(
@@ -155,6 +174,21 @@ async def _llm_detect_contradictions(self, pairs: list[tuple[Belief, Belief]]) -
             if text.startswith("```"):
                 text = text.split("\n", 1)[-1].rsplit("```", 1)[0]
             indices = json.loads(text)
+            if cost_safety is not None:
+                try:
+                    from deepr.experts.chat import _chat_token_cost as _tc
+
+                    actual_cost = _tc(response.usage, "gpt-5.2") if response.usage else est_cost
+                    cost_safety.record_cost(
+                        session_id="conflict_resolver",
+                        operation_type="conflict_detection",
+                        actual_cost=float(actual_cost),
+                        provider="openai",
+                        model="gpt-5.2",
+                        source="experts.conflict_resolver.detect_pairs",
+                    )
+                except Exception:
+                    pass
             return [pairs[i] for i in indices if 0 <= i < len(pairs)]
         except Exception as e:
             logger.warning("LLM contradiction detection failed: %s", e)

deepr/experts/curriculum.py

@@ -440,6 +440,27 @@ async def _call_gpt5_with_retry(
                     # Use the cheapest curriculum model (sorted by cost)
                     model_name = curriculum_models[0].model if curriculum_models else "gpt-5.2"
 
+                    # Cost-safety gate before the curriculum LLM call.
+                    try:
+                        from deepr.experts.cost_safety import get_cost_safety_manager
+
+                        _cost_safety = get_cost_safety_manager()
+                        _est_cost = 0.05  # Curriculum prompts are larger; bumpier estimate
+                        _allowed, _deny_reason, _ = _cost_safety.check_operation(
+                            session_id="curriculum",
+                            operation_type="curriculum_plan",
+                            estimated_cost=_est_cost,
+                            require_confirmation=False,
+                        )
+                        if not _allowed:
+                            logger.warning("Curriculum planning blocked by cost-safety: %s", _deny_reason)
+                            raise RuntimeError(f"Curriculum blocked: {_deny_reason}")
+                    except RuntimeError:
+                        raise
+                    except Exception:
+                        _cost_safety = None  # type: ignore[assignment]
+                        _est_cost = 0.0
+
                     response_obj = await client.chat.completions.create(
                         model=model_name,
                         messages=[
@@ -456,6 +477,23 @@ async def _call_gpt5_with_retry(
                     # Extract response from chat completion
                     response = response_obj.choices[0].message.content or ""
 
+                    # Settle the cost into the canonical ledger.
+                    if _cost_safety is not None:
+                        try:
+                            from deepr.experts.chat import _chat_token_cost as _tc
+
+                            _actual_cost = _tc(response_obj.usage, model_name) if response_obj.usage else _est_cost
+                            _cost_safety.record_cost(
+                                session_id="curriculum",
+                                operation_type="curriculum_plan",
+                                actual_cost=float(_actual_cost),
+                                provider="openai",
+                                model=model_name,
+                                source="experts.curriculum.generate_curriculum",
+                            )
+                        except Exception:
+                            pass
+
                     if progress:
                         progress.complete("Done")
 

deepr/experts/embedding_cache.py

@@ -135,6 +135,16 @@ async def add_documents(
         if not uncached:
             return 0
 
+        # Cost-safety gate. Embedding a corpus is cheap-per-doc but can
+        # add up over thousands of files; the rate limit also exposes a
+        # real-money path through misconfigured loops. Gate per-document.
+        try:
+            from deepr.experts.cost_safety import get_cost_safety_manager
+
+            _cost_safety = get_cost_safety_manager()
+        except Exception:
+            _cost_safety = None  # type: ignore[assignment]
+
         # Batch embed new documents (more efficient than one-by-one)
         new_embeddings = []
         new_metadata = []
@@ -146,9 +156,38 @@ async def add_documents(
             # Truncate content for embedding (model limit)
             embed_content = content[:8000]
 
+            if _cost_safety is not None:
+                try:
+                    _est = 0.0002  # text-embedding-3-small ~$0.02/M tokens
+                    _allowed, _reason, _ = _cost_safety.check_operation(
+                        session_id=f"embed:{self.expert_name}",
+                        operation_type="embed_document",
+                        estimated_cost=_est,
+                        require_confirmation=False,
+                    )
+                    if not _allowed:
+                        logger.warning("Embedding for %s blocked by cost-safety: %s", filename, _reason)
+                        continue
+                except Exception:
+                    _est = 0.0
+            else:
+                _est = 0.0
+
             try:
                 response = await client.embeddings.create(model=model, input=embed_content)
                 embedding = np.array(response.data[0].embedding)
+                if _cost_safety is not None:
+                    try:
+                        _cost_safety.record_cost(
+                            session_id=f"embed:{self.expert_name}",
+                            operation_type="embed_document",
+                            actual_cost=float(_est),
+                            provider="openai",
+                            model=model,
+                            source="experts.embedding_cache.add_documents",
+                        )
+                    except Exception:
+                        pass
 
                 content_hash = self._content_hash(content)
 
@@ -205,10 +244,41 @@ async def search(self, query: str, client, top_k: int = 5, model: str = "text-em
         if self.embeddings is None or len(self.embeddings) == 0:
             return []
 
+        # Cost-safety gate for query embedding.
+        try:
+            from deepr.experts.cost_safety import get_cost_safety_manager
+
+            _cost_safety = get_cost_safety_manager()
+            _est = 0.0001
+            _allowed, _reason, _ = _cost_safety.check_operation(
+                session_id=f"embed_query:{self.expert_name}",
+                operation_type="embed_query",
+                estimated_cost=_est,
+                require_confirmation=False,
+            )
+            if not _allowed:
+                logger.warning("Query embedding blocked by cost-safety: %s", _reason)
+                return []
+        except Exception:
+            _cost_safety = None  # type: ignore[assignment]
+            _est = 0.0
+
         # Embed query (single API call)
         try:
             response = await client.embeddings.create(model=model, input=query)
             query_embedding = np.array(response.data[0].embedding)
+            if _cost_safety is not None:
+                try:
+                    _cost_safety.record_cost(
+                        session_id=f"embed_query:{self.expert_name}",
+                        operation_type="embed_query",
+                        actual_cost=float(_est),
+                        provider="openai",
+                        model=model,
+                        source="experts.embedding_cache.search",
+                    )
+                except Exception:
+                    pass
         except Exception as e:
             logger.error("Error embedding query: %s", e)
             return []

deepr/experts/gap_discovery.py

@@ -204,6 +204,26 @@ async def _generate_gaps_for_thin_areas(
             "Priority 1-5 (5=most important). Output ONLY the JSON."
         )
 
+        # Pre-flight cost-safety gate so the gap-discovery pipeline can't
+        # silently fan out paid LLM calls without daily-budget enforcement.
+        try:
+            from deepr.experts.cost_safety import get_cost_safety_manager
+
+            cost_safety = get_cost_safety_manager()
+            est_cost = 0.02
+            allowed, deny_reason, _ = cost_safety.check_operation(
+                session_id="gap_discovery",
+                operation_type="gap_discovery_thin_areas",
+                estimated_cost=est_cost,
+                require_confirmation=False,
+            )
+            if not allowed:
+                logger.warning("Gap discovery (thin areas) blocked by cost-safety: %s", deny_reason)
+                return []
+        except Exception:
+            cost_safety = None  # type: ignore[assignment]
+            est_cost = 0.0
+
         try:
             client = await self._get_client()
             response = await client.chat.completions.create(
@@ -217,7 +237,23 @@ async def _generate_gaps_for_thin_areas(
             text = response.choices[0].message.content or "[]"
             if text.startswith("```"):
                 text = text.split("\n", 1)[-1].rsplit("```", 1)[0]
-            return json.loads(text)
+            result = json.loads(text)
+            if cost_safety is not None:
+                try:
+                    from deepr.experts.chat import _chat_token_cost as _tc
+
+                    actual_cost = _tc(response.usage, "gpt-5.2") if response.usage else est_cost
+                    cost_safety.record_cost(
+                        session_id="gap_discovery",
+                        operation_type="gap_discovery_thin_areas",
+                        actual_cost=float(actual_cost),
+                        provider="openai",
+                        model="gpt-5.2",
+                        source="experts.gap_discovery._generate_gaps_for_thin_areas",
+                    )
+                except Exception:
+                    pass
+            return result
         except Exception as e:
             logger.warning("Gap generation failed: %s", e)
             return []