Compare commits

...

7 Commits

Author SHA1 Message Date
bvandeusen f2c2117b25 Merge pull request 'feat: add News link to main navigation header' (#20) from dev into main 2026-04-06 22:20:27 +00:00
bvandeusen b9d0716b01 feat: add News link to main navigation header
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-06 17:49:22 -04:00
bvandeusen 9af8ab8f70 fix(briefing): use briefing context for follow-ups; add slot separator
- build_context: when conversation_type is 'briefing', inject a system
  prompt instruction telling the model to answer from conversation history
  and article context instead of searching the web
- Consolidate briefing conversation type detection to one DB query (was
  being checked twice — once for the system prompt addition, once for
  article context injection)
- ChatPanel: render a visual 'New Briefing Update' separator line before
  2nd+ briefing slot messages (identified by metadata.rss_item_ids)
- types/chat.ts: add metadata field to Message interface

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-06 06:15:18 -04:00
bvandeusen a171210224 feat(tools): confirmed guard for deletes, update_person/place, get/update_profile, calculate
- delete_note / delete_task: add confirmed parameter + requires_confirmation guard
  (find the note first, then ask, consistent with create_note/task pattern)
- get_note: description now mentions notes AND tasks
- update_person / update_place: new tools to update existing entity notes in-place
- get_profile / update_profile: surface and edit the user's stored profile
  (expertise, tone, response style, job title, interests)
- calculate: eval math expressions via Python math module; solves precision issues
  on multi-step arithmetic and supports sqrt/log/trig/etc.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-05 22:58:14 -04:00
bvandeusen eb92b2a976 feat(research): multi-note pipeline — outline + parallel section synthesis + index note
Replaces the single monolithic research note with topic-driven section notes
plus an index note. Two new LLM calls: _generate_outline (JSON outline, 3-8
sections) and _synthesize_section (300-600 word focused note per section,
parallelised via asyncio.gather). Public signature of run_research_pipeline
unchanged; falls back to single-note synthesis on outline failure or if all
sections fail.

Also extracts _build_sources_block helper and adds full test suite.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-05 22:53:14 -04:00
bvandeusen be805073a7 docs: add research multi-note redesign spec 2026-04-05 22:42:51 -04:00
bvandeusen e4c812a603 feat(voice): improve TTS logging for root-cause diagnosis
- Route now logs every synthesis request (char count, voice, speed)
- Route logs char count + text preview when the 8000-char limit is hit
- Route logs empty audio with preview (helps spot no-chunk-produced edge case)
- Route logs success with byte count and duration
- Kokoro synthesise() logs per-call: samples produced, elapsed, chars/s
- Kokoro synthesise() logs warning when zero audio chunks returned with preview
- Kokoro synthesise() catches and logs pipeline-internal errors with preview
- Frontend: console.warn now includes char count + 80-char preview on failure and retry

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-05 22:36:43 -04:00
11 changed files with 929 additions and 62 deletions
@@ -0,0 +1,162 @@
# Research Pipeline — Multi-Note Redesign
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** Replace the single monolithic research note with a set of focused, topic-driven notes plus an index note that links them — making research output browsable, TTS-friendly, and well-organized.
**Architecture:** Two new LLM calls (outline generation + N parallel section syntheses) replace the single large synthesis call. Public API unchanged — callers receive the index note. Fallback to single-note behavior on any outline failure.
**Tech Stack:** Python/Quart backend, existing `research.py` service, asyncio.gather for parallelism.
---
## Problem
The current pipeline synthesizes one note with a minimum of 2500 words and 6 sections. This creates:
- Notes too large to read or listen to comfortably
- No way to navigate directly to a specific sub-topic
- TTS failures on long prose (8000-char route limit, unbounded sentence buffers)
---
## Pipeline Flow
Public signature unchanged:
```python
async def run_research_pipeline(
topic: str,
user_id: int,
model: str,
buf=None,
project_id: int | None = None,
) -> Note: # returns the index note
```
Execution order:
```
1. Generate sub-queries (unchanged)
2. Search + fetch sources (unchanged)
3. Generate topic outline (NEW — one LLM call → 37 section dicts)
4. Synthesize each section note (NEW — parallelized via asyncio.gather)
5. Create all section notes in DB (sequential, tagged ["research"], same project_id)
6. Create index note (NEW — links all sections)
7. Return index note
```
Status messages via `buf.append_event("status", ...)`:
- `"Generating outline…"`
- `"Writing: [Section Title]…"` (one per section, emitted before synthesis starts)
- `"Saving [N] notes…"`
No note content is streamed into chat. After the tool call resolves, the LLM writes a brief conversational summary citing the index note title and section count.
---
## Outline Generation
New function: `_generate_outline(topic, sources, model) -> list[dict]`
Sends all fetched sources to the model with a prompt requesting a JSON array:
```json
[
{"title": "Quantum Entanglement: Mechanisms", "focus": "How entanglement works at the physical level"},
{"title": "Quantum Computing Hardware", "focus": "Ion traps, superconducting qubits, photonic approaches"}
]
```
**Prompt requirements:**
- Produce 37 sections covering distinct aspects of the topic
- Titles must work as standalone note titles (no "Overview" or "Introduction" generics)
- No overlap between sections
- `focus` is one sentence describing what this section should specifically cover
**Guardrails:**
- Fewer than 3 sections parsed → fall back to single-note synthesis
- JSON parse failure → fall back to single-note synthesis
- More than 8 sections → truncate to 8
**Model params:** `max_tokens=400, num_ctx=16384` (outline is short)
---
## Section Synthesis
New function: `_synthesize_section(section_title, section_focus, sources, model) -> tuple[str, str]`
Returns `(title, body_markdown)`.
All sections receive all fetched sources. The `section_focus` field in the prompt directs the model to draw only what's relevant to that section's scope.
**Prompt requirements:**
- 300600 words of substantive prose
- Do NOT include a `# Title` heading (title is set separately)
- End with a brief `## Sources` list of relevant URLs from the provided sources
- Focus strictly on `section_focus` — ignore source material outside that scope
**Model params:** `num_predict=2048, num_ctx=16384` (reduced from 8192 — sufficient for 600 words, prevents rambling)
**Parallelism:** All section synthesis calls run via `asyncio.gather`. Wall-clock time stays close to a single synthesis call despite producing N notes.
---
## Note Creation and Index Note
**Section notes:**
- Tags: `["research"]`
- `project_id`: same as passed to pipeline (or None)
- Title: from outline `title` field
- Created sequentially (avoids DB contention)
**Index note:**
- Tags: `["research", "research-index"]`
- `project_id`: same as section notes
- Title: `"Research: [topic]"`
- Created last (after all section notes exist)
**Index note body format:**
```markdown
Research overview for **[topic]** — [YYYY-MM-DD]
Generated from [N] web sources across [M] sections.
## Sections
- **[Section 1 Title]** — [focus sentence]
- **[Section 2 Title]** — [focus sentence]
...
*Search for any section title to read it.*
```
The index note is what `run_research_pipeline` returns. The existing `research_topic` tool handler uses `note.id` and `note.title` — both remain valid with the index note.
---
## Error Handling
| Scenario | Behaviour |
|---|---|
| Outline generation raises | Fall back to single-note synthesis (current behaviour) |
| Outline JSON unparseable | Fall back to single-note synthesis |
| Outline returns < 3 sections | Fall back to single-note synthesis |
| Outline returns > 8 sections | Truncate to 8, continue |
| A section synthesis raises | Log warning, skip that section; continue with remaining |
| All section syntheses fail | Fall back to single-note synthesis |
| A section note DB save fails | Log warning, skip from index; index note still created |
| No sources fetched | Raise `ValueError` as today — unchanged |
The fallback in every case is the current single-note pipeline. Research never silently produces nothing.
---
## What Is NOT Changing
- Public function signature of `run_research_pipeline`
- Sub-query generation (`_generate_sub_queries`)
- SearXNG search and URL fetching
- `_search_searxng`, `_search_searxng_images`, `fetch_url_content`
- The `research_topic` tool definition and handler in `tools.py`
- The `quick_capture` research path
- Any frontend component
+1
View File
@@ -77,6 +77,7 @@ router.afterEach(() => {
<router-link to="/chat" :class="['nav-link', { 'router-link-active': isChatActive }]">Chat</router-link> <router-link to="/chat" :class="['nav-link', { 'router-link-active': isChatActive }]">Chat</router-link>
<router-link to="/briefing" class="nav-link">Briefing</router-link> <router-link to="/briefing" class="nav-link">Briefing</router-link>
<router-link to="/calendar" class="nav-link">Calendar</router-link> <router-link to="/calendar" class="nav-link">Calendar</router-link>
<router-link to="/news" class="nav-link">News</router-link>
<router-link to="/tasks" class="nav-link">Tasks</router-link> <router-link to="/tasks" class="nav-link">Tasks</router-link>
<router-link to="/projects" class="nav-link">Projects</router-link> <router-link to="/projects" class="nav-link">Projects</router-link>
</div> </div>
+37 -5
View File
@@ -299,12 +299,22 @@ defineExpose({ focus, prefill, send })
<!-- Message list --> <!-- Message list -->
<div ref="messagesEl" class="messages-container"> <div ref="messagesEl" class="messages-container">
<div class="messages-inner"> <div class="messages-inner">
<ChatMessage <template
v-for="msg in store.currentConversation?.messages ?? []" v-for="(msg, index) in store.currentConversation?.messages ?? []"
:key="msg.id" :key="msg.id"
:message="msg" >
@save-as-note="handleSaveAsNote" <!-- Briefing slot separator: shown before a 2nd+ briefing slot message -->
/> <div
v-if="briefingMode && index > 0 && msg.role === 'assistant' && msg.metadata?.rss_item_ids"
class="briefing-slot-separator"
>
<span>New Briefing Update</span>
</div>
<ChatMessage
:message="msg"
@save-as-note="handleSaveAsNote"
/>
</template>
<!-- Streaming bubble --> <!-- Streaming bubble -->
<ChatStreamingBubble v-if="store.streaming" /> <ChatStreamingBubble v-if="store.streaming" />
<!-- Queued messages --> <!-- Queued messages -->
@@ -523,6 +533,28 @@ defineExpose({ focus, prefill, send })
justify-content: flex-end; justify-content: flex-end;
} }
/* Briefing slot separator */
.briefing-slot-separator {
display: flex;
align-items: center;
gap: 0.75rem;
margin: 1.25rem 0 0.5rem;
color: var(--color-text-muted, #888);
font-size: 0.7rem;
font-weight: 600;
letter-spacing: 0.08em;
text-transform: uppercase;
opacity: 0.7;
}
.briefing-slot-separator::before,
.briefing-slot-separator::after {
content: '';
flex: 1;
height: 1px;
background: var(--color-border, #333);
opacity: 0.5;
}
/* Context sidebar */ /* Context sidebar */
.context-sidebar { .context-sidebar {
width: 200px; width: 200px;
+12 -2
View File
@@ -86,11 +86,21 @@ export function useStreamingTts(options: UseStreamingTtsOptions): UseStreamingTt
try { try {
blob = await synthesiseSpeech(stripped) blob = await synthesiseSpeech(stripped)
} catch (e) { } catch (e) {
console.warn('[StreamingTTS] Synthesis failed, retrying sentence', { sentence: stripped, error: e }) const errMsg = e instanceof Error ? e.message : String(e)
console.warn('[StreamingTTS] Synthesis failed, retrying', {
chars: stripped.length,
preview: stripped.slice(0, 80),
error: errMsg,
})
try { try {
blob = await synthesiseSpeech(stripped) blob = await synthesiseSpeech(stripped)
} catch (e2) { } catch (e2) {
console.warn('[StreamingTTS] Retry also failed, skipping sentence', { sentence: stripped, error: e2 }) const errMsg2 = e2 instanceof Error ? e2.message : String(e2)
console.warn('[StreamingTTS] Retry failed, sentence dropped', {
chars: stripped.length,
preview: stripped.slice(0, 80),
error: errMsg2,
})
} }
} finally { } finally {
pendingCount.value-- pendingCount.value--
+1
View File
@@ -28,6 +28,7 @@ export interface Message {
context_note_id: number | null; context_note_id: number | null;
context_note_title?: string | null; context_note_title?: string | null;
tool_calls?: ToolCallRecord[] | null; tool_calls?: ToolCallRecord[] | null;
metadata?: Record<string, unknown> | null;
created_at: string; created_at: string;
timing?: GenerationTiming; timing?: GenerationTiming;
thinking?: string; thinking?: string;
+25 -2
View File
@@ -117,7 +117,12 @@ async def synthesise_speech():
if not text: if not text:
return jsonify({"error": "text is required"}), 400 return jsonify({"error": "text is required"}), 400
if len(text) > 8000: char_count = len(text)
if char_count > 8000:
logger.warning(
"TTS request rejected: text too long (%d chars, limit 8000). Preview: %r",
char_count, text[:120],
)
return jsonify({"error": "text too long (max 8000 characters)"}), 400 return jsonify({"error": "text too long (max 8000 characters)"}), 400
voice = str(data.get("voice", "af_heart")) voice = str(data.get("voice", "af_heart"))
@@ -154,11 +159,29 @@ async def synthesise_speech():
except Exception: except Exception:
pass pass
blend_desc = f"blend({len(voice_blend)} voices)" if voice_blend else voice
logger.info("TTS synthesis start: %d chars, voice=%s, speed=%.2f", char_count, blend_desc, speed)
t0 = time.monotonic()
try: try:
wav_bytes = await synthesise(text, voice=voice, speed=speed, voice_blend=voice_blend) wav_bytes = await synthesise(text, voice=voice, speed=speed, voice_blend=voice_blend)
except Exception: except Exception:
logger.exception("TTS synthesis failed") logger.exception(
"TTS synthesis failed: %d chars, voice=%s. Preview: %r",
char_count, blend_desc, text[:120],
)
return jsonify({"error": "Synthesis failed"}), 500 return jsonify({"error": "Synthesis failed"}), 500
duration_ms = round((time.monotonic() - t0) * 1000)
if not wav_bytes:
logger.warning(
"TTS synthesis returned empty audio: %d chars, voice=%s, %dms. Preview: %r",
char_count, blend_desc, duration_ms, text[:120],
)
else:
logger.info(
"TTS synthesis complete: %d chars → %d bytes in %dms (voice=%s)",
char_count, len(wav_bytes), duration_ms, blend_desc,
)
from quart import Response from quart import Response
return Response(wav_bytes, mimetype="audio/wav") return Response(wav_bytes, mimetype="audio/wav")
+23 -10
View File
@@ -607,6 +607,25 @@ async def build_context(
f"\n\n--- Earlier Conversation ---\n{history_summary}\n--- End Earlier Conversation ---" f"\n\n--- Earlier Conversation ---\n{history_summary}\n--- End Earlier Conversation ---"
) )
# Detect briefing conversation — used for both system prompt instruction and article injection
_is_briefing_conv = False
if conv_id is not None:
from fabledassistant.models import async_session as _async_session
from fabledassistant.models.conversation import Conversation as _Conversation
async with _async_session() as _sess:
_conv = await _sess.get(_Conversation, conv_id)
if _conv and getattr(_conv, "conversation_type", None) == "briefing":
_is_briefing_conv = True
if _is_briefing_conv:
system_content += (
"\n\nYou are in a briefing conversation. "
"The conversation history contains today's briefing — news stories, weather, and tasks. "
"When the user asks about a topic, person, or event from the briefing, answer directly "
"from the conversation history and the article context that follows. "
"Do NOT search the web for information that is already present in the briefing."
)
context_meta: dict = { context_meta: dict = {
"context_note_id": None, "context_note_id": None,
"context_note_title": None, "context_note_title": None,
@@ -769,16 +788,10 @@ async def build_context(
) )
# Briefing article context for follow-up Q&A # Briefing article context for follow-up Q&A
if conv_id is not None: if _is_briefing_conv:
from fabledassistant.models import async_session as _async_session article_context = await _build_briefing_article_context(conv_id) # type: ignore[arg-type]
from fabledassistant.models.conversation import Conversation if article_context:
user_context_parts.append(article_context.strip())
async with _async_session() as _sess:
_conv = await _sess.get(Conversation, conv_id)
if _conv and getattr(_conv, "conversation_type", None) == "briefing":
article_context = await _build_briefing_article_context(conv_id)
if article_context:
user_context_parts.append(article_context.strip())
# Build final user message — context prefix (if any) followed by the actual message # Build final user message — context prefix (if any) followed by the actual message
if user_context_parts: if user_context_parts:
+188 -35
View File
@@ -21,6 +21,106 @@ MAX_SYNTHESIS_SOURCES = 12 # deduplicated sources passed to synthesis LLM
CHARS_PER_SOURCE = 2000 # content chars per source sent to synthesis CHARS_PER_SOURCE = 2000 # content chars per source sent to synthesis
def _build_sources_block(sources: list[dict]) -> str:
"""Format fetched sources into a text block for LLM prompts."""
parts = []
for i, s in enumerate(sources, 1):
content = (s.get("content") or s.get("snippet") or "")[:CHARS_PER_SOURCE]
parts.append(
f"[Source {i}] {s['title']}\nURL: {s['url']}\nSearch query: {s['query']}\n\n{content}"
)
return "\n\n" + ("" * 60) + "\n\n".join(parts)
async def _generate_outline(topic: str, sources: list[dict], model: str) -> list[dict]:
"""Generate a topic outline from fetched research sources.
Returns a list of {"title": str, "focus": str} dicts (38 entries).
Returns [] on failure — callers must fall back to single-note synthesis.
"""
import json as _json
sources_block = _build_sources_block(sources) if sources else "(no sources)"
messages = [
{
"role": "system",
"content": (
"You are a research organizer. Given research sources on a topic, produce a JSON array "
"of section objects that together cover the topic comprehensively from distinct angles.\n\n"
"Rules:\n"
"- Return exactly 37 sections\n"
"- Each section must cover a unique angle — no overlap between sections\n"
"- Titles must work as standalone note titles (specific, not generic like 'Overview')\n"
"- focus: one sentence describing exactly what this section covers\n"
"- Respond with ONLY a JSON array, no other text\n\n"
'Example: [{"title": "CRISPR: Molecular Mechanisms", "focus": "How Cas9 identifies and cuts DNA at guide-RNA-specified sites"}]'
),
},
{
"role": "user",
"content": f"Topic: {topic}\n\nSources:\n{sources_block}",
},
]
try:
raw = await generate_completion(messages, model, max_tokens=400)
raw = raw.strip()
raw = re.sub(r"^```(?:json)?\s*", "", raw)
raw = re.sub(r"\s*```$", "", raw)
idx = raw.find("[")
if idx >= 0:
parsed, _ = _json.JSONDecoder().raw_decode(raw[idx:])
if isinstance(parsed, list):
sections = [
s for s in parsed
if isinstance(s, dict) and s.get("title") and s.get("focus")
]
if len(sections) >= 3:
return sections[:8]
except Exception:
logger.warning("Outline generation failed for topic '%s'", topic, exc_info=True)
return []
async def _synthesize_section(
section_title: str,
section_focus: str,
sources: list[dict],
model: str,
) -> tuple[str, str]:
"""Synthesize one focused note section.
Returns (section_title, body_markdown). Does not stream.
"""
sources_block = _build_sources_block(sources) if sources else "(no sources provided)"
messages = [
{
"role": "system",
"content": (
"You are a focused research writer. Write a single well-structured note section "
"on the specific topic provided.\n\n"
"Requirements:\n"
f"- Focus strictly on: {section_focus}\n"
"- 300600 words of substantive prose\n"
"- Use ### for subsections only when they genuinely aid clarity\n"
"- Do NOT include a top-level # heading — the title is set separately\n"
"- Write in detailed prose paragraphs — not bullet points\n"
"- End with a '## Sources' section listing relevant source URLs as markdown hyperlinks\n"
"- Ignore source material that falls outside your assigned focus"
),
},
{
"role": "user",
"content": (
f"Section title: {section_title}\n"
f"Focus: {section_focus}\n\n"
f"Sources:\n{sources_block}"
),
},
]
raw = await generate_completion(messages, model, max_tokens=2048, num_ctx=16384)
return section_title, raw.strip()
async def run_research_pipeline( async def run_research_pipeline(
topic: str, topic: str,
user_id: int, user_id: int,
@@ -28,14 +128,17 @@ async def run_research_pipeline(
buf=None, buf=None,
project_id: int | None = None, project_id: int | None = None,
) -> Note: ) -> Note:
"""Full research pipeline: search → fetch → synthesize → create note. """Full research pipeline: search → fetch → outline → section notes → index note.
Emits status events via buf.append_event throughout (when buf is provided). Emits status events via buf throughout (when buf is provided).
Returns the created Note. Returns the index note (or a single fallback note on outline failure).
""" """
def _status(msg: str) -> None:
if buf is not None:
buf.append_event("status", {"status": msg})
# Step 1: Generate sub-queries # Step 1: Generate sub-queries
if buf is not None: _status("Generating search queries...")
buf.append_event("status", {"status": "Generating search queries..."})
queries = await _generate_sub_queries(topic, model) queries = await _generate_sub_queries(topic, model)
logger.info("Research: generated %d sub-queries for topic '%s'", len(queries), topic) logger.info("Research: generated %d sub-queries for topic '%s'", len(queries), topic)
@@ -43,8 +146,7 @@ async def run_research_pipeline(
async def _search_with_stagger(i: int, query: str) -> tuple[str, list[dict]]: async def _search_with_stagger(i: int, query: str) -> tuple[str, list[dict]]:
if i > 0: if i > 0:
await asyncio.sleep(0.2 * i) await asyncio.sleep(0.2 * i)
if buf is not None: _status(f"Searching: {query}...")
buf.append_event("status", {"status": f"Searching: {query}..."})
results = await _search_searxng(query) results = await _search_searxng(query)
logger.info("Research: query '%s'%d results", query, len(results)) logger.info("Research: query '%s'%d results", query, len(results))
return query, results return query, results
@@ -66,8 +168,7 @@ async def run_research_pipeline(
# Fetch all unique URLs in parallel # Fetch all unique URLs in parallel
async def _fetch_source(url: str, result: dict, query: str) -> dict: async def _fetch_source(url: str, result: dict, query: str) -> dict:
title = result.get("title", url) title = result.get("title", url)
if buf is not None: _status(f"Reading: {title[:60]}...")
buf.append_event("status", {"status": f"Reading: {title[:60]}..."})
content = await fetch_url_content(url) content = await fetch_url_content(url)
return { return {
"url": url, "url": url,
@@ -84,39 +185,97 @@ async def run_research_pipeline(
if not all_sources: if not all_sources:
raise ValueError(f"No results found for '{topic}'") raise ValueError(f"No results found for '{topic}'")
# Step 3: Filter failed fetches good_sources = [s for s in all_sources if not s["content"].startswith("[Failed to fetch")]
good_sources = [
s for s in all_sources
if not s["content"].startswith("[Failed to fetch")
]
if not good_sources: if not good_sources:
raise ValueError(f"Could not read any sources for '{topic}'") raise ValueError(f"Could not read any sources for '{topic}'")
# Limit to top N sources for synthesis (already deduplicated by URL)
synthesis_sources = good_sources[:MAX_SYNTHESIS_SOURCES] synthesis_sources = good_sources[:MAX_SYNTHESIS_SOURCES]
logger.info( logger.info(
"Research: %d/%d sources successfully fetched, using %d for synthesis", "Research: %d/%d sources successfully fetched, using %d for synthesis",
len(good_sources), len(all_sources), len(synthesis_sources), len(good_sources), len(all_sources), len(synthesis_sources),
) )
# Step 4: Synthesize (streams tokens into chat as the note is being written) # Step 3: Generate topic outline
if buf is not None: _status("Generating outline...")
buf.append_event("status", {"status": f"Synthesizing report from {len(synthesis_sources)} sources..."}) outline = await _generate_outline(topic, synthesis_sources, model)
title, body = await _synthesize_note(topic, synthesis_sources, model, buf)
# Step 5: Create note # Fallback: outline failed or too short → single monolithic note
if buf is not None: if not outline:
buf.append_event("status", {"status": "Saving note..."}) logger.warning("Research outline empty, falling back to single note for '%s'", topic)
note = await create_note( _status("Synthesizing report...")
title, body = await _synthesize_note(topic, synthesis_sources, model, buf=None)
note = await create_note(
user_id=user_id, title=title, body=body, tags=["research"], project_id=project_id,
)
logger.info("Research (fallback): created note id=%d title='%s'", note.id, note.title)
return note
# Step 4: Synthesize each section in parallel
for section in outline:
_status(f"Writing: {section['title']}...")
raw_results = await asyncio.gather(
*[_synthesize_section(s["title"], s["focus"], synthesis_sources, model) for s in outline],
return_exceptions=True,
)
# Step 5: Create section notes sequentially
_status(f"Saving {len(outline)} notes...")
section_note_pairs: list[tuple[dict, Note]] = []
for section, result in zip(outline, raw_results):
if isinstance(result, Exception):
logger.warning("Section synthesis failed for '%s': %s", section["title"], result)
continue
sec_title, sec_body = result
try:
note = await create_note(
user_id=user_id,
title=sec_title,
body=sec_body,
tags=["research"],
project_id=project_id,
)
section_note_pairs.append((section, note))
except Exception:
logger.warning("Failed to save section note '%s'", sec_title, exc_info=True)
# All sections failed — fall back to single note
if not section_note_pairs:
logger.warning("All section syntheses failed, falling back to single note for '%s'", topic)
_status("Synthesizing report (fallback)...")
title, body = await _synthesize_note(topic, synthesis_sources, model, buf=None)
note = await create_note(
user_id=user_id, title=title, body=body, tags=["research"], project_id=project_id,
)
return note
# Step 6: Create index note
from datetime import date as _date
index_lines = [
f"Research overview for **{topic}** — {_date.today().isoformat()}",
"",
f"Generated from {len(synthesis_sources)} web sources across {len(section_note_pairs)} sections.",
"",
"## Sections",
"",
]
for section, note in section_note_pairs:
index_lines.append(f"- **{note.title}** — {section['focus']}")
index_lines += ["", "*Search for any section title to read it.*"]
index_note = await create_note(
user_id=user_id, user_id=user_id,
title=title, title=f"Research: {topic}",
body=body, body="\n".join(index_lines),
tags=["research"], tags=["research", "research-index"],
project_id=project_id, project_id=project_id,
) )
logger.info("Research: created note id=%d title='%s'", note.id, note.title) logger.info(
return note "Research: created %d section notes + index id=%d for topic '%s'",
len(section_note_pairs), index_note.id, topic,
)
return index_note
async def _generate_sub_queries(topic: str, model: str) -> list[str]: async def _generate_sub_queries(topic: str, model: str) -> list[str]:
@@ -248,13 +407,7 @@ async def _synthesize_note(
When buf is provided, tokens are streamed into the chat buffer in real time When buf is provided, tokens are streamed into the chat buffer in real time
so the user can see the note being written. Uses an extended context window. so the user can see the note being written. Uses an extended context window.
""" """
sources_text_parts = [] sources_block = _build_sources_block(sources)
for i, s in enumerate(sources, 1):
content = (s.get("content") or s.get("snippet") or "")[:CHARS_PER_SOURCE]
sources_text_parts.append(
f"[Source {i}] {s['title']}\nURL: {s['url']}\nSearch query: {s['query']}\n\n{content}"
)
sources_block = "\n\n" + ("" * 60) + "\n\n".join(sources_text_parts)
messages = [ messages = [
{ {
+275 -4
View File
@@ -258,8 +258,8 @@ _CORE_TOOLS = [
"function": { "function": {
"name": "get_note", "name": "get_note",
"description": ( "description": (
"Retrieve the full content of a specific note. Use this when the user asks to read, " "Retrieve the full content of a specific note or task. Use this when the user asks to read, "
"view, or check what a particular note says. Returns the complete note body." "view, or check what a particular note or task says. Returns the complete body."
), ),
"parameters": { "parameters": {
"type": "object", "type": "object",
@@ -316,7 +316,7 @@ _CORE_TOOLS = [
"name": "delete_note", "name": "delete_note",
"description": ( "description": (
"Delete a note permanently. Use ONLY when the user explicitly asks to delete or remove a note. " "Delete a note permanently. Use ONLY when the user explicitly asks to delete or remove a note. "
"This action requires user confirmation and cannot be undone." "Always confirm with the user first — this cannot be undone."
), ),
"parameters": { "parameters": {
"type": "object", "type": "object",
@@ -325,6 +325,10 @@ _CORE_TOOLS = [
"type": "string", "type": "string",
"description": "Title or keyword to find the note to delete", "description": "Title or keyword to find the note to delete",
}, },
"confirmed": {
"type": "boolean",
"description": "Must be true — only set after the user has explicitly confirmed they want this note deleted.",
},
}, },
"required": ["query"], "required": ["query"],
}, },
@@ -336,7 +340,7 @@ _CORE_TOOLS = [
"name": "delete_task", "name": "delete_task",
"description": ( "description": (
"Delete a task permanently. Use ONLY when the user explicitly asks to delete or remove a task. " "Delete a task permanently. Use ONLY when the user explicitly asks to delete or remove a task. "
"This action requires user confirmation and cannot be undone." "Always confirm with the user first — this cannot be undone."
), ),
"parameters": { "parameters": {
"type": "object", "type": "object",
@@ -345,6 +349,10 @@ _CORE_TOOLS = [
"type": "string", "type": "string",
"description": "Title or keyword to find the task to delete", "description": "Title or keyword to find the task to delete",
}, },
"confirmed": {
"type": "boolean",
"description": "Must be true — only set after the user has explicitly confirmed they want this task deleted.",
},
}, },
"required": ["query"], "required": ["query"],
}, },
@@ -937,6 +945,28 @@ _RAG_TOOLS = [
}, },
}, },
}, },
{
"type": "function",
"function": {
"name": "calculate",
"description": (
"Evaluate a mathematical expression and return the exact result. Use this for any "
"arithmetic, percentages, unit conversions, or multi-step calculations where precision "
"matters. Supports standard math operators (+, -, *, /, **, %) and all Python math "
"module functions (sqrt, log, sin, cos, floor, ceil, etc.)."
),
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "A valid Python math expression (e.g. '(450 * 1.13) / 12', 'sqrt(144)', 'log(1000, 10)')",
},
},
"required": ["expression"],
},
},
},
] ]
_ENTITY_TOOLS = [ _ENTITY_TOOLS = [
@@ -964,6 +994,29 @@ _ENTITY_TOOLS = [
}, },
}, },
}, },
{
"type": "function",
"function": {
"name": "update_person",
"description": (
"Update details about a saved person. Use this when the user corrects or adds new "
"information about someone already in the knowledge base."
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Name or keyword to find the person"},
"relationship": {"type": "string", "description": "Updated relationship to the user"},
"phone": {"type": "string", "description": "Updated phone number"},
"email": {"type": "string", "description": "Updated email address"},
"birthday": {"type": "string", "description": "Updated birthday in YYYY-MM-DD format"},
"address": {"type": "string", "description": "Updated home or mailing address"},
"notes": {"type": "string", "description": "Updated free-form notes about this person"},
},
"required": ["query"],
},
},
},
{ {
"type": "function", "type": "function",
"function": { "function": {
@@ -987,6 +1040,28 @@ _ENTITY_TOOLS = [
}, },
}, },
}, },
{
"type": "function",
"function": {
"name": "update_place",
"description": (
"Update details about a saved place. Use this when the user corrects or adds new "
"information about a location already in the knowledge base."
),
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "Name or keyword to find the place"},
"address": {"type": "string", "description": "Updated street address"},
"phone": {"type": "string", "description": "Updated phone number"},
"hours": {"type": "string", "description": "Updated opening hours"},
"url": {"type": "string", "description": "Updated website URL"},
"notes": {"type": "string", "description": "Updated free-form notes about this place"},
},
"required": ["query"],
},
},
},
{ {
"type": "function", "type": "function",
"function": { "function": {
@@ -1051,12 +1126,71 @@ _ENTITY_TOOLS = [
] ]
_PROFILE_TOOLS = [
{
"type": "function",
"function": {
"name": "get_profile",
"description": (
"Retrieve the user's stored profile: name, job title, industry, expertise level, "
"preferred response style, tone, and interests. Use this when you need to personalise "
"a response and the user's profile hasn't already been injected into context."
),
"parameters": {
"type": "object",
"properties": {},
},
},
},
{
"type": "function",
"function": {
"name": "update_profile",
"description": (
"Update the user's stored profile when they share personal information: their name, "
"job, industry, expertise, preferred response style, tone, or interests. "
"Only set fields the user has explicitly mentioned."
),
"parameters": {
"type": "object",
"properties": {
"display_name": {"type": "string", "description": "User's preferred display name"},
"job_title": {"type": "string", "description": "Job title (e.g. 'Software Engineer')"},
"industry": {"type": "string", "description": "Industry (e.g. 'Healthcare', 'Finance')"},
"expertise_level": {
"type": "string",
"enum": ["novice", "intermediate", "expert"],
"description": "User's general expertise level",
},
"response_style": {
"type": "string",
"enum": ["concise", "balanced", "detailed"],
"description": "Preferred response length/depth",
},
"tone": {
"type": "string",
"enum": ["casual", "professional", "technical"],
"description": "Preferred communication tone",
},
"interests": {
"type": "array",
"items": {"type": "string"},
"description": "List of topics or hobbies the user is interested in",
},
},
},
},
},
]
async def get_tools_for_user(user_id: int) -> list[dict]: async def get_tools_for_user(user_id: int) -> list[dict]:
"""Build the tool list for a user based on their configured integrations.""" """Build the tool list for a user based on their configured integrations."""
tools = list(_CORE_TOOLS) tools = list(_CORE_TOOLS)
tools.extend(_URL_TOOLS) tools.extend(_URL_TOOLS)
tools.extend(_RAG_TOOLS) tools.extend(_RAG_TOOLS)
tools.extend(_ENTITY_TOOLS) tools.extend(_ENTITY_TOOLS)
tools.extend(_PROFILE_TOOLS)
if await is_caldav_configured(user_id): if await is_caldav_configured(user_id):
tools.extend(_CALDAV_TOOLS) tools.extend(_CALDAV_TOOLS)
if Config.searxng_enabled(): if Config.searxng_enabled():
@@ -1698,6 +1832,12 @@ async def execute_tool(
note = notes[0] note = notes[0]
if note.status is not None: if note.status is not None:
return {"success": False, "error": f"'{note.title}' is a task. Use delete_task instead."} return {"success": False, "error": f"'{note.title}' is a task. Use delete_task instead."}
if not arguments.get("confirmed"):
return {
"success": False,
"requires_confirmation": True,
"error": f"Deleting '{note.title}' is permanent. Ask the user to confirm, then retry with confirmed=true.",
}
deleted = await delete_note(user_id, note.id) deleted = await delete_note(user_id, note.id)
if not deleted: if not deleted:
return {"success": False, "error": "Failed to delete note."} return {"success": False, "error": "Failed to delete note."}
@@ -1717,6 +1857,12 @@ async def execute_tool(
note = notes[0] note = notes[0]
if note.status is None: if note.status is None:
return {"success": False, "error": f"'{note.title}' is a note. Use delete_note instead."} return {"success": False, "error": f"'{note.title}' is a note. Use delete_note instead."}
if not arguments.get("confirmed"):
return {
"success": False,
"requires_confirmation": True,
"error": f"Deleting '{note.title}' is permanent. Ask the user to confirm, then retry with confirmed=true.",
}
deleted = await delete_note(user_id, note.id) deleted = await delete_note(user_id, note.id)
if not deleted: if not deleted:
return {"success": False, "error": "Failed to delete task."} return {"success": False, "error": "Failed to delete task."}
@@ -2172,6 +2318,72 @@ async def execute_tool(
_schedule_embedding(note.id, user_id, name, note.body) _schedule_embedding(note.id, user_id, name, note.body)
return {"success": True, "type": "place", "data": {"id": note.id, "name": name}} return {"success": True, "type": "place", "data": {"id": note.id, "name": name}}
elif tool_name == "update_person":
query = str(arguments.get("query", "")).strip()
if not query:
return {"success": False, "error": "query is required"}
existing, _ = await list_notes(user_id=user_id, q=query, is_task=False, limit=10)
target = next((n for n in existing if n.note_type == "person"), None)
if target is None:
return {"success": False, "error": f"No person found matching '{query}'. Use create_person to add them."}
meta = dict(target.entity_meta or {})
for field in ("relationship", "phone", "email", "birthday", "address"):
val = arguments.get(field)
if val is not None:
meta[field] = str(val)
body_lines = []
if meta.get("relationship"):
body_lines.append(f"**Relationship:** {meta['relationship']}")
if meta.get("phone"):
body_lines.append(f"**Phone:** {meta['phone']}")
if meta.get("email"):
body_lines.append(f"**Email:** {meta['email']}")
if meta.get("birthday"):
body_lines.append(f"**Birthday:** {meta['birthday']}")
if meta.get("address"):
body_lines.append(f"**Address:** {meta['address']}")
extra_notes = arguments.get("notes")
if extra_notes is not None:
body_lines.append(f"\n{extra_notes}")
new_body = "\n".join(body_lines)
updated = await update_note(user_id=user_id, note_id=target.id, body=new_body, entity_meta=meta)
if updated is None:
return {"success": False, "error": "Failed to update person."}
_schedule_embedding(target.id, user_id, target.title, new_body)
return {"success": True, "type": "person_updated", "data": {"id": target.id, "name": target.title}}
elif tool_name == "update_place":
query = str(arguments.get("query", "")).strip()
if not query:
return {"success": False, "error": "query is required"}
existing, _ = await list_notes(user_id=user_id, q=query, is_task=False, limit=10)
target = next((n for n in existing if n.note_type == "place"), None)
if target is None:
return {"success": False, "error": f"No place found matching '{query}'. Use create_place to add it."}
meta = dict(target.entity_meta or {})
for field in ("address", "phone", "hours", "url"):
val = arguments.get(field)
if val is not None:
meta[field] = str(val)
body_lines = []
if meta.get("address"):
body_lines.append(f"**Address:** {meta['address']}")
if meta.get("phone"):
body_lines.append(f"**Phone:** {meta['phone']}")
if meta.get("hours"):
body_lines.append(f"**Hours:** {meta['hours']}")
if meta.get("url"):
body_lines.append(f"**Website:** {meta['url']}")
extra_notes = arguments.get("notes")
if extra_notes is not None:
body_lines.append(f"\n{extra_notes}")
new_body = "\n".join(body_lines)
updated = await update_note(user_id=user_id, note_id=target.id, body=new_body, entity_meta=meta)
if updated is None:
return {"success": False, "error": "Failed to update place."}
_schedule_embedding(target.id, user_id, target.title, new_body)
return {"success": True, "type": "place_updated", "data": {"id": target.id, "name": target.title}}
elif tool_name == "create_list": elif tool_name == "create_list":
name = str(arguments.get("name", "")).strip() name = str(arguments.get("name", "")).strip()
if not name: if not name:
@@ -2235,6 +2447,65 @@ async def execute_tool(
_schedule_embedding(target.id, user_id, target.title, cleaned) _schedule_embedding(target.id, user_id, target.title, cleaned)
return {"success": True, "type": "list_cleared", "data": {"id": target.id, "name": target.title}} return {"success": True, "type": "list_cleared", "data": {"id": target.id, "name": target.title}}
elif tool_name == "get_profile":
from fabledassistant.services.user_profile import get_profile as _get_profile
profile = await _get_profile(user_id)
return {
"success": True,
"type": "profile",
"data": {
"display_name": profile.display_name or "",
"job_title": profile.job_title or "",
"industry": profile.industry or "",
"expertise_level": profile.expertise_level or "intermediate",
"response_style": profile.response_style or "balanced",
"tone": profile.tone or "casual",
"interests": profile.interests or [],
},
}
elif tool_name == "update_profile":
from fabledassistant.services.user_profile import update_profile as _update_profile, VALID_EXPERTISE, VALID_STYLES, VALID_TONES
data: dict = {}
for field in ("display_name", "job_title", "industry"):
val = arguments.get(field)
if val is not None:
data[field] = str(val)
expertise = arguments.get("expertise_level")
if expertise in VALID_EXPERTISE:
data["expertise_level"] = expertise
style = arguments.get("response_style")
if style in VALID_STYLES:
data["response_style"] = style
tone = arguments.get("tone")
if tone in VALID_TONES:
data["tone"] = tone
interests = arguments.get("interests")
if isinstance(interests, list):
data["interests"] = [str(i) for i in interests if str(i).strip()]
if not data:
return {"success": False, "error": "No valid fields provided to update."}
profile = await _update_profile(user_id, data)
return {
"success": True,
"type": "profile_updated",
"data": {"fields_updated": list(data.keys())},
}
elif tool_name == "calculate":
import math as _math
expr = str(arguments.get("expression", "")).strip()
if not expr:
return {"success": False, "error": "expression is required"}
allowed_names = {k: v for k, v in vars(_math).items() if not k.startswith("_")}
allowed_names["abs"] = abs
allowed_names["round"] = round
try:
result = eval(expr, {"__builtins__": {}}, allowed_names) # noqa: S307
except Exception as calc_err:
return {"success": False, "error": f"Could not evaluate expression: {calc_err}"}
return {"success": True, "type": "calculation", "data": {"expression": expr, "result": result}}
else: else:
return {"success": False, "error": f"Unknown tool: {tool_name}"} return {"success": False, "error": f"Unknown tool: {tool_name}"}
+19 -4
View File
@@ -239,17 +239,32 @@ async def synthesise(
voice_param = _build_voice_param() voice_param = _build_voice_param()
t0 = time.monotonic() t0 = time.monotonic()
audio_chunks: list = [] audio_chunks: list = []
for _, _, audio in _pipeline(text, voice=voice_param, speed=speed): # type: ignore[misc] try:
if audio is not None: for _, _, audio in _pipeline(text, voice=voice_param, speed=speed): # type: ignore[misc]
audio_chunks.append(audio) if audio is not None:
audio_chunks.append(audio)
except Exception:
logger.exception(
"Kokoro pipeline error during synthesis: %d chars, preview=%r",
len(text), text[:80],
)
raise
if not audio_chunks: if not audio_chunks:
logger.warning(
"Kokoro produced no audio chunks: %d chars, preview=%r",
len(text), text[:80],
)
return b"" return b""
combined = np.concatenate(audio_chunks) combined = np.concatenate(audio_chunks)
buf = io.BytesIO() buf = io.BytesIO()
sf.write(buf, combined, samplerate=24000, format="WAV", subtype="PCM_16") sf.write(buf, combined, samplerate=24000, format="WAV", subtype="PCM_16")
logger.debug("TTS synthesis took %.2fs for %d chars", time.monotonic() - t0, len(text)) elapsed = time.monotonic() - t0
logger.info(
"Kokoro synthesis: %d chars → %d samples (%.2fs, %.0f chars/s)",
len(text), len(combined), elapsed, len(text) / elapsed if elapsed > 0 else 0,
)
return buf.getvalue() return buf.getvalue()
loop = asyncio.get_running_loop() loop = asyncio.get_running_loop()
+186
View File
@@ -0,0 +1,186 @@
"""Tests for the multi-note research pipeline."""
import json
import pytest
from unittest.mock import AsyncMock, patch
@pytest.mark.asyncio
async def test_generate_outline_parses_valid_json():
"""_generate_outline returns parsed sections when model returns valid JSON."""
from fabledassistant.services.research import _generate_outline
outline_json = json.dumps([
{"title": "Section One", "focus": "Focus one"},
{"title": "Section Two", "focus": "Focus two"},
{"title": "Section Three", "focus": "Focus three"},
])
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=outline_json):
result = await _generate_outline("test topic", [], "test-model")
assert len(result) == 3
assert result[0]["title"] == "Section One"
assert result[0]["focus"] == "Focus one"
@pytest.mark.asyncio
async def test_generate_outline_returns_empty_on_parse_failure():
"""_generate_outline returns [] when model output is not parseable JSON."""
from fabledassistant.services.research import _generate_outline
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value="not json at all"):
result = await _generate_outline("test topic", [], "test-model")
assert result == []
@pytest.mark.asyncio
async def test_generate_outline_returns_empty_when_fewer_than_3_sections():
"""_generate_outline returns [] when model returns fewer than 3 valid sections."""
from fabledassistant.services.research import _generate_outline
outline_json = json.dumps([{"title": "Only One", "focus": "Focus"}])
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=outline_json):
result = await _generate_outline("test topic", [], "test-model")
assert result == []
@pytest.mark.asyncio
async def test_generate_outline_truncates_to_8():
"""_generate_outline truncates results to at most 8 sections."""
from fabledassistant.services.research import _generate_outline
sections = [{"title": f"Section {i}", "focus": f"Focus {i}"} for i in range(10)]
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=json.dumps(sections)):
result = await _generate_outline("test topic", [], "test-model")
assert len(result) == 8
@pytest.mark.asyncio
async def test_generate_outline_skips_malformed_entries():
"""_generate_outline filters out entries missing title or focus."""
from fabledassistant.services.research import _generate_outline
outline_json = json.dumps([
{"title": "Good One", "focus": "Good focus"},
{"title": "Missing focus"},
{"focus": "Missing title"},
{"title": "Good Two", "focus": "Good focus two"},
{"title": "Good Three", "focus": "Good focus three"},
])
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value=outline_json):
result = await _generate_outline("test topic", [], "test-model")
assert len(result) == 3
assert all(s["title"] and s["focus"] for s in result)
@pytest.mark.asyncio
async def test_synthesize_section_returns_title_and_body():
"""_synthesize_section returns (section_title, body) from model output."""
from fabledassistant.services.research import _synthesize_section
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value="The body of this section."):
title, body = await _synthesize_section(
section_title="Quantum Entanglement: Mechanisms",
section_focus="How entanglement works at the particle level",
sources=[],
model="test-model",
)
assert title == "Quantum Entanglement: Mechanisms"
assert body == "The body of this section."
@pytest.mark.asyncio
async def test_synthesize_section_strips_whitespace():
"""_synthesize_section strips leading/trailing whitespace from body."""
from fabledassistant.services.research import _synthesize_section
with patch("fabledassistant.services.research.generate_completion", new_callable=AsyncMock, return_value="\n\n body text \n\n"):
title, body = await _synthesize_section("Title", "Focus", [], "model")
assert body == "body text"
@pytest.mark.asyncio
async def test_pipeline_creates_section_notes_and_index():
"""run_research_pipeline creates N section notes + 1 index note, returns index."""
from unittest.mock import MagicMock
outline = [
{"title": "Section A", "focus": "Focus A"},
{"title": "Section B", "focus": "Focus B"},
{"title": "Section C", "focus": "Focus C"},
]
note_id_counter = iter(range(10, 20))
def _make_note(user_id, title, body, tags, project_id=None):
n = MagicMock()
n.id = next(note_id_counter)
n.title = title
return n
with patch("fabledassistant.services.research._generate_sub_queries", new_callable=AsyncMock, return_value=["q1"]), \
patch("fabledassistant.services.research._search_searxng", new_callable=AsyncMock, return_value=[{"url": "http://x.com", "title": "X", "snippet": "s"}]), \
patch("fabledassistant.services.research.fetch_url_content", new_callable=AsyncMock, return_value="content"), \
patch("fabledassistant.services.research._generate_outline", new_callable=AsyncMock, return_value=outline), \
patch("fabledassistant.services.research._synthesize_section", new_callable=AsyncMock, side_effect=lambda t, f, s, m: (t, f"Body for {t}")), \
patch("fabledassistant.services.research.create_note", new_callable=AsyncMock, side_effect=_make_note):
from fabledassistant.services.research import run_research_pipeline
result = await run_research_pipeline("test topic", user_id=1, model="test-model")
assert result.title == "Research: test topic"
@pytest.mark.asyncio
async def test_pipeline_falls_back_to_single_note_when_outline_empty():
"""run_research_pipeline falls back to single-note synthesis when _generate_outline returns []."""
from unittest.mock import MagicMock
single_note = MagicMock()
single_note.id = 99
single_note.title = "Research: test topic"
with patch("fabledassistant.services.research._generate_sub_queries", new_callable=AsyncMock, return_value=["q1"]), \
patch("fabledassistant.services.research._search_searxng", new_callable=AsyncMock, return_value=[{"url": "http://x.com", "title": "X", "snippet": "s"}]), \
patch("fabledassistant.services.research.fetch_url_content", new_callable=AsyncMock, return_value="content"), \
patch("fabledassistant.services.research._generate_outline", new_callable=AsyncMock, return_value=[]), \
patch("fabledassistant.services.research._synthesize_note", new_callable=AsyncMock, return_value=("Research: test topic", "body")), \
patch("fabledassistant.services.research.create_note", new_callable=AsyncMock, return_value=single_note):
from fabledassistant.services.research import run_research_pipeline
result = await run_research_pipeline("test topic", user_id=1, model="test-model")
assert result.id == 99
@pytest.mark.asyncio
async def test_pipeline_falls_back_when_all_sections_fail():
"""run_research_pipeline falls back to single note when all section syntheses raise."""
from unittest.mock import MagicMock
outline = [
{"title": "Section A", "focus": "Focus A"},
{"title": "Section B", "focus": "Focus B"},
{"title": "Section C", "focus": "Focus C"},
]
fallback_note = MagicMock()
fallback_note.id = 77
fallback_note.title = "Research: test topic"
with patch("fabledassistant.services.research._generate_sub_queries", new_callable=AsyncMock, return_value=["q1"]), \
patch("fabledassistant.services.research._search_searxng", new_callable=AsyncMock, return_value=[{"url": "http://x.com", "title": "X", "snippet": "s"}]), \
patch("fabledassistant.services.research.fetch_url_content", new_callable=AsyncMock, return_value="content"), \
patch("fabledassistant.services.research._generate_outline", new_callable=AsyncMock, return_value=outline), \
patch("fabledassistant.services.research._synthesize_section", new_callable=AsyncMock, side_effect=RuntimeError("synthesis failed")), \
patch("fabledassistant.services.research._synthesize_note", new_callable=AsyncMock, return_value=("Research: test topic", "body")), \
patch("fabledassistant.services.research.create_note", new_callable=AsyncMock, return_value=fallback_note):
from fabledassistant.services.research import run_research_pipeline
result = await run_research_pipeline("test topic", user_id=1, model="test-model")
assert result.id == 77