From 2c5eda35d9157ea6bafff4dbf5103ca08b50d122 Mon Sep 17 00:00:00 2001 From: jay Date: Fri, 26 Jun 2026 14:33:29 -0700 Subject: [PATCH 01/11] enhancements --- README.md | 32 +- backend/cli.py | 104 +- backend/long_term_memory.py | 29 +- backend/orchestrator.py | 251 +- backend/search_result_store.py | 19 +- backend/smoke_test.py | 93 + backend/unit_tests.py | 84 + docs/API_SERVER.md | 281 +- docs/BENCHMARKING.md | 8 +- docs/MCP_SERVERS.md | 17 +- docs/ORCHESTRATOR.md | 2 +- docs/RAG_NOTES.md | 12 +- src/App.tsx | 79 +- src/api/._client.ts | Bin 0 -> 4096 bytes src/api/client.ts | 21 +- src/components/ChatContainer.tsx | 29 +- src/components/ChatInput.tsx | 1 + src/components/ChatMessage.tsx | 20 +- src/components/DocsViewer.tsx | 31 +- src/components/FileUploader.tsx | 19 +- src/components/MCPServerManager.tsx | 642 + src/components/ModelSettings.tsx | 13 +- src/components/ProgressSpinner.tsx | 11 +- src/components/ReportViewer.tsx | 21 +- src/components/ResearchReportViewer.tsx | 67 +- src/components/Sidebar.tsx | 86 +- src/components/SynthesisMessage.tsx | 14 +- src/contexts/AppContext.tsx | 64 +- src/hooks/useConversations.ts | 20 +- src/hooks/useTheme.ts | 19 +- src/hooks/useWebSocket.ts | 22 +- src/index.css | 22 +- src/package-lock.json | 25215 ++++++---------------- src/package.json | 35 +- src/tailwind.config.js | 10 +- src/tsconfig.json | 6 +- src/vite.config.ts | 2 +- 37 files changed, 8048 insertions(+), 19353 deletions(-) create mode 100644 src/api/._client.ts create mode 100644 src/components/MCPServerManager.tsx diff --git a/README.md b/README.md index 7f170dd..91e6882 100644 --- a/README.md +++ b/README.md @@ -20,7 +20,7 @@ An AI-powered deep research system built on a hierarchical multi-agent architect - In-process RAG: `SearchResultStore` with sentence-transformer embeddings (cosine ranking) scoped per session - In-process long-term memory: `AsyncLongTermMemory` persists cross-session findings to Neo4j (graph database). Two layers: flat vector store (Neo4j vector indexes) for raw evidence + `KnowledgeGraph` (GraphRAG) layer for typed entities, directed triples with confidence scores, and LLM-generated community clusters. Relationships are deduplicated and include temporal tracking (`last_confirmed`, `confirmation_count`). Supports hierarchy (`IS_A`), contradiction (`CONTRADICTS`), and provenance (`SOURCED_FROM → Source`) edges. Graph context is recalled during planning; communities are updated post-synthesis only when enough new facts have landed (`GRAPH_COMMUNITY_MIN_MUTATIONS`); confidence is decayed over time (`CONFIDENCE_DECAY_HALF_LIFE`). A `prune()` method archives stale low-confidence relationships and orphaned entities. - Self-optimization mode: agent reads session logs and the knowledge graph, then updates its own research playbook (`backend/instructions/RESEARCH-METHODS.md`) -- MCP servers: UA rotation, per-domain rate limiting (2s), concurrency cap (3), 429 retry, 50KB output cap +- MCP servers: UA rotation, per-domain rate limiting (2s), concurrency cap (3), 429 retry, 100KB scrape output cap ## Quick Start @@ -44,7 +44,7 @@ make bench # Run benchmark script | `make restart` | Stop + restart backend + MCP servers | | `make restart-all` | Stop + restart everything | | `make init` | Install backend venv and frontend node_modules | -| `make clean` | Remove `.venv` and `node_modules` | +| `make clean` | Remove backend `venv` and frontend `node_modules` / `dist` | **Local development (without Docker):** @@ -68,31 +68,45 @@ Create `backend/.env`: AGENT_MODE=research # research | chat | self-optimization # ── Model backend ───────────────────────────────────────────── -MODEL_BACKEND=ollama # openai | ollama | bedrock | azure | gcp | huggingface +MODEL_BACKEND=ollama # openai | azure | aws | bedrock | gcp | ollama | huggingface | anthropic + +# The research pipeline selects a "heavy" model for planning/synthesis and a +# "light" model for search/analyst work. Each backend has its own pair of +# *_HEAVY_MODEL / *_LIGHT_MODEL vars. A legacy single *_MODEL field also exists +# and is used by the /chat endpoint and CLI chat mode. # Ollama OLLAMA_BASE_URL=http://localhost:11434 -OLLAMA_MODEL=nemotron-3-nano +OLLAMA_HEAVY_MODEL=nemotron-3-nano +OLLAMA_LIGHT_MODEL=nemotron-3-nano # OpenAI (or any OpenAI-compatible endpoint) # OPENAI_API_KEY=sk-... -# OPENAI_MODEL=gpt-4o +# OPENAI_HEAVY_MODEL=gpt-5.2 +# OPENAI_LIGHT_MODEL=gpt-5-nano # OPENAI_BASE_URL=https://api.openai.com/v1 -# AWS Bedrock (uses default boto3 credential chain if AWS_API_KEY is unset) -# AWS_MODEL=global.anthropic.claude-sonnet-4-5-20250929-v1:0 +# AWS Bedrock — two backends: +# MODEL_BACKEND=aws → OpenAI-compatible gateway (needs AWS_BASE_URL + AWS_API_KEY) +# MODEL_BACKEND=bedrock → native boto3 (needs AWS_REGION; default credential chain) +# AWS_HEAVY_MODEL=global.anthropic.claude-sonnet-4-6 +# AWS_LIGHT_MODEL=global.anthropic.claude-haiku-4-5-20251001-v1:0 # AWS_API_KEY=... # AWS_BASE_URL=... +# AWS_REGION=us-east-1 # Azure OpenAI # AZURE_OPENAI_ENDPOINT=https://your-resource.openai.azure.com # AZURE_OPENAI_API_KEY=... # AZURE_OPENAI_DEPLOYMENT=your-deployment-name +# AZURE_HEAVY_MODEL=gpt-5.2 +# AZURE_LIGHT_MODEL=gpt-5-nano # AZURE_API_VERSION=2024-02-15-preview # GCP Vertex AI # GCP_ENDPOINT=https://us-central1-aiplatform.googleapis.com/v1 -# GCP_MODEL=gemini-2.5-pro +# GCP_HEAVY_MODEL=gemini-2.5-pro +# GCP_LIGHT_MODEL=gemini-2.5-flash # GCP_API_KEY=... # HuggingFace (local TGI endpoint) @@ -152,7 +166,7 @@ See [docs/API_SERVER.md](docs/API_SERVER.md) for the full environment variable r Long-term memory is handled in-process by `backend/long_term_memory.py`, backed by Neo4j. It has two layers: - **Flat store** (`:Memory` nodes with vector index): raw evidence and claims, recalled via cosine similarity over Neo4j vector indexes. -- **Knowledge graph**: typed nodes (`:Entity`, `:Community`, `:Source`) and edges (`:RELATES_TO`, `:IS_A`, `:CONTRADICTS`, `:SOURCED_FROM`, `:MEMBER_OF`). Features: relationship deduplication with merge-on-conflict, temporal tracking, hierarchy traversal, contradiction detection, URL-level provenance, cross-session path finding, confidence decay, and graph pruning. Nine REST endpoints under `/graph/` expose the graph for external inspection and maintenance. +- **Knowledge graph**: typed nodes (`:Entity`, `:Community`, `:Source`) and edges (`:RELATES_TO`, `:IS_A`, `:CONTRADICTS`, `:SOURCED_FROM`, `:MEMBER_OF`). Features: relationship deduplication with merge-on-conflict, temporal tracking, hierarchy traversal, contradiction detection, URL-level provenance, cross-session path finding, confidence decay, and graph pruning. Twelve REST endpoints under `/graph/` expose the graph for external inspection and maintenance. There is no separate memory MCP server. diff --git a/backend/cli.py b/backend/cli.py index cba97b3..3447885 100644 --- a/backend/cli.py +++ b/backend/cli.py @@ -1324,7 +1324,10 @@ async def _run_plan(self, query: str) -> None: self._log(f"[bold red]Planning failed: {exc}[/bold red]") finally: self._busy = False - self.query_one("#btn-run-research").disabled = False + # Keep the run button disabled while a plan is awaiting user approval. + # The approve / deny / modify handlers re-enable it when appropriate. + if self._plan is None: + self.query_one("#btn-run-research").disabled = False @work(exclusive=True, thread=False) async def _run_modify_plan(self, feedback: str) -> None: @@ -2051,7 +2054,7 @@ def compose(self) -> ComposeResult: yield Button("✘ Remove Selected", id="btn-mcp-remove") with ScrollableContainer(id="mcp-server-list"): - yield RichLog(id="mcp-log", highlight=True, markup=True) + yield OptionList(id="mcp-log") with ScrollableContainer(id="mcp-tools-panel"): yield Label("🔧 Tools", id="mcp-tools-title") @@ -2112,11 +2115,11 @@ def _refresh_servers(self) -> None: ) sidebar.mount(lbl) - # Update main log - log = self.query_one("#mcp-log", RichLog) - log.clear() + # Update main OptionList + server_list = self.query_one("#mcp-log", OptionList) + server_list.clear_options() if not servers: - log.write(Text("No MCP servers registered.", style="dim")) + server_list.add_option(Option("No MCP servers registered.", id="__empty__")) return for info in servers: name = info["name"] @@ -2126,13 +2129,23 @@ def _refresh_servers(self) -> None: builtin = "builtin" if info.get("builtin") else "user" url = info.get("url", info.get("command", "")) style = "green" if status == "connected" else "red" - log.write( - Text( - f" {name} [{transport}] {url} — {tools_count} tool(s) ({status}, {builtin})", - style=style, + server_list.add_option( + Option( + Text( + f" {name} [{transport}] {url} — {tools_count} tool(s) ({status}, {builtin})", + style=style, + ), + id=name, ) ) + @on(OptionList.OptionSelected, "#mcp-log") + def _server_selected(self, event: OptionList.OptionSelected) -> None: + server_name = event.option_id + if server_name and server_name != "__empty__": + self._selected_server = server_name + self._show_tools_for(server_name) + def _show_tools_for(self, server_name: str) -> None: tools_log = self.query_one("#mcp-tools-log", RichLog) tools_log.clear() @@ -2206,14 +2219,18 @@ async def _register_server(self) -> None: @work(exclusive=True, thread=False) async def _remove_server(self) -> None: feedback = self.query_one("#home-feedback", Label) + if not self._selected_server: + feedback.update("Select a server from the list first.") + return + # Prevent removing built-in servers servers = self.app.mcp_registry.get_all_server_info() - # Remove the last non-builtin server (user should select in the future) - user_servers = [s for s in servers if not s.get("builtin")] - if not user_servers: - feedback.update("No user-added servers to remove.") + info = next((s for s in servers if s["name"] == self._selected_server), None) + if info and info.get("builtin"): + feedback.update(f"'{self._selected_server}' is a built-in server and cannot be removed.") return - target = user_servers[-1]["name"] + target = self._selected_server await self.app.mcp_registry.unregister(target) + self._selected_server = None feedback.update(f"✔ Removed '{target}'") self._refresh_servers() @@ -2261,7 +2278,7 @@ def compose(self) -> ComposeResult: yield Button("▶ Resume Selected", id="btn-session-resume") with ScrollableContainer(id="sessions-list"): - yield RichLog(id="sessions-log", highlight=True, markup=True) + yield OptionList(id="sessions-log") with ScrollableContainer(id="sessions-detail"): yield Label("Session Detail", id="sessions-detail-title") @@ -2277,8 +2294,8 @@ def on_mount(self) -> None: def _load_sessions(self) -> None: """Scan disk checkpoints to list past sessions.""" self._session_ids = [] - log = self.query_one("#sessions-log", RichLog) - log.clear() + session_list = self.query_one("#sessions-log", OptionList) + session_list.clear_options() try: self.query_one("#sidebar-sess-loading").remove() @@ -2296,7 +2313,7 @@ def _load_sessions(self) -> None: child.remove() if not _SESSIONS_DIR.exists(): - log.write(Text("No sessions found.", style="dim")) + session_list.add_option(Option("No sessions found.", id="__empty__")) return # Collect checkpoint files @@ -2307,7 +2324,7 @@ def _load_sessions(self) -> None: ) if not files: - log.write(Text("No sessions found.", style="dim")) + session_list.add_option(Option("No sessions found.", id="__empty__")) return for i, f in enumerate(files[:20]): # Show last 20 @@ -2323,10 +2340,10 @@ def _load_sessions(self) -> None: if ev.get("type") == "plan": plan_data = ev.get("plan", ev.get("data", {}).get("plan", {})) if isinstance(plan_data, dict): - goal = plan_data.get("goal", "")[:50] + goal = plan_data.get("goal", "")[:60] break elif ev.get("type") == "status" and not goal: - goal = ev.get("message", "")[:50] + goal = ev.get("message", "")[:60] except Exception: state = "unknown" @@ -2340,10 +2357,11 @@ def _load_sessions(self) -> None: }.get(state, "") icon = {"complete": "✔", "error": "✘", "executing": "⏳"}.get(state, "○") - display = f" {icon} [{i+1}] {sid[:12]}… {state} {created}" + display = f" {icon} [{i + 1}] {sid[:14]}… {state} {created}" if goal: - display += f"\n {goal}" - log.write(Text(display, style="green" if state == "complete" else "dim")) + display += f" | {goal}" + style = "green" if state == "complete" else "dim" + session_list.add_option(Option(Text(display, style=style), id=str(i))) sidebar.mount( Label( @@ -2353,6 +2371,42 @@ def _load_sessions(self) -> None: ) ) + @on(OptionList.OptionSelected, "#sessions-log") + def _session_selected(self, event: OptionList.OptionSelected) -> None: + opt_id = event.option_id + if opt_id and opt_id != "__empty__": + self._selected_index = int(opt_id) + self._show_session_detail(self._selected_index) + self.query_one("#btn-session-resume").display = True + + def _show_session_detail(self, index: int) -> None: + if index < 0 or index >= len(self._session_ids): + return + sid = self._session_ids[index] + checkpoint_file = _SESSIONS_DIR / f"{sid}.json" + detail_log = self.query_one("#sessions-detail-log", RichLog) + detail_log.clear() + self.query_one("#sessions-detail").display = True + try: + data = json.loads(checkpoint_file.read_text(encoding="utf-8")) + goal = "" + for ev in data.get("replay_log", []): + if ev.get("type") == "plan": + plan_data = ev.get("plan", ev.get("data", {}).get("plan", {})) + if isinstance(plan_data, dict): + goal = plan_data.get("goal", "") + break + state = data.get("state", "unknown") + created = data.get("created_at", "?")[:19] + replay_count = len(data.get("replay_log", [])) + detail_log.write(Text(f"Session ID: {sid}", style="bold")) + detail_log.write(Text(f"State: {state} | Created: {created}", style="dim")) + detail_log.write(Text(f"Events: {replay_count}", style="dim")) + if goal: + detail_log.write(Text(f"Goal: {goal}", style="green")) + except Exception as exc: + detail_log.write(Text(f"Error reading session: {exc}", style="red")) + def on_button_pressed(self, event: Button.Pressed) -> None: bid = event.button.id or "" if bid == "btn-session-new": diff --git a/backend/long_term_memory.py b/backend/long_term_memory.py index 73c3d3b..9d21cdc 100644 --- a/backend/long_term_memory.py +++ b/backend/long_term_memory.py @@ -865,7 +865,7 @@ def _build_union_vector_search( f"YIELD node, score " f"RETURN node, score{extra_return_fields}" ) - subqueries.append(f"{{ {sq} }}") + subqueries.append(sq) union = " UNION ALL ".join(subqueries) outer = ( @@ -927,6 +927,7 @@ async def upsert_entity( ) result = await session.run( f"{union_q} " + "WITH node, score " "WHERE score >= $threshold " "RETURN node, score LIMIT 1", vector=vector, @@ -1047,6 +1048,19 @@ async def upsert_entity( # Relationship CRUD # ------------------------------------------------------------------ + async def _entity_exists(self, name: str) -> bool: + """Return True if an entity node with *name* already exists.""" + if not self.available or not name.strip(): + return False + assert self._driver is not None + all_labels = "|".join(ENTITY_TYPES) + async with self._driver.session(database=self._database) as session: + result = await session.run( + f"MATCH (n:{all_labels} {{name: $name}}) RETURN n.id AS id LIMIT 1", + name=name.strip(), + ) + return (await result.single()) is not None + async def store_relationship( self, source: str, @@ -1085,6 +1099,19 @@ async def store_relationship( evid = evidence.strip() rel_label = relationship_label or classify_relation(rel) + # Guarantee both endpoints exist before the MATCH ... MATCH ... CREATE + # below. If either entity is missing, that pattern binds nothing, the + # CREATE silently no-ops, and the edge is lost — yet the method would + # still report success. Auto-create any missing endpoint as a generic + # concept entity so relationships are never silently dropped. + for endpoint in (src, tgt): + if not await self._entity_exists(endpoint): + await self.upsert_entity( + name=endpoint, + entity_type="concept", + session_id=session_id, + ) + # Build Cypher that matches any typed entity node for source/target. # We can't know the label of source/target at this point, so we use # a label-agnostic match via property. diff --git a/backend/orchestrator.py b/backend/orchestrator.py index 7cdf0d0..6e3c17d 100644 --- a/backend/orchestrator.py +++ b/backend/orchestrator.py @@ -399,7 +399,6 @@ class AnalystAgent(SubAgent): { "claim": "...", "sources": ["url1", "url2"], - "corroborated": true | false, "confidence": "corroborated" | "partially_corroborated" | "single_source" }, ... @@ -810,6 +809,11 @@ def __init__( # Forwarded to subsequent SearchAgent calls so search focus can be # refined based on gaps identified by prior steps' analysts. self._analyst_recommendations: List[Dict[str, Any]] = [] + # Long-term memory recalled once per execution batch and shared across + # that batch's (possibly parallel) steps so every Search/Analyst agent + # can build on prior-session knowledge without paying the recall cost + # per step. Refreshed at the start of each batch in execute(). + self._batch_memory: str = "" # Internal registry of live parallel-worker tasks so they can be # canceled atomically when the outer session task is canceled. self._active_tasks: List[asyncio.Task] = [] @@ -825,12 +829,26 @@ def __init__( # minimal changes. # ------------------------------------------------------------------ - async def plan(self, query: str) -> AsyncIterator[ResponseMessage]: + async def plan( + self, + query: str, + local_documents: Optional[List[Dict[str, str]]] = None, + ) -> AsyncIterator[ResponseMessage]: """ Phase 1: analyse the query and emit a ``ResearchPlan``. Yields ``ResponseMessage`` events; the final event carries ``type="plan"`` with the plan dict in ``plan``. + + Parameters + ---------- + query: + The user's research question. + local_documents: + Optional list of dicts with ``"name"`` and ``"content"`` keys + representing locally discovered documents to treat as primary + source material. When provided, their text is injected into + the planning prompt so the root agent can incorporate them. """ self._reset_state() @@ -871,6 +889,27 @@ async def plan(self, query: str) -> AsyncIterator[ResponseMessage]: + "\n\n" ) + # Inject locally-discovered documents provided by the user so the root + # agent can treat them as primary sources and avoid redundant searches. + if local_documents: + doc_blocks = [] + for doc in local_documents: + name = doc.get("name", "unnamed") + content = doc.get("content", "").strip() + if content: + doc_blocks.append(f"[Document: {name}]\n{content}") + if doc_blocks: + plan_prompt += ( + "\n\nLocal context documents (treat as primary sources; " + "prefer these over re-fetching the same information):\n\n" + + "\n\n".join(doc_blocks) + + "\n\n" + ) + yield ResponseMessage( + type="status", + message=f"Root agent: incorporating {len(doc_blocks)} local document(s)…", + ) + # Inject the current date to aid root agent with its temporal reasoning and source evaluation # (e.g. "if your training cutoff is in 2021 but today's date is 2024, you should prioritize # current sources and be skeptical of outdated info in your training data"). @@ -888,7 +927,7 @@ async def plan(self, query: str) -> AsyncIterator[ResponseMessage]: max_tokens=self.config.root_max_tokens, ) - plan = self._parse_plan(response.content) + plan = self._parse_plan(response.content, query=query) self._pending_plan = plan self.context.save_step("query", query) @@ -955,7 +994,26 @@ async def execute(self) -> AsyncIterator[ResponseMessage]: ) # ── Execute each batch ─────────────────────────────────────────────── + self._batch_memory = "" for batch_idx, batch in enumerate(batches, start=1): + # ── Batch-level long-term memory recall ───────────────────────── + # Recall once per batch (cached across the batch's parallel steps) + # so each step's Search and Analyst agents can build on knowledge + # from prior research sessions. Failures are swallowed inside + # _recall_memories, so this never breaks execution. + batch_recall_query = " ".join(s.description for s in batch).strip() + self._batch_memory = await self._recall_memories( + batch_recall_query or query, limit=5 + ) + if self._batch_memory: + yield ResponseMessage( + type="status", + message=( + f"Batch {batch_idx}: recalled prior-session knowledge from " + "long-term memory to inform step research." + ), + ) + is_parallel = len(batch) > 1 if is_parallel: @@ -1198,6 +1256,7 @@ async def _run_step( else "" ), status_sink=_search_status, + ltm_context=getattr(self, "_batch_memory", "") or None, ) for _msg in _search_status: yield _msg @@ -1209,7 +1268,13 @@ async def _run_step( type="status", message=f"[Analyst] Extracting claims for step {step.id}…", ) - analyst_result = await self._run_analyst(step, query, search_result, tools) + analyst_result = await self._run_analyst( + step, + query, + search_result, + tools, + ltm_context=getattr(self, "_batch_memory", "") or None, + ) self.context.save_step( f"analyst_step_{step.id}", json.dumps(analyst_result) ) @@ -1479,9 +1544,49 @@ async def _run_step( json.dumps(analyst_result), ) + # Mark the actionable contradictions just re-investigated as + # resolved. These are the same Contradiction instances + # recorded in self._contradictions, so the report's + # "unresolved contradictions" caveats and observability now + # accurately reflect that targeted re-search addressed them. + _new_chunk_count = max(chunks_after - chunks_before, 0) + for _c in actionable: + _c.resolved = True + _c.resolution = ( + f"Targeted re-search gathered {_new_chunk_count} new " + "source chunk(s); claims re-extracted and reconciled." + ) + # Persist QA-vetted findings to long-term memory for future sessions await self._store_analyst_findings(step, analyst_result) + # Persist this step's raw evidence incrementally so learnings survive + # a mid-session crash and are available to later steps/sessions — + # rather than only writing everything after synthesis completes. + try: + await self._search_store.persist_to_long_term_memory( + query=step.description, + top_k=10, + step_id_filter=step.id, + ) + except Exception as exc: + logger.debug( + "[Orchestrator] Incremental evidence persist skipped for " + "step %d: %s", + step.id, + exc, + ) + + # Deduplicate the tools invoked across the initial search and any QA + # re-search passes, preserving first-seen order, for both the + # observability event and the step_complete payload below. + _seen_tools: set = set() + tools_used = [ + t + for t in all_tools_used + if not (t in _seen_tools or _seen_tools.add(t)) # type: ignore[func-returns-value] + ] + # Emit a structured event so analyst notes for every step are # captured in per-session log dumps for pipeline diagnostics. record_event( @@ -1492,6 +1597,7 @@ async def _run_step( claim_count=len(analyst_result.get("claims", [])), tension_count=len(analyst_result.get("tensions", [])), qa_retries=qa_retry_count, + tools_used=tools_used, analyst_notes=str(analyst_result.get("analyst_notes") or ""), claims=[ { @@ -1514,32 +1620,15 @@ async def _run_step( ], ) - # ── Supersede contradicted source chunks in the RAG store ──────── - # When the LoopAgent resolves a contradiction it identifies which - # claim (and therefore which source step) is wrong. We flag - # the losing step's chunks as superseded so they don't pollute - # future retrievals for other steps or the synthesis phase. - resolved_contradictions = [ - c for c in self._contradictions if c.resolved and c.resolution - ] - for contradiction in resolved_contradictions: - # Heuristic: if source_a looks like a step reference ("step_N"), - # supersede that step's chunks; otherwise supersede by step_id. - for src_field in (contradiction.source_a, contradiction.source_b): - if src_field and src_field.startswith("step_"): - try: - losing_step_id = int(src_field.split("_")[1]) - if losing_step_id != step.id: - self._search_store.mark_superseded_by_step( - losing_step_id, - reason=f"Contradiction resolved: {contradiction.context}", - ) - self.context.supersede( - losing_step_id, - reason=f"Contradiction resolved: {contradiction.context}", - ) - except (IndexError, ValueError): - pass + # ── Contradiction resolution bookkeeping ───────────────────────── + # Contradictions are detected WITHIN a single step's analyst output, + # so their source_a/source_b fields are source URLs/labels from this + # step — never cross-step references. Resolution is therefore + # reflected via the Contradiction.resolved flag (set in the QA loop + # after a successful targeted re-search), which excludes them from + # the report's unresolved-contradiction caveats. We do not supersede + # whole-step chunks here because that would discard this step's valid + # evidence along with the contradicted claim. # ── Generate per-step summary (Active Context briefing) ────────── # This ≤config.step_summary_max_chars bullet summary is stored in @@ -1586,12 +1675,6 @@ async def _run_step( # ── Final bookkeeping ──────────────────────────────────────────── self._step_sources[step.id] = self._extract_sources(search_result) - # Deduplicate tools list while preserving order - seen: set = set() - tools_used = [ - t for t in all_tools_used if not (t in seen or seen.add(t)) # type: ignore[func-returns-value] - ] - self._raw_findings.append(search_result) self._analyst_output = self._merge_analyst_outputs( self._analyst_output, analyst_result @@ -1967,6 +2050,11 @@ async def synthesize(self) -> AsyncIterator[ResponseMessage]: if structured_claims_block: draft_prompt_parts.append(structured_claims_block) if not section_evidence and not structured_claims_block: + logger.warning( + "[Synthesis] No RAG evidence or structured claims for section " + "'%s' — falling back to raw analyst output.", + sec_title, + ) # Last-resort fallback: pull relevant subset from analyst output analyst_str = json.dumps(self._analyst_output, indent=2) draft_prompt_parts.append(f"Research Findings:\n{analyst_str[:6000]}") @@ -2108,8 +2196,19 @@ async def synthesize(self) -> AsyncIterator[ResponseMessage]: ) # ── Assemble final document ────────────────────────────────────────── + # Separate genuine multi-perspective source disagreements (legitimate + # research findings) from unresolved factual contradictions (gaps or + # errors) so the report frames each appropriately instead of lumping + # valid disagreements under "unresolved contradictions". + source_perspective_disagreements = [ + c.to_dict() + for c in self._contradictions + if not c.resolved and c.contradiction_type == "source_disagreement" + ] unresolved_step_contradictions = [ - c.to_dict() for c in self._contradictions if not c.resolved + c.to_dict() + for c in self._contradictions + if not c.resolved and c.contradiction_type != "source_disagreement" ] doc_parts: List[str] = [ @@ -2143,6 +2242,21 @@ async def synthesize(self) -> AsyncIterator[ResponseMessage]: ) doc_parts.append("") + if source_perspective_disagreements: + doc_parts.append("SOURCE PERSPECTIVES AND DISAGREEMENTS") + doc_parts.append( + "The following reflect genuine differences in how credible sources " + "assess the same question — not factual errors. They are presented " + "so readers can weigh the competing perspectives." + ) + for i, sd in enumerate(source_perspective_disagreements, start=1): + doc_parts.append( + f"{i}. {sd.get('context', '')} " + f"(Perspective A: {sd.get('source_a', '')} — {sd.get('claim_a', '')}; " + f"Perspective B: {sd.get('source_b', '')} — {sd.get('claim_b', '')})" + ) + doc_parts.append("") + if all_sources: doc_parts.append("REFERENCES") for i, src in enumerate(all_sources, start=1): @@ -2344,7 +2458,9 @@ async def modify_plan(self, plan_id: str, feedback: str) -> ResponseMessage: temperature=self.config.root_temperature, max_tokens=self.config.root_max_tokens, ) - modified_plan = self._parse_plan(response.content) + modified_plan = self._parse_plan( + response.content, query=self.context.get_step_result("query") or "" + ) self._pending_plan = modified_plan self.context.save_step("plan", json.dumps(modified_plan.to_dict())) @@ -2707,6 +2823,7 @@ async def _run_search( tools: List[Dict[str, Any]], extra_context: Optional[str] = None, status_sink: Optional[List] = None, + ltm_context: Optional[str] = None, ) -> Dict[str, Any]: """Run the SearchAgent for a single plan step. @@ -2730,6 +2847,13 @@ async def _run_search( "general summaries. Retrieve multiple independent sources per claim " "and explore sub-topics thoroughly before concluding." ) + if ltm_context: + prompt_parts.append( + "Relevant knowledge from prior research sessions (long-term memory). " + "Use it to avoid redundant searching and to build on what is already " + "known — treat it as prior context to verify, not as ground truth:\n" + + ltm_context + ) if extra_context: prompt_parts.append(f"Additional Context:\n{extra_context}") prompt_parts.append( @@ -2895,6 +3019,7 @@ async def _run_analyst( search_output: Dict[str, Any], tools: List[Dict[str, Any]], extra_context: Optional[str] = None, + ltm_context: Optional[str] = None, ) -> Dict[str, Any]: """ Run the AnalystAgent over the search results for a single step. @@ -2978,6 +3103,26 @@ async def _run_analyst( ) search_context = "Search Results:\n" + raw_str + # ── Surface the SearchAgent's self-identified coverage gaps ────────── + # so the analyst can flag under-covered claims instead of silently + # extracting over them. The search phase records these gaps in the + # "coverage_notes" field, which the primary RAG path would otherwise + # drop entirely. + coverage_note = "" + _final_msg = search_output.get("final_message", "") + if _final_msg: + try: + _cov = _final_msg + if "```json" in _cov: + _cov = _cov.split("```json")[1].split("```")[0].strip() + elif "```" in _cov: + _cov = _cov.split("```")[1].split("```")[0].strip() + coverage_note = str( + json.loads(_cov).get("coverage_notes", "") or "" + ).strip() + except Exception: + coverage_note = "" + # ── 2. Cross-step corroboration context ────────────────────────────── # Retrieve top-5 chunks from OTHER completed steps so the analyst can # triangulate claims against independently-gathered evidence. @@ -3032,6 +3177,12 @@ async def _run_analyst( f"Current Step: {step.description}", search_context, ] + if coverage_note: + prompt_parts.append( + "Search-phase coverage notes (gaps the search agent flagged — " + "treat affected claims as lower-confidence and note the missing " + "evidence rather than overstating):\n" + coverage_note + ) if cross_step_context: prompt_parts.append( "Corroborating evidence from prior steps " @@ -3043,6 +3194,13 @@ async def _run_analyst( "(use to corroborate, extend, or identify conflicts with your new claims):\n\n" + prior_claims_block ) + if ltm_context: + prompt_parts.append( + "Relevant knowledge from prior research sessions (long-term memory). " + "Corroborate your new claims against it or extend it — do not blindly " + "trust it; flag conflicts with prior knowledge as tensions:\n" + + ltm_context + ) if extra_context: prompt_parts.append(f"Additional Context:\n{extra_context}") if self.research_depth == "deep": @@ -3137,7 +3295,7 @@ def _merge_analyst_outputs( # ── Plan parsing ───────────────────────────────────────────────────────── @staticmethod - def _parse_plan(raw: str) -> ResearchPlan: + def _parse_plan(raw: str, query: str = "") -> ResearchPlan: """Parse the Root agent's JSON output into a ``ResearchPlan``.""" content = raw if "```json" in content: @@ -3159,15 +3317,22 @@ def _parse_plan(raw: str) -> ResearchPlan: ] return ResearchPlan(goal=data.get("goal", "Research goal"), steps=steps) except Exception as exc: - logger.warning("Could not parse plan JSON (%s) — using fallback.", exc) - # Fallback: single generic step + logger.warning( + "Could not parse plan JSON (%s) — using query-based fallback. " + "Raw response head: %.300s", + exc, + raw, + ) + # Fallback: a single step that still reflects the user's query so a + # parse failure does not silently discard the research topic. + topic = (query or "the research topic").strip() return ResearchPlan( - goal="Research goal", + goal=f"Research: {topic[:200]}", steps=[ ResearchStep( id=1, name="Step 1", - description="Gather information on the research topic.", + description=f"Gather comprehensive information on: {topic[:300]}", status=StepStatus.PENDING, ) ], diff --git a/backend/search_result_store.py b/backend/search_result_store.py index 09d0e34..32b6f2d 100644 --- a/backend/search_result_store.py +++ b/backend/search_result_store.py @@ -385,14 +385,18 @@ async def persist_to_long_term_memory( self, query: str, top_k: int = 20, + step_id_filter: Optional[int] = None, ) -> int: """ Persist the top-k semantically-ranked chunks from this session into the in-process ``AsyncLongTermMemory`` store. - Called at the end of ``Orchestrator.synthesize()`` so that raw evidence - that did not rise to the level of a named "claim" is still retained for - future sessions. + Called incrementally after each research step (scoped via + ``step_id_filter``) and once more at the end of + ``Orchestrator.synthesize()`` so that raw evidence that did not rise to + the level of a named "claim" is still retained for future sessions. + Long-term memory deduplicates near-identical content on write, so the + per-step and end-of-session passes do not create duplicates. Parameters ---------- @@ -400,6 +404,8 @@ async def persist_to_long_term_memory( The original research query — used to rank chunks before persisting. top_k: Maximum number of chunks to write to long-term memory. + step_id_filter: + If set, only persist chunks belonging to this step id. Returns ------- @@ -412,7 +418,12 @@ async def persist_to_long_term_memory( ) return 0 - active_chunks = [c for c in self._chunks if not c.superseded] + active_chunks = [ + c + for c in self._chunks + if not c.superseded + and (step_id_filter is None or c.step_id == step_id_filter) + ] if not active_chunks: return 0 diff --git a/backend/smoke_test.py b/backend/smoke_test.py index 020f304..5d0edb7 100644 --- a/backend/smoke_test.py +++ b/backend/smoke_test.py @@ -448,6 +448,94 @@ def test_run_analyst_cross_step_retrieval(): print("✅ _run_analyst uses exclude_step_id for cross-step corroboration") +def test_contradiction_resolution_marks_resolved(): + """_run_step must mark actionable contradictions resolved after re-search. + + Regression guard: Contradiction.resolved was never set, so contradicted + sources were never retired and every contradiction surfaced as + "unresolved" in the report. + """ + from orchestrator import Orchestrator + import inspect + + src = inspect.getsource(Orchestrator._run_step) + assert ( + ".resolved = True" in src + ), "_run_step must set Contradiction.resolved = True after targeted re-search" + assert ( + ".resolution" in src + ), "_run_step must record a resolution note on resolved contradictions" + print("✅ _run_step marks actionable contradictions resolved after re-search") + + +def test_report_splits_source_disagreements(): + """Report assembly must separate genuine source disagreements (valid + multi-perspective findings) from unresolved factual contradictions.""" + with open("orchestrator.py") as fh: + source = fh.read() + + assert ( + "SOURCE PERSPECTIVES AND DISAGREEMENTS" in source + ), "report assembly must emit a dedicated source-perspectives section" + assert ( + "UNRESOLVED SOURCE CONTRADICTIONS" in source + ), "report assembly must keep an unresolved-contradictions section" + assert ( + "source_perspective_disagreements" in source + ), "report assembly must build a source_perspective_disagreements list" + assert ( + 'contradiction_type == "source_disagreement"' in source + ), "report assembly must branch on contradiction_type == 'source_disagreement'" + print("✅ report assembly splits source disagreements from factual contradictions") + + +def test_batch_memory_recall_wiring(): + """execute() must recall long-term memory per batch and pass it to the + search and analyst agents via an ltm_context parameter.""" + from orchestrator import Orchestrator + import inspect + + exec_src = inspect.getsource(Orchestrator.execute) + assert ( + "_batch_memory" in exec_src + ), "execute must maintain a per-batch long-term memory cache" + assert ( + "_recall_memories(" in exec_src + ), "execute must recall long-term memory at batch start" + + search_params = inspect.signature(Orchestrator._run_search).parameters + analyst_params = inspect.signature(Orchestrator._run_analyst).parameters + assert ( + "ltm_context" in search_params + ), "_run_search must accept an ltm_context parameter" + assert ( + "ltm_context" in analyst_params + ), "_run_analyst must accept an ltm_context parameter" + print("✅ batch-level LTM recall wired into execute, _run_search, _run_analyst") + + +def test_persist_step_id_filter(): + """persist_to_long_term_memory must support step_id_filter and _run_step + must use it for incremental per-step persistence.""" + from search_result_store import SearchResultStore + import inspect + + persist_params = inspect.signature( + SearchResultStore.persist_to_long_term_memory + ).parameters + assert ( + "step_id_filter" in persist_params + ), "persist_to_long_term_memory must accept a step_id_filter parameter" + + from orchestrator import Orchestrator + + run_step_src = inspect.getsource(Orchestrator._run_step) + assert ( + "step_id_filter=step.id" in run_step_src + ), "_run_step must persist incrementally with step_id_filter=step.id" + print("✅ persist_to_long_term_memory step_id_filter honored by _run_step") + + def test_qa_loop_data_poverty_exit(): """Verify QA retry path checks chunk_count for data poverty early exit.""" from orchestrator import Orchestrator @@ -2144,6 +2232,11 @@ def test_config_graph_options(): test_run_analyst_strips_coverage_notes, test_run_analyst_cross_step_retrieval, test_qa_loop_data_poverty_exit, + # Audit-remediation regression guards + test_contradiction_resolution_marks_resolved, + test_report_splits_source_disagreements, + test_batch_memory_recall_wiring, + test_persist_step_id_filter, # Self-optimize pipeline fixes test_self_optimize_uses_long_term_memory, test_self_optimize_api_server_wiring, diff --git a/backend/unit_tests.py b/backend/unit_tests.py index 44766af..328774e 100644 --- a/backend/unit_tests.py +++ b/backend/unit_tests.py @@ -692,6 +692,55 @@ async def test_qa_retries_on_flagged_contradiction(self, make_orchestrator): # Audit must have been called at least twice (initial + retry) assert orch.agents.loop.audit.await_count >= 2 + async def test_actionable_contradiction_marked_resolved_after_research( + self, make_orchestrator + ): + """Regression: an actionable contradiction whose targeted re-search yields + new evidence must be flagged ``resolved=True`` with a resolution note. + + Previously ``Contradiction.resolved`` was never set, so contradicted + sources were never retired and every contradiction surfaced as + "unresolved" in the final report. + """ + from orchestrator import AuditResult, Contradiction + from search_result_store import SearchResultStore + + contradiction = Contradiction( + source_a="http://a.com", + claim_a="X is true", + source_b="http://b.com", + claim_b="X is false", + contradiction_type="factual_error", + targeted_query="Is X true or false? primary source", + ) + flagged = AuditResult(contradictions=[contradiction], verdict="flagged") + clean = AuditResult(contradictions=[], verdict="clean") + + orch = make_orchestrator(depth="moderate") + orch.agents.loop.audit = AsyncMock(side_effect=[flagged, clean]) + orch.context.save_step("query", "test") + + # chunk_count must increase across the supplemental search so the + # data-poverty early-exit is not taken and the resolution path runs. + orch._search_store = MagicMock(spec=SearchResultStore) + orch._search_store.chunk_count = MagicMock(side_effect=[0, 1]) + orch._search_store.retrieve = AsyncMock(return_value="") + orch._search_store.add = AsyncMock() + orch._search_store.mark_superseded_by_step = MagicMock(return_value=0) + orch._search_store.filter_long_term_memories = AsyncMock(return_value=[]) + orch._search_store.persist_to_long_term_memory = AsyncMock(return_value=0) + + from pipeline import PipelineRunner + + runner = PipelineRunner() + step = self._step() + + await _collect(orch._run_step(step, "test", tools=[], pipeline_runner=runner)) + + assert contradiction.resolved is True + assert contradiction.resolution # non-empty resolution note recorded + assert any(c.resolved for c in orch._contradictions) + # =========================================================================== # Group 6 — Distillation helpers @@ -1242,3 +1291,38 @@ async def test_store_source_executes_merge_on_source_node(self): called_cyphers = [str(c.args[0]) for c in session_mock.run.call_args_list] # Should MERGE a :Document node (not :Source) assert any("Document" in q or "MERGE" in q for q in called_cyphers) + + async def test_store_relationship_auto_upserts_missing_endpoints(self): + """store_relationship must auto-create missing endpoint entities. + + Regression: ``MATCH (s) MATCH (t) CREATE (s)-[...]->(t)`` silently + no-ops (binding nothing) when either endpoint entity does not yet + exist, yet the method still returned ``success: True`` — losing the + edge. The fix verifies both endpoints first and upserts any missing + one before creating the relationship. + """ + # Default session.single() -> None, so _entity_exists() returns False + # for both endpoints, forcing the auto-upsert path. + ltm, session_mock = self._build_ltm_with_mocked_driver() + ltm.graph.upsert_entity = AsyncMock( + return_value={"success": True, "entity_id": "ent-x"} + ) + + with patch("embeddings.EMBEDDINGS_ENABLED", False): + result = await ltm.graph.store_relationship( + "Solar power", + "Carbon emissions", + "reduces", + session_id="sess1", + ) + + upserted_names = { + c.kwargs.get("name") for c in ltm.graph.upsert_entity.await_args_list + } + assert "Solar power" in upserted_names + assert "Carbon emissions" in upserted_names + # A relationship CREATE must still run after the endpoints are ensured. + called_cyphers = [str(c.args[0]) for c in session_mock.run.call_args_list] + assert any("CREATE" in q for q in called_cyphers) + assert result.get("success") is True + diff --git a/docs/API_SERVER.md b/docs/API_SERVER.md index 540647f..93605a6 100644 --- a/docs/API_SERVER.md +++ b/docs/API_SERVER.md @@ -14,16 +14,21 @@ 5. [REST Endpoints](#rest-endpoints) - [GET /](#get-) - [GET /health](#get-health) + - [GET /agent/research-methods](#get-agentresearch-methods) + - [GET /project-docs](#get-project-docs) + - [GET /project-docs/{filename}](#get-project-docsfilename) - [GET /mcp/servers](#get-mcpservers) + - [POST /mcp/servers](#post-mcpservers) + - [GET /mcp/servers/{name}](#get-mcpserversname) + - [DELETE /mcp/servers/{name}](#delete-mcpserversname) - [GET /mcp/tools](#get-mcptools) - - [POST /files/upload](#post-filesupload) - [GET /config](#get-config) - [POST /config](#post-config) - [POST /chat](#post-chat) - [Knowledge Graph Endpoints](#knowledge-graph-endpoints) 6. [WebSocket Endpoint](#websocket-endpoint) - [/ws/research (Orchestrator)](#wsresearch-orchestrator) -7. [Shared Message Contract](#shared-message-contract) +7. [Message Contract](#message-contract) - [Inbound Messages](#inbound-messages) - [Outbound Messages](#outbound-messages) 8. [Serialisation Helper](#serialisation-helper) @@ -54,17 +59,25 @@ create_model_backend(config) ← selects backend from MODEL_BACKEND env ▼ create_mcp_registry(config) ← connects to all MCP servers │ - ├── memory (MEMORY_SERVER_URL) ├── web_search (SEARCH_SERVER_URL) - ├── fetch (FETCH_SERVER_URL) + ├── fetch (FETCH_SERVER_URL) └── file_handler (FILE_SERVER_URL) │ - ├── agent = AdvancedResearchAgent | SelfOptimizingAgent - │ (stored on app.state.agent; used for /mcp/* and /chat) - ├── agent_pool = AgentPool.from_single_backend(backend) - │ (shared across all WebSocket sessions) - ├── session_store = SessionStore() (in-memory, TTL-based) - └── session_manager = ResearchSessionManager() (owns all background tasks) + ▼ +AsyncLongTermMemory(...).async_init() ← in-process Neo4j-backed memory + │ (flat vector store + knowledge graph) + │ + ├── agent = ResearchAgent | SelfOptimizingAgent + │ (stored on app.state.agent; used for /mcp/* and /chat) + ├── self_optimizing_agent = SelfOptimizingAgent + │ (always created; powers the self_optimize WS command) + ├── agent_pool = AgentPool.from_single_backend(backend) + │ (shared across all WebSocket sessions) + ├── session_store = SessionStore() (in-memory, TTL-based) + └── session_manager = ResearchSessionManager() (owns all background tasks) + │ + ▼ +warm_up_embeddings() ← pre-loads the sentence-transformer model │ ▼ mcp_registry.start_keepalive() ← 60 s pings to all MCP servers @@ -77,7 +90,9 @@ Application shutdown └── mcp_registry.close() ``` -The `agent_pool` is created once at startup and shared across all WebSocket connections. Each incoming `query` message creates a **fresh, isolated `Orchestrator`** instance (via `_make_orchestrator()`) so concurrent research sessions never share mutable plan or execution state. +There is **no standalone memory MCP server** — long-term memory is handled in-process by `AsyncLongTermMemory`, backed by Neo4j (see [docs/RAG_NOTES.md](./RAG_NOTES.md)). + +The `agent_pool` is created once at startup and shared across all WebSocket connections. Each incoming `query` message creates a **fresh, isolated `Orchestrator`** instance (via `_make_orchestrator()`) so concurrent research sessions never share mutable plan or execution state. The expensive shared resources — `AgentPool`, MCP registry, and long-term memory store — are reused from app state. --- @@ -87,11 +102,11 @@ The `agent_mode` field in `Config` (set via the `AGENT_MODE` environment variabl | `agent_mode` value | Agent class | Notes | |---|---|---| -| `"research"` *(default)* | `AdvancedResearchAgent` | Full research pipeline with parallel step execution and source credibility scoring | -| `"chat"` | `AdvancedResearchAgent` | Same class; mode affects prompt behaviour | -| `"self-optimization"` | `SelfOptimizingAgent` | Experimental agent that analyses stored memories and develops new research strategies | +| `"research"` *(default)* | `ResearchAgent` | Single-agent helper used by the `/chat` and `/mcp/*` REST endpoints | +| `"chat"` | `ResearchAgent` | Same class; mode affects prompt behaviour | +| `"self-optimization"` | `SelfOptimizingAgent` | Agent that analyses stored memories and the knowledge graph, then rewrites `RESEARCH-METHODS.md` | -The `agent` instance is used only by REST endpoints (`/chat`, `/mcp/*`) and the v1 AdvancedResearchAgent codepath. The WebSocket `/ws/research` endpoint always uses a fresh `Orchestrator` per session regardless of `AGENT_MODE`. +The `agent` instance is used only by the REST endpoints (`/chat`, `/mcp/*`). The WebSocket `/ws/research` endpoint always uses a fresh `Orchestrator` per session regardless of `AGENT_MODE`. A dedicated `SelfOptimizingAgent` is also created at startup (independent of `AGENT_MODE`) to power the `self_optimize` WebSocket command. --- @@ -137,62 +152,125 @@ Detailed health check. Confirms the agent has been initialised. --- -### GET `/mcp/servers` +### GET `/agent/research-methods` -Lists the names of all MCP servers that are currently registered and connected. +Returns the contents of the `RESEARCH-METHODS.md` instruction document that the Orchestrator injects into planning. This is the playbook the self-optimization pipeline rewrites. **Response** ```json -{ - "servers": ["memory", "web_search", "fetch", "file_handler"] -} +{ "content": "# Research Methods\n\n..." } ``` -**503** — agent not initialised. +**404** — the instruction file could not be found. --- -### GET `/mcp/tools` +### GET `/project-docs` -Lists every tool available across all connected MCP servers. The response is the raw tool-spec list used internally for model tool-calling. +Lists the Markdown documentation filenames available under the backend `docs/` directory (bind-mounted from the repo `docs/` folder). + +**Response** + +```json +{ "docs": ["API_SERVER.md", "BENCHMARKING.md", "CLI.md", "MCP_SERVERS.md", "ORCHESTRATOR.md", "RAG_NOTES.md"] } +``` + +> **Note:** the route is `/project-docs`, not `/docs` — the latter is reserved by FastAPI for the Swagger UI. + +--- + +### GET `/project-docs/{filename}` + +Returns the raw content of a single documentation file. The filename is sanitised (`os.path.basename`, `.md`-only) to prevent path traversal. + +**Response** + +```json +{ "filename": "API_SERVER.md", "content": "# API Server — Architecture & Reference\n\n..." } +``` + +**400** — invalid filename. **404** — file not found. + +--- + +### GET `/mcp/servers` + +Returns full info for every registered MCP server (built-in and user-defined), including connection status and transport. **Response** ```json { - "tools": [ - { - "type": "function", - "name": "web_search", - "description": "Search the web for a query", - "parameters": { "type": "object", "properties": { "query": { "type": "string" } }, "required": ["query"] } - }, - ... + "servers": [ + { "name": "web_search", "transport": "streamable-http", "builtin": true, "status": "connected" }, + { "name": "fetch", "transport": "streamable-http", "builtin": true, "status": "connected" }, + { "name": "file_handler", "transport": "streamable-http", "builtin": true, "status": "connected" } ] } ``` -**503** — agent not initialised. - --- -### POST `/files/upload` +### POST `/mcp/servers` -Uploads a file and saves it under `backend/data/uploaded_files/`. The file is stored on the server's local filesystem and is then accessible to the `file_handler` MCP server. +Registers a new **user-defined** MCP server at runtime and persists it. Supports `stdio`, `sse`, and `streamable-http` transports. Built-in servers cannot be replaced. -**Request** — `multipart/form-data` +**Request body** (`application/json`) | Field | Type | Required | Description | |---|---|---|---| -| `file` | binary | ✓ | The file to upload | +| `name` | `string` | ✓ | Unique server name | +| `transport` | `string` | ✗ | `streamable-http` *(default)*, `sse`, or `stdio` | +| `url` | `string` | ✗ | Endpoint URL (for `sse` / `streamable-http`) | +| `command` | `string` | ✗ | Executable (for `stdio`) | +| `args` | `string[]` | ✗ | Command arguments (for `stdio`) | +| `env` | `object` | ✗ | Environment variables (for `stdio`) | +| `headers` | `object` | ✗ | Extra HTTP headers | + +**Response** — the stored server config. **400** — missing name / unsupported transport. **409** — name collides with a built-in server. **500** — registration failed. + +--- + +### GET `/mcp/servers/{name}` + +Returns the stored config plus live capability listings (tools, resources, resource templates, prompts) for a single server. If the server is registered but not connected, capability lists are empty and `status` is `"disconnected"`. + +**404** — server not found. + +--- + +### DELETE `/mcp/servers/{name}` + +Removes a user-defined MCP server and persists the change. + +**Response** + +```json +{ "status": "deleted", "name": "my_server" } +``` + +**403** — built-in servers cannot be deleted. **404** — server not found. + +--- + +### GET `/mcp/tools` + +Lists every tool available across all connected MCP servers. The response is the raw tool-spec list used internally for model tool-calling. **Response** ```json { - "status": "success", - "filename": "my_document.pdf" + "tools": [ + { + "type": "function", + "name": "web_search", + "description": "Search the web for a query", + "parameters": { "type": "object", "properties": { "query": { "type": "string" } }, "required": ["query"] } + }, + ... + ] } ``` @@ -209,10 +287,11 @@ Returns a snapshot of the current `Config` object (all fields, including model n ```json { "config": { - "model_backend": "openai", - "openai_model": "gpt-5.2", - "ollama_model": "nemotron-3-nano", - "max_iterations": 10, + "model_backend": "ollama", + "openai_heavy_model": "gpt-5.2", + "openai_light_model": "gpt-5-nano", + "ollama_heavy_model": "nemotron-3-nano", + "max_iterations": 5, "default_temperature": 0.7, ... } @@ -223,29 +302,33 @@ Returns a snapshot of the current `Config` object (all fields, including model n ### POST `/config` -Updates the active agent's model backend and/or model name at runtime. Reinitialises the `AdvancedResearchAgent` with the new configuration — the `Orchestrator` is **not** reinitialised (restart the server to apply backend changes to the Orchestrator). +Updates model configuration at runtime, persists it to `configs/model_settings.json`, and **reinitialises the model backend and `agent_pool`** so changes take effect for all new research sessions without a restart. In-flight sessions are unaffected. (Note: persisted settings are re-applied on the next startup.) -**Request body** (`application/json`) — `ConfigUpdate` +**Request body** (`application/json`) — `ConfigUpdate`. All fields are optional; only the ones provided are applied. Credential/model fields are routed to the currently-selected backend. -| Field | Type | Required | Description | -|---|---|---|---| -| `model_backend` | `string` | ✗ | One of `openai`, `ollama`, `aws`, `azure`, `gcp`, `huggingface` | -| `model` | `string` | ✗ | Model name to use for the selected backend | +| Field | Type | Description | +|---|---|---| +| `model_backend` | `string` | Switch active backend: `openai`, `azure`, `aws`, `bedrock`, `gcp`, `ollama`, `huggingface`, `anthropic` | +| `api_key` | `string` | API key for the selected backend | +| `api_base_url` | `string` | Base URL / endpoint for the selected backend (for `bedrock`, the AWS region) | +| `heavy_model` | `string` | Model for heavy tasks (planning, synthesis, report) | +| `light_model` | `string` | Model for light tasks (search execution, analyst extraction) | +| `model` | `string` | Legacy single-model field | +| `root_model_override` / `search_model_override` / `analyst_model_override` / `qa_model_override` | `string` | Per-agent model override (empty string clears it) | +| `root_temperature` / `search_temperature` / `analyst_temperature` / `qa_temperature` | `float` | Per-agent sampling temperature | +| `root_top_p` / `search_top_p` / `analyst_top_p` / `qa_top_p` | `float` | Per-agent nucleus-sampling threshold | +| `root_max_tokens` / `search_max_tokens` / `analyst_max_tokens` / `qa_max_tokens` | `int` | Per-agent max output tokens | **Response** ```json { "status": "success", - "message": "Configuration updated", - "config": { - "model_backend": "openai", - "model": "gpt-5.2" - } + "config": { "model_backend": "openai", "openai_heavy_model": "gpt-5.2", ... } } ``` -**500** — invalid configuration or failed to reinitialise the agent. +**500** — invalid configuration or failed to reinitialise the model backend / agent pool. --- @@ -258,6 +341,7 @@ Streaming chat endpoint for direct conversational interaction with the agent, by | Field | Type | Required | Description | |---|---|---|---| | `content` | `string` | ✓ | The user's message | +| `role` | `string` | ✗ | Message role (e.g. `user`); ignored by the handler | **Response** — `text/plain` streamed in real-time as the model generates tokens. @@ -271,7 +355,7 @@ The latest advancements in renewable energy include... ### Knowledge Graph Endpoints -All graph endpoints query the `KnowledgeGraph` layer of `AsyncLongTermMemory`. They return `{"message": "LongTermMemory not available"}` if Neo4j is not connected. +All graph endpoints query the `KnowledgeGraph` layer of `AsyncLongTermMemory`. They return **503** (`{"detail": "Knowledge graph not available"}`) if Neo4j is not connected. #### GET `/graph/stats` @@ -397,9 +481,9 @@ See [ORCHESTRATOR.md](./ORCHESTRATOR.md) for the complete architecture reference --- -## Shared Message Contract +## Message Contract -Both WebSocket endpoints share the same JSON message protocol. +The `/ws/research` endpoint uses the JSON message protocol below. ### Inbound Messages @@ -412,13 +496,15 @@ Starts a new research session. Triggers plan generation. ```json { "type": "query", - "content": "What are the long-term cardiovascular effects of GLP-1 agonists?" + "content": "What are the long-term cardiovascular effects of GLP-1 agonists?", + "research_depth": "shallow" } ``` | Field | Type | Required | Description | |---|---|---|---| | `content` | string | ✓ | The research question | +| `research_depth` | string | ✗ | `shallow` *(default)*, `moderate`, or `deep`. Controls whether the QA/contradiction loop runs and how many retries it gets. Invalid values fall back to `shallow`. | --- @@ -492,6 +578,18 @@ Reconnects to an existing research session by `session_id`. The server replays t --- +#### `self_optimize` + +Runs the five-phase self-optimization workflow on the dedicated `SelfOptimizingAgent`: it analyses session logs and the knowledge graph, then rewrites `RESEARCH-METHODS.md`. Available regardless of `AGENT_MODE`. Emits an `optimize_started` event, a stream of `optimize_progress` events, and a terminal `optimize_complete` event. + +```json +{ + "type": "self_optimize" +} +``` + +--- + ### Outbound Messages All messages sent **from the server to the client** have at minimum `type` and `message` fields. @@ -627,9 +725,9 @@ A plan step finished successfully. --- -#### `step_failed` *(v1 only)* +#### `step_failed` -A plan step failed. Research continues with remaining steps. +A plan step failed. Research continues with the remaining steps. ```json { @@ -654,30 +752,22 @@ All plan steps have finished executing. Synthesis is about to begin. --- -#### `synthesis` *(v1 only)* +#### `synthesis_progress` -The final structured research synthesis from `AdvancedResearchAgent`. Follows the `SynthesisResult` shape. +Incremental progress updates emitted during the multi-pass synthesis phase (outline → section drafting → assembly). ```json { - "type": "synthesis", - "message": "Synthesis complete.", - "data": { - "summary": "GLP-1 agonists demonstrate consistent MACE reduction…", - "key_insights": ["..."], - "patterns": ["..."], - "recommendations": ["..."], - "creative_applications": ["..."], - "knowledge_gaps": ["..."] - } + "type": "synthesis_progress", + "message": "Drafting section 2 of 5: Cardiovascular Outcomes…" } ``` --- -#### `report` *(v2 only)* +#### `report` -The final Markdown research report from the `ReportComposer`. Sent as the last event of a v2 session. +The final Markdown research report from the `ReportComposer`. Sent as the last substantive event of a session. ```json { @@ -704,14 +794,14 @@ Acknowledges the plan rejection and suggests next steps. --- -#### `research_paused` / `research_resumed` / `research_stopped` *(v1 only)* +#### `optimize_started` / `optimize_progress` / `optimize_complete` -Acknowledgements for in-flight control actions. +Emitted by the `self_optimize` command. `optimize_started` is sent once when the workflow begins, followed by a stream of `optimize_progress` updates as each phase runs, and a terminal `optimize_complete` carrying the result. ```json -{ "type": "research_paused", "message": "Research paused. Send 'resume' to continue." } -{ "type": "research_resumed", "message": "Research resumed." } -{ "type": "research_stopped", "message": "Research stopped." } +{ "type": "optimize_started", "message": "Self-optimization workflow starting…" } +{ "type": "optimize_progress", "message": "Analysing 42 stored memories…" } +{ "type": "optimize_complete", "message": "RESEARCH-METHODS.md updated." } ``` --- @@ -760,35 +850,40 @@ This is applied to every outbound WebSocket message before transmission. ## Environment Variables -All variables are read via the `Config` Pydantic model at startup. The most operationally relevant ones for the API server are: +All variables are read via the `Config` Pydantic model at startup. The most operationally relevant ones for the API server are below; see [config.py](../backend/config.py) and the root `README.md` for the complete list. | Variable | Default | Description | |---|---|---| -| `MODEL_BACKEND` | `ollama` | Model provider: `openai`, `ollama`, `aws`, `azure`, `gcp`, `huggingface` | +| `MODEL_BACKEND` | `ollama` | Model provider: `openai`, `azure`, `aws`, `bedrock`, `gcp`, `ollama`, `huggingface` | | `AGENT_MODE` | `research` | Agent class for REST endpoints: `research`, `chat`, `self-optimization` | | `LOG_LEVEL` | `20` (INFO) | Python logging level integer | -| `MEMORY_SERVER_URL` | `http://localhost:9494/mcp` | MCP memory server URL | | `SEARCH_SERVER_URL` | `http://localhost:9393/mcp` | MCP web search server URL | -| `SCRAPER_SERVER_URL` | `http://localhost:9292/mcp` | MCP web scraper server URL | +| `FETCH_SERVER_URL` | `http://localhost:9292/mcp` | MCP fetch (web scraper) server URL | | `FILE_SERVER_URL` | `http://localhost:9191/mcp` | MCP file handler server URL | | `OPENAI_API_KEY` | — | Required when `MODEL_BACKEND=openai` | -| `OPENAI_MODEL` | `gpt-5.2` | OpenAI model name | +| `OPENAI_HEAVY_MODEL` | `gpt-5.2` | OpenAI model for heavy tasks (planning, synthesis) | +| `OPENAI_LIGHT_MODEL` | `gpt-5-nano` | OpenAI model for light tasks (search, analyst) | | `OLLAMA_BASE_URL` | `http://localhost:11434` | Ollama server base URL | -| `OLLAMA_MODEL` | `nemotron-3-nano` | Ollama model name | -| `AWS_MODEL` | `global.anthropic.claude-sonnet-4-6` | Bedrock model ID | -| `AWS_BASE_URL` | — | Required when `MODEL_BACKEND=aws` | -| `AWS_API_KEY` | — | Required when `MODEL_BACKEND=aws` | -| `AZURE_OPENAI_ENDPOINT` | — | Required when `MODEL_BACKEND=azure` | -| `AZURE_OPENAI_API_KEY` | — | Required when `MODEL_BACKEND=azure` | +| `OLLAMA_HEAVY_MODEL` / `OLLAMA_LIGHT_MODEL` | `nemotron-3-nano` | Ollama model names | +| `AWS_MODEL` | `global.anthropic.claude-sonnet-4-5-20250929-v1:0` | Default Bedrock model ID | +| `AWS_HEAVY_MODEL` / `AWS_LIGHT_MODEL` | `…sonnet-4-6` / `…haiku-4-5…` | Bedrock heavy/light model IDs | +| `AWS_BASE_URL` / `AWS_API_KEY` | — | Required when `MODEL_BACKEND=aws` (OpenAI-compatible gateway) | +| `AWS_REGION` | `us-east-1` | Required when `MODEL_BACKEND=bedrock` (native boto3) | +| `AZURE_OPENAI_ENDPOINT` / `AZURE_OPENAI_API_KEY` | — | Required when `MODEL_BACKEND=azure` | | `AZURE_OPENAI_DEPLOYMENT` | — | Azure deployment name | -| `GCP_ENDPOINT` | `https://us-central1-aiplatform…` | GCP Vertex AI endpoint | +| `GCP_ENDPOINT` | `https://us-central1-aiplatform.googleapis.com/v1` | GCP Vertex AI endpoint | | `GCP_MODEL` | `gemini-2.5-pro` | Vertex AI model name | | `GCP_API_KEY` | — | Required when `MODEL_BACKEND=gcp` | -| `MAX_ITERATIONS` | `3` | Max tool-calling iterations per research step | +| `MAX_ITERATIONS` | `5` | Max tool-calling iterations per research step | | `MAX_QA_RETRIES` | `2` | Max LoopAgent contradiction-retry cycles per step | | `MAX_TOKENS` | `4096` | Max output tokens per generation call | | `DEFAULT_TEMPERATURE` | `0.7` | Default generation temperature | | `MAX_SOURCES_PER_QUERY` | `5` | Max sources retrieved per search step | | `MAX_DEPTH` | `2` | Max recursive scraping depth | +| `MAX_WORKERS` | `10` | Concurrent steps per research session | +| `MAX_CONCURRENT_SESSIONS` | `10` | Max research sessions executing simultaneously (0 = unlimited) | | `ENABLE_PARALLEL_EXECUTION` | `True` | Run independent plan steps concurrently | +| `NEO4J_URI` | `bolt://localhost:7687` | Long-term memory Neo4j Bolt URI | +| `NEO4J_USER` / `NEO4J_PASSWORD` | `neo4j` / `research_pass` | Neo4j credentials | +| `EMBEDDINGS_ENABLED` | `true` | Enable in-process sentence-transformer embeddings | | `MCP_KEEPALIVE_INTERVAL` | `60` | Seconds between MCP server keepalive pings | diff --git a/docs/BENCHMARKING.md b/docs/BENCHMARKING.md index 18201ce..228ce95 100644 --- a/docs/BENCHMARKING.md +++ b/docs/BENCHMARKING.md @@ -53,8 +53,8 @@ Each completed task is written to the output file immediately. If the run is int |------|---------|-------------| | `--ws-url` | `ws://localhost:9999/ws/research` | Backend WebSocket URL | | `--research-depth` | `shallow` | `shallow` / `moderate` / `deep` | -| `--concurrency` | `3` | Concurrent tasks (hard ceiling: `MAX_CONCURRENT_PIPELINES=5`) | -| `--task-timeout` | `1800` | Per-task timeout in seconds | +| `--concurrency` | `10` | Concurrent tasks (the backend caps simultaneous pipelines via `MAX_CONCURRENT_SESSIONS`, default `10`) | +| `--task-timeout` | `43200` | Per-task timeout in seconds (default: 12 hours) | | `--lang` | `both` | `en`, `zh`, or `both` | | `--limit N` | off | Run only the first N tasks (useful for smoke testing) | | `--resume` / `--no-resume` | resume on | Skip tasks already in the output file | @@ -124,6 +124,6 @@ The script connects a fresh WebSocket per task and follows the standard session 2. Receive `session_created` → receive `plan` → send `{"type": "approve_plan", "planId": ""}` 3. Wait for `{"type": "report", "data": {"document": "..."}}` -All intermediate events (`status`, `step_start`, `step_complete`, `research_complete`, `section_draft`) are silently consumed. Errors, timeouts, and early `plan_denied` / `research_stopped` events are captured and written as failure records. +All intermediate events (`status`, `step_start`, `step_complete`, `research_complete`, `synthesis_progress`, `section_draft`) are silently consumed. Errors, timeouts, and early `plan_denied` events are captured and written as failure records. -Concurrency is managed with `asyncio.Semaphore(--concurrency)`. The backend enforces its own `MAX_CONCURRENT_PIPELINES=5` limit independently. +Concurrency is managed with `asyncio.Semaphore(--concurrency)`. The backend enforces its own `MAX_CONCURRENT_SESSIONS` limit (default `10`, `0` = unlimited) independently. diff --git a/docs/MCP_SERVERS.md b/docs/MCP_SERVERS.md index 433c37d..4e5f29d 100644 --- a/docs/MCP_SERVERS.md +++ b/docs/MCP_SERVERS.md @@ -21,7 +21,7 @@ - [Tools](#tools-web-search) - [Result Shapes](#result-shapes) - [Configuration](#configuration-web-search) -4. [Web Scraper Server](#web-scraper-server) +4. [Fetch Server](#fetch-server) - [Purpose](#purpose-web-scraper) - [Scraping Pipeline](#scraping-pipeline) - [Tools](#tools-web-scraper) @@ -322,11 +322,12 @@ No API keys are required for DuckDuckGo, Wikipedia, or arXiv. All search backend --- -## Web Scraper Server +## Fetch Server -> **Directory:** `mcp/web_scrape/` +> **Directory:** `mcp/fetch/` > **Port:** `9292` -> **Service name:** `mcp-web-scraping-server` +> **Service name:** `mcp-fetch-server` +> **Registry key:** `fetch` ### Purpose @@ -389,7 +390,7 @@ The `httpx` client is a **module-level singleton** with `follow_redirects=True`. ### Tools (Web Scraper) -#### `scrape_url` +#### `fetch_url` Fetches and extracts the complete content of a web page. @@ -654,11 +655,11 @@ The agent interacts with these servers at different points in the research lifec | Phase | Component | Call | Purpose | |---|---|---|---| | `_execute_step` | Web Search MCP | `web_search`, `search_wikipedia`, `search_github` | Find relevant URLs and snippets | -| `_execute_step` | Web Scraper MCP | `scrape_url` | Fetch full page content from URLs found by search | +| `_execute_step` | Fetch MCP | `fetch_url` | Fetch full page content from URLs found by search | | `_execute_step` | File Handler MCP | `read_file`, `list_files` | Read any uploaded reference documents | | `synthesize_results` | File Handler MCP | `write_file` | Save the final Markdown research report | -In the v2 Orchestrator, the same servers are used but routing goes through the specialist sub-agents: the `SearchAgent` primarily drives `web_search` and `scrape_url`, while the `ReportComposer` output is saved manually by the Orchestrator after `synthesise()` completes. +In the Orchestrator, the same servers are used but routing goes through the specialist sub-agents: the `SearchAgent` primarily drives `web_search` and `fetch_url`, while the `ReportComposer` output is saved by the Orchestrator after `synthesise()` completes. --- @@ -680,7 +681,7 @@ Verify individual MCP server health: ```bash curl http://localhost:9393/health # Web Search -curl http://localhost:9292/health # Web Scraper +curl http://localhost:9292/health # Fetch curl http://localhost:9191/health # File Handler ``` diff --git a/docs/ORCHESTRATOR.md b/docs/ORCHESTRATOR.md index 3e657ea..837af4f 100644 --- a/docs/ORCHESTRATOR.md +++ b/docs/ORCHESTRATOR.md @@ -677,7 +677,7 @@ Client Server (/ws/research) | **Backend sharing** | Single backend | Per-role backends via `AgentPool` | | **State reset** | `_clear_state()` | `_reset_state()` | -The `/ws/research` endpoint is exclusively Orchestrator-powered. `ResearchAgent` and `AdvancedResearchAgent` are used only by the REST `/chat` endpoint and the CLI; they are not exposed over WebSocket. +The `/ws/research` endpoint is exclusively Orchestrator-powered. The legacy `ResearchAgent` is used only by the REST `/chat` endpoint and the CLI chat mode; it is not exposed over WebSocket. --- diff --git a/docs/RAG_NOTES.md b/docs/RAG_NOTES.md index 27acd8e..1b1feac 100644 --- a/docs/RAG_NOTES.md +++ b/docs/RAG_NOTES.md @@ -18,7 +18,7 @@ The RAG system has two layers: Built entirely **in-process** — no external vector database required. Every research session gets its own isolated `SearchResultStore` instance created at the start of `Orchestrator._reset_state()`. ``` -MCP tool result (web_scrape / web_search) +MCP tool result (fetch / web_search) │ ▼ Orchestrator._distill_tool_result() ← strips navigation chrome, @@ -108,13 +108,13 @@ These are `Config` fields in `config.py` (Pydantic, loaded from environment vari | `Config` field | Default | Env var | Purpose | |---|---|---|---| -| `analyst_top_k` | `8` | `ANALYST_TOP_K` | Chunks retrieved per AnalystAgent RAG query | +| `analyst_top_k` | `10` | `ANALYST_TOP_K` | Chunks retrieved per AnalystAgent RAG query | | `max_tool_result_chars_in_message` | `5000` | `MAX_TOOL_RESULT_CHARS_IN_MESSAGE` | Characters of a tool result kept in SearchAgent's sliding message window (full content in RAG store) | | `max_search_history_messages` | `8` | `MAX_SEARCH_HISTORY_MESSAGES` | Sliding window size for SearchAgent's `execution_messages` | | `max_analyst_fallback_chars` | `8000` | `MAX_ANALYST_FALLBACK_CHARS` | Chars of raw search output used when the RAG store has no chunks yet (first iteration) | -| `distill_max_chars` | `2000` | `DISTILL_MAX_CHARS` | Max chars the distillation filter preserves from a raw tool result before chunking | -| `step_summary_max_chars` | `800` | `STEP_SUMMARY_MAX_CHARS` | Max chars for per-step summary bullets fed to the multi-pass synthesis Outline Phase | -| `section_draft_top_k` | `6` | `SECTION_DRAFT_TOP_K` | Chunks retrieved per report section during Section-Drafting | +| `distill_max_chars` | `4000` | `DISTILL_MAX_CHARS` | Max chars the distillation filter preserves from a raw tool result before chunking | +| `step_summary_max_chars` | `1500` | `STEP_SUMMARY_MAX_CHARS` | Max chars for per-step summary bullets fed to the multi-pass synthesis Outline Phase | +| `section_draft_top_k` | `10` | `SECTION_DRAFT_TOP_K` | Chunks retrieved per report section during Section-Drafting | --- @@ -298,4 +298,4 @@ Each `Orchestrator` instance (one per WebSocket `query` message) creates its own - **Embedding throughput:** `all-MiniLM-L6-v2` handles batches of 20–30 chunks in < 100 ms on CPU. - **Retrieval latency:** Cosine similarity over a typical session's store (100–400 chunks) completes in < 10 ms. - **Dominant cost:** LLM inference (Bedrock/Claude Haiku 4.5 = ~7.4 s/call), not the RAG layer. -- **Log volume caution:** MCP `scrape_url` results can be very large (PDF scrapes > 100 MB). The `mcp_client` should truncate tool result bodies before logging at DEBUG level to avoid log rotation exhaustion. +- **Log volume caution:** MCP `fetch_url` results can be very large (PDF scrapes > 100 MB). The `mcp_client` should truncate tool result bodies before logging at DEBUG level to avoid log rotation exhaustion. diff --git a/src/App.tsx b/src/App.tsx index 424a23c..f64b907 100644 --- a/src/App.tsx +++ b/src/App.tsx @@ -19,6 +19,17 @@ const WS_URL = import.meta.env.VITE_WS_URL || `${location.protocol.replace('http const LS_SESSION_ID = 'deep_research_session_id'; const LS_CONV_ID = 'deep_research_conv_id'; +/** + * crypto.randomUUID() requires a secure context (https or localhost). + * Fall back to a simple pseudo-random id on plain http (e.g. LAN dev). + */ +function generateId(): string { + if (typeof crypto !== 'undefined' && typeof crypto.randomUUID === 'function') { + return crypto.randomUUID(); + } + return `${Date.now().toString(36)}-${Math.random().toString(36).slice(2)}-${Math.random().toString(36).slice(2)}`; +} + const AppContent: React.FC = () => { const { messages, @@ -67,7 +78,9 @@ const AppContent: React.FC = () => { const activeEventIndexRef = useRef(0); const [currentStatus, setCurrentStatus] = useState('Researching'); - const [sidebarOpen, setSidebarOpen] = useState(true); + const [sidebarOpen, setSidebarOpen] = useState(() => + typeof window !== 'undefined' && window.innerWidth >= 768 + ); const [researchDepth, setResearchDepth] = useState('shallow'); const { theme, toggleTheme } = useTheme(); const [currentPlanID, setCurrentPlanID] = useState(null); @@ -134,8 +147,7 @@ const AppContent: React.FC = () => { const { isConnected, isReconnecting, - messageQueue, - clearMessageQueue, + drainMessageQueue, sendMessage, reconnect, } = useWebSocket(WS_URL, { onReconnected: handleReconnected, onInitialConnect: handleInitialConnect }); @@ -175,9 +187,8 @@ const AppContent: React.FC = () => { // ===================== useEffect(() => { - if (!messageQueue.length) return; - const msgs = [...messageQueue]; - clearMessageQueue(); + const msgs = drainMessageQueue(); + if (!msgs.length) return; for (const rawMessage of msgs) { const { type, data, message, plan } = rawMessage; @@ -262,7 +273,7 @@ const AppContent: React.FC = () => { // Notify the user that we've reconnected successfully. if (researchConvIdRef.current) { addMessageToConv(researchConvIdRef.current, { - id: crypto.randomUUID(), + id: generateId(), role: 'system', type: 'system', content: rawMessage.complete @@ -333,7 +344,7 @@ const AppContent: React.FC = () => { if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'plan_approval', content: 'I have created a research plan. Please review it below.', @@ -360,7 +371,7 @@ const AppContent: React.FC = () => { if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'system', content: 'The research plan was denied. Please submit a new query to start again.', @@ -380,7 +391,7 @@ const AppContent: React.FC = () => { if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'step_start', content: `Starting step: ${data?.step?.description || 'Unknown Step'}`, @@ -404,7 +415,7 @@ const AppContent: React.FC = () => { } if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'step_complete', content: data?.skipped_synthesis @@ -427,7 +438,7 @@ const AppContent: React.FC = () => { } if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'step_failed', content: `${message || 'Step failed'}. Error: ${stepError || 'Unknown error.'}`, @@ -459,7 +470,7 @@ const AppContent: React.FC = () => { // Session is still alive until 'report' arrives; keep activeSessionIdRef. if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'research_complete', content: message || (data?.content ?? 'Research complete.'), @@ -483,7 +494,7 @@ const AppContent: React.FC = () => { const formattedSummary = synthParts.join('\n\n') || data?.summary || message || 'No synthesis data available.'; if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'synthesis', content: message || 'Synthesis complete.', @@ -519,7 +530,7 @@ const AppContent: React.FC = () => { } if (!isReplayedEvent && researchConvIdRef.current) { addMessageToConv(researchConvIdRef.current, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'error', content: message || (data?.message ?? 'An unexpected error occurred.'), @@ -561,7 +572,7 @@ const AppContent: React.FC = () => { setResearchEndTime(researchConvIdRef.current!, new Date().toISOString()); if (!isReplayedEvent) { addMessageToConv(researchConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'report', content: message || 'Research report complete.', @@ -611,6 +622,7 @@ const AppContent: React.FC = () => { break; case 'optimize_progress': { + if (isReplayedEvent) break; const phase = data?.phase; const phaseStatus = data?.status; console.log('[optimize] Phase:', phase, phaseStatus); @@ -622,7 +634,7 @@ const AppContent: React.FC = () => { } setCurrentStatus(message || 'Optimizing…'); addMessageToConv(optimizeConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'system', content: message || 'Optimization in progress…', @@ -632,12 +644,13 @@ const AppContent: React.FC = () => { } case 'optimize_complete': { + if (isReplayedEvent) break; console.log('[optimize] Complete:', data); setIsOptimizing(false); setCurrentStatus('Optimization Complete'); const methodsCount = data?.new_methods_count ?? 0; addMessageToConv(optimizeConvIdRef.current!, { - id: crypto.randomUUID(), + id: generateId(), role: 'assistant', type: 'system', content: `✅ Self-optimization complete. ${methodsCount} new research method(s) integrated into RESEARCH-METHODS.md.`, @@ -679,8 +692,7 @@ const AppContent: React.FC = () => { setEventIndex(researchConvIdRef.current, activeEventIndexRef.current); } }, [ - messageQueue, - clearMessageQueue, + drainMessageQueue, addMessageToConv, setIsResearching, setPlanStatus, @@ -702,6 +714,7 @@ const AppContent: React.FC = () => { updateOptimizePhase, getEventIndex, setEventIndex, + drainMessageQueue, // researchConvIdRef / optimizeConvIdRef / replayTotalRef / // pendingTotalEventsRef / activeEventIndexRef are refs; // no need to list them @@ -723,7 +736,7 @@ const AppContent: React.FC = () => { initOptimizeGraph(convId); setActiveView('graph'); addMessageToConv(convId, { - id: crypto.randomUUID(), + id: generateId(), role: 'user', type: 'user', content: '⚙ Run self-optimization workflow', @@ -755,7 +768,7 @@ const AppContent: React.FC = () => { localStorage.setItem(LS_CONV_ID, convId); addMessageToConv(convId, { - id: crypto.randomUUID(), + id: generateId(), role: 'user', type: 'user', content, @@ -778,14 +791,14 @@ const AppContent: React.FC = () => { } addMessageToConv(targetConvId, { - id: crypto.randomUUID(), + id: generateId(), role: 'user', type: 'user', content, timestamp: new Date(), }); - const responseMsgId = crypto.randomUUID(); + const responseMsgId = generateId(); addMessageToConv(targetConvId, { id: responseMsgId, role: 'assistant', @@ -840,7 +853,7 @@ const AppContent: React.FC = () => { setCurrentStatus('Researching'); addMessage({ - id: crypto.randomUUID(), + id: generateId(), role: 'system', type: 'system', content: '✅ Research plan approved. Starting research...', @@ -875,7 +888,7 @@ const AppContent: React.FC = () => { setCurrentStatus('Regenerating Plan'); addMessage({ - id: crypto.randomUUID(), + id: generateId(), role: 'system', type: 'system', content: `✏️ Plan modification requested: "${feedback}". Generating updated plan...`, @@ -940,7 +953,7 @@ const AppContent: React.FC = () => { setCurrentStatus('Regenerating Plan'); addMessage({ - id: crypto.randomUUID(), + id: generateId(), role: 'system', type: 'system', content: '🔄 Plan denied. Generating a new research plan...', @@ -970,7 +983,7 @@ const AppContent: React.FC = () => { setCurrentStatus(''); addMessage({ - id: crypto.randomUUID(), + id: generateId(), role: 'system', type: 'system', content: '🛑 Research inquiry ended. Submit a new query to start again.', @@ -982,7 +995,7 @@ const AppContent: React.FC = () => { return ( <> -
+
{/* Sidebar */} {sidebarOpen && ( { researchDepth={researchDepth} onResearchDepthChange={setResearchDepth} isResearching={isResearching} + onMobileClose={() => setSidebarOpen(false)} /> )} @@ -1046,8 +1060,7 @@ const AppContent: React.FC = () => {
diff --git a/src/components/FileUploader.tsx b/src/components/FileUploader.tsx index c353d5b..a1d318c 100644 --- a/src/components/FileUploader.tsx +++ b/src/components/FileUploader.tsx @@ -8,6 +8,8 @@ interface ServerFile { } interface LocalFile { + /** Stable unique key for React rendering. */ + id: string; name: string; size: number; uploading?: boolean; @@ -26,7 +28,7 @@ export const FileUploader: React.FC = () => { setLoadError(null); try { const data: ServerFile[] = await apiClient.listFiles(); - setServerFiles(data.map(f => ({ name: f.name, size: f.size_bytes }))); + setServerFiles(data.map((f) => ({ id: f.name, name: f.name, size: f.size_bytes }))); } catch { setLoadError('Could not load files from the file handler server.'); } finally { @@ -66,23 +68,26 @@ export const FileUploader: React.FC = () => { const uploadFiles = async (newFiles: File[]) => { if (newFiles.length === 0) return; - // Optimistically add uploading placeholders - const placeholders: LocalFile[] = newFiles.map(f => ({ + // Optimistically add uploading placeholders with a stable unique id. + const placeholders: LocalFile[] = newFiles.map((f) => ({ + id: `upload-${Date.now()}-${Math.random().toString(36).slice(2)}-${f.name}`, name: f.name, size: f.size, uploading: true, })); setServerFiles(prev => [...prev, ...placeholders]); - for (const file of newFiles) { + for (let i = 0; i < newFiles.length; i++) { + const file = newFiles[i]; + const placeholder = placeholders[i]; try { await apiClient.uploadFile(file); setServerFiles(prev => - prev.map(f => f.name === file.name && f.uploading ? { name: f.name, size: f.size } : f) + prev.map(f => f.id === placeholder.id ? { id: f.name, name: f.name, size: f.size } : f) ); } catch { setServerFiles(prev => - prev.map(f => f.name === file.name && f.uploading ? { ...f, uploading: false, error: true } : f) + prev.map(f => f.id === placeholder.id ? { ...f, uploading: false, error: true } : f) ); } } @@ -181,7 +186,7 @@ export const FileUploader: React.FC = () => {
{serverFiles.map((file) => (
diff --git a/src/components/MCPServerManager.tsx b/src/components/MCPServerManager.tsx new file mode 100644 index 0000000..9ac7b74 --- /dev/null +++ b/src/components/MCPServerManager.tsx @@ -0,0 +1,642 @@ +import React, { useState, useEffect, useCallback } from 'react'; +import { Plus, Server, CheckCircle, XCircle, Trash2, Lock, ChevronLeft, RefreshCw, Wrench, FileText, MessageSquare, Code } from 'lucide-react'; +import { MCPServer } from '../types'; +import { apiClient } from '../api/client'; + +type Transport = 'streamable-http' | 'sse' | 'stdio'; + +interface NewServerForm { + name: string; + transport: Transport; + // HTTP transports + url: string; + headers: string; // JSON string — only for sse / streamable-http + // stdio + command: string; + args: string; + env: string; +} + +interface MCPTool { + name: string; + description?: string; + parameters?: Record; +} + +interface MCPResource { + uri: string; + name?: string; + description?: string; + mimeType?: string; +} + +interface MCPResourceTemplate { + uriTemplate: string; + name?: string; + description?: string; + mimeType?: string; +} + +interface MCPPrompt { + name: string; + description?: string; + arguments?: Array<{ name: string; description?: string; required?: boolean }>; +} + +interface ServerDetail { + server: MCPServer; + tools: MCPTool[]; + resources: MCPResource[]; + resource_templates: MCPResourceTemplate[]; + prompts: MCPPrompt[]; +} + +const EMPTY_FORM: NewServerForm = { + name: '', + transport: 'streamable-http', + url: '', + headers: '', + command: '', + args: '', + env: '', +}; + +/** Return true if a header key looks like it carries a secret. */ +function isSensitiveHeader(key: string): boolean { + const lower = key.toLowerCase(); + return ( + lower.includes('auth') || + lower.includes('token') || + lower.includes('key') || + lower.includes('secret') || + lower.includes('password') || + lower.includes('credential') + ); +} + +function maskValue(value: string): string { + if (value.length <= 8) return '••••••••'; + return value.slice(0, 4) + '••••••••' + value.slice(-4); +} + +const TRANSPORT_LABELS: Record = { + 'streamable-http': 'Streamable HTTP', + sse: 'SSE', + stdio: 'stdio', +}; + +function ServerStatus({ status }: { status: MCPServer['status'] }) { + if (status === 'connected') + return ; + if (status === 'error') + return ; + return ; +} + +type DetailTab = 'tools' | 'resources' | 'prompts'; + +function ServerDetailPanel({ + detail, + onBack, + onRefresh, + loading, +}: { + detail: ServerDetail; + onBack: () => void; + onRefresh: () => void; + loading: boolean; +}) { + const [tab, setTab] = useState('tools'); + const { server, tools, resources, resource_templates, prompts } = detail; + + const allResources = [ + ...resources.map((r) => ({ ...r, isTemplate: false })), + ...resource_templates.map((r) => ({ uri: r.uriTemplate, name: r.name, description: r.description, mimeType: r.mimeType, isTemplate: true })), + ]; + + const tabCount: Record = { + tools: tools.length, + resources: allResources.length, + prompts: prompts.length, + }; + + const tabClass = (t: DetailTab) => + `px-3 py-1.5 text-xs font-medium rounded transition-colors flex items-center gap-1 ${tab === t + ? 'bg-blue-500 dark:bg-emerald-600 text-white' + : 'text-gray-600 dark:text-gray-400 hover:bg-gray-200 dark:hover:bg-gray-700' + }`; + + const itemBase = + 'p-3 rounded-lg bg-gray-50 dark:bg-gray-800 border border-gray-100 dark:border-gray-700'; + + return ( +
+ {/* Header */} +
+ +
+
+ + + {server.name} + + {server.builtin && ( + + )} + + {TRANSPORT_LABELS[server.transport as Transport] ?? server.transport} + +
+

+ {server.transport === 'stdio' + ? [server.command, ...(server.args || [])].filter(Boolean).join(' ') + : server.url} +

+ {server.headers && Object.keys(server.headers).length > 0 && ( +
+ {Object.entries(server.headers).map(([k, v]) => ( + + {k}: + {isSensitiveHeader(k) ? maskValue(v) : v} + + ))} +
+ )} +
+
+ + +
+
+ + {/* Tabs */} +
+ + + +
+ + {/* Content */} +
+ {tab === 'tools' && ( + <> + {tools.length === 0 ? ( +

No tools available

+ ) : ( + tools.map((tool) => ( +
+
+ +
+

+ {tool.name} +

+ {tool.description && ( +

+ {tool.description} +

+ )} + {tool.parameters?.properties && Object.keys(tool.parameters.properties).length > 0 && ( +
+ {Object.entries(tool.parameters.properties as Record).map(([param, schema]) => { + const required = (tool.parameters?.required as string[] | undefined)?.includes(param); + return ( + + {param} + {schema?.type ? `: ${schema.type}` : ''} + {!required && '?'} + + ); + })} +
+ )} +
+
+
+ )) + )} + + )} + + {tab === 'resources' && ( + <> + {allResources.length === 0 ? ( +

No resources available

+ ) : ( + allResources.map((res, i) => ( +
+
+ +
+
+

+ {res.uri} +

+ {res.isTemplate && ( + + template + + )} + {res.mimeType && ( + {res.mimeType} + )} +
+ {res.name && res.name !== res.uri && ( +

{res.name}

+ )} + {res.description && ( +

+ {res.description} +

+ )} +
+
+
+ )) + )} + + )} + + {tab === 'prompts' && ( + <> + {prompts.length === 0 ? ( +

No prompts available

+ ) : ( + prompts.map((prompt) => ( +
+
+ +
+

+ {prompt.name} +

+ {prompt.description && ( +

+ {prompt.description} +

+ )} + {prompt.arguments && prompt.arguments.length > 0 && ( +
+ {prompt.arguments.map((arg) => ( + + {arg.name}{!arg.required && '?'} + + ))} +
+ )} +
+
+
+ )) + )} + + )} +
+
+ ); +} + +export const MCPServerManager: React.FC = () => { + const [servers, setServers] = useState([]); + const [showAddForm, setShowAddForm] = useState(false); + const [form, setForm] = useState(EMPTY_FORM); + const [error, setError] = useState(null); + const [loadError, setLoadError] = useState(null); + const [selectedDetail, setSelectedDetail] = useState(null); + const [detailLoading, setDetailLoading] = useState(false); + + useEffect(() => { + loadServers(); + }, []); + + const loadServers = async () => { + setLoadError(null); + try { + const data = await apiClient.listMCPServers(); + setServers(data.servers || []); + } catch (err) { + const msg = err instanceof Error ? err.message : String(err); + console.error('Failed to load MCP servers:', err); + setLoadError(`Failed to load servers: ${msg}`); + } + }; + + const handleAddServer = async (e: React.FormEvent) => { + e.preventDefault(); + setError(null); + + try { + const payload: Parameters[0] = { + name: form.name.trim(), + transport: form.transport, + }; + + if (form.transport === 'stdio') { + payload.command = form.command.trim(); + payload.args = form.args.split(',').map((s) => s.trim()).filter(Boolean); + if (form.env.trim()) { + try { + payload.env = JSON.parse(form.env); + } catch { + setError('Environment variables must be valid JSON (e.g. {"KEY": "value"})'); + return; + } + } + } else { + payload.url = form.url.trim(); + if (form.headers.trim()) { + try { + payload.headers = JSON.parse(form.headers); + } catch { + setError('Headers must be valid JSON (e.g. {"Authorization": "Bearer token"})'); + return; + } + } + } + + await apiClient.registerMCPServer(payload); + setForm(EMPTY_FORM); + setShowAddForm(false); + await loadServers(); + } catch (err) { + console.error('Failed to register server:', err); + setError('Failed to register server. Check console for details.'); + } + }; + + const handleDelete = async (name: string) => { + if (!window.confirm(`Delete server "${name}"? This cannot be undone.`)) return; + try { + await apiClient.deleteMCPServer(name); + await loadServers(); + } catch (err) { + console.error('Failed to delete server:', err); + setLoadError(`Failed to delete "${name}". Please try again.`); + } + }; + + const loadDetail = useCallback(async (name: string) => { + setDetailLoading(true); + try { + const data = await apiClient.getMCPServerDetail(name); + setSelectedDetail(data); + } catch (err) { + console.error('Failed to load server detail:', err); + } finally { + setDetailLoading(false); + } + }, []); + + const inputClass = + 'w-full px-3 py-2 rounded border border-gray-300 dark:border-gray-600 ' + + 'bg-white dark:bg-gray-700 text-gray-900 dark:text-gray-100 ' + + 'focus:outline-none focus:ring-2 focus:ring-blue-500 dark:focus:ring-emerald-500 text-sm'; + + const serverSubtitle = (s: MCPServer) => { + if (s.transport === 'stdio') { + const cmd = [s.command, ...(s.args || [])].filter(Boolean).join(' '); + return cmd || 'stdio'; + } + return s.url || s.transport; + }; + + // ── Detail view ────────────────────────────────────────────────────────────── + if (selectedDetail) { + return ( + setSelectedDetail(null)} + onRefresh={() => loadDetail(selectedDetail.server.name)} + /> + ); + } + + return ( +
+
+

MCP Servers

+ +
+ + {showAddForm && ( +
+ setForm({ ...form, name: e.target.value })} + required + className={inputClass} + /> + +
+ +
+ {(['streamable-http', 'sse', 'stdio'] as Transport[]).map((t) => ( + + ))} +
+
+ + {form.transport !== 'stdio' ? ( + <> + setForm({ ...form, url: e.target.value })} + required + className={inputClass} + /> + setForm({ ...form, headers: e.target.value })} + className={inputClass} + /> + + ) : ( + <> + setForm({ ...form, command: e.target.value })} + required + className={inputClass} + /> + setForm({ ...form, args: e.target.value })} + className={inputClass} + /> + setForm({ ...form, env: e.target.value })} + className={inputClass} + /> + + )} + + {error && ( +

{error}

+ )} + +
+ + +
+
+ )} + + {loadError && ( +
+

{loadError}

+ +
+ )} + +
+ {servers.map((server) => ( +
loadDetail(server.name)} + className="flex items-start gap-3 p-3 bg-gray-50 dark:bg-gray-800 rounded-lg + cursor-pointer hover:bg-gray-100 dark:hover:bg-gray-700 transition-colors" + > + +
+
+

+ {server.name} +

+ {server.builtin && ( + + )} + + {TRANSPORT_LABELS[server.transport] ?? server.transport} + +
+

+ {serverSubtitle(server)} +

+ {server.tools_count !== undefined && server.tools_count > 0 && ( +

+ {server.tools_count} tool{server.tools_count !== 1 ? 's' : ''} +

+ )} +
+
+ + {!server.builtin && ( + + )} +
+
+ ))} + {servers.length === 0 && ( +

+ No MCP servers registered +

+ )} +
+
+ ); +}; diff --git a/src/components/ModelSettings.tsx b/src/components/ModelSettings.tsx index e80efdc..de7c478 100644 --- a/src/components/ModelSettings.tsx +++ b/src/components/ModelSettings.tsx @@ -111,13 +111,12 @@ function buildPayload(form: FormState, original: FormState): Record = { }, 'Verifying Sources': { icon: '🕵🏽', - color: 'text-grey-500', + color: 'text-gray-500', }, 'Generating Report': { icon: '📝', @@ -68,9 +68,14 @@ export const ProgressSpinner: React.FC = ({ status }) => { const config = STATUS_CONFIG[status] || STATUS_CONFIG['Researching']; return ( -
+
{/* Animated spinner ring */} -
+