From a7002a89a02f66212113105de52cf30108ff6b9d Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Fri, 22 May 2026 09:03:24 -0400 Subject: [PATCH] feat(journal): chat model has no tools; curator runs them async (Phase 1a) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backend half of the conversation+curator architecture (Fable #172). Decouples the journal chat surface from tool calling: the chat model now sees `tools=[]` and just talks, while a separate curator pass extracts beats and fires the tool calls. services/generation_task.py: - When conversation_type == "journal", pass `tools=[]` to Ollama regardless of what the journal tool set would normally provide. The chat model literally cannot fire record_moment / create_task / etc., so it cannot lie about firing them — the primary failure mode this architecture removes. services/curator.py (new): - `run_curator_for_conversation(conv_id, since=None)` loads recent messages, builds a curator-specific system prompt (extract beats, emit tool calls, optionally a one-line summary), and iterates the Ollama tool-call loop using the user's background_model so the chat model's KV cache survives. - Same tool registry as a normal journal conversation (record_moment, search_notes, update_task, create_task, save_person, save_place, etc.). The curator chooses naturally among them; no need for a separate curator-specific filter. - Returns CuratorRunResult with per-call status + a summary line. - Caps at 4 tool-call rounds — bounded task (extract beats from a fixed transcript), shouldn't need more. - Errors land in result.error rather than raising; the manual trigger surface (and later the scheduler) want a structured result, not exceptions. routes/journal.py: - New POST /api/journal/curator/run/ for manual triggers. Validates conv ownership before running. Returns the CuratorRunResult dict so the UI can show what was captured. What's not in this commit (deferred to later phases): - The scheduler that auto-runs the curator (phase 2 — adds the `conversations.last_curator_run_at` column + APScheduler job). - Curator → chat feedback loop (phase 3 — summary gets injected into subsequent chat system prompts). - Right-rail captures panel in JournalView (phase 1b — pure frontend work, separate commit for clean review). - Research surface separation (phase 4). Co-Authored-By: Claude Opus 4.7 (1M context) --- src/fabledassistant/routes/journal.py | 33 ++ src/fabledassistant/services/curator.py | 323 ++++++++++++++++++ .../services/generation_task.py | 17 +- 3 files changed, 372 insertions(+), 1 deletion(-) create mode 100644 src/fabledassistant/services/curator.py diff --git a/src/fabledassistant/routes/journal.py b/src/fabledassistant/routes/journal.py index 910a9b0..cee486e 100644 --- a/src/fabledassistant/routes/journal.py +++ b/src/fabledassistant/routes/journal.py @@ -171,6 +171,39 @@ async def list_days(): return jsonify({"days": [d.isoformat() for d in rows]}) +@journal_bp.post("/curator/run/") +@login_required +async def trigger_curator_run(conv_id: int): + """Manually run the journal curator over a conversation. + + The curator reads recent messages and fires tool calls (record_moment, + update_task, etc.) the chat model can't (chat models have tools=[]). + Returns a summary of what was captured. + + See services/curator.py for the architectural background. + """ + user_id = get_current_user_id() + + # Confirm the conversation belongs to this user (curator runs against + # arbitrary conv_ids would otherwise leak data across tenants). + from sqlalchemy import select as _select + from fabledassistant.models import async_session as _async_session + from fabledassistant.models.conversation import Conversation as _Conversation + async with _async_session() as _sess: + _res = await _sess.execute( + _select(_Conversation).where( + _Conversation.id == conv_id, + _Conversation.user_id == user_id, + ) + ) + if _res.scalar_one_or_none() is None: + return jsonify({"error": "Conversation not found"}), 404 + + from fabledassistant.services.curator import run_curator_for_conversation + result = await run_curator_for_conversation(conv_id) + return jsonify(result.to_dict()) + + @journal_bp.post("/trigger-prep") @login_required async def trigger_prep(): diff --git a/src/fabledassistant/services/curator.py b/src/fabledassistant/services/curator.py new file mode 100644 index 0000000..8401b89 --- /dev/null +++ b/src/fabledassistant/services/curator.py @@ -0,0 +1,323 @@ +"""Journal curator: async LLM pass that extracts captures from a chat. + +Architecture (2026-05-22 brainstorm, Fable note #172): + +The journal chat model has no tools — it just talks. This curator is the +second LLM pass that reads recent journal messages and fires the tool +calls (record_moment, update_task, etc.) the chat model can't. + +Runs against the user's `background_model` so the chat model's KV cache +isn't disturbed. Can be triggered manually via the journal route or by +the scheduler (phase 2). The chat model is unaffected during a curator +run; this is intentionally fire-and-go. + +The curator does NOT add its own messages to the conversation. Only the +side-effects of its tool calls land in the database (moments table, +notes table, tasks, etc.). The user sees those side-effects via the +existing journal data surfaces, not via a chat-stream injection — see +the brainstorm doc's surfacing decision (right-rail captures panel, not +inline observer voice). +""" +from __future__ import annotations + +import json +import logging +import time +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone + +from sqlalchemy import select + +from fabledassistant.config import Config +from fabledassistant.models import async_session +from fabledassistant.models.conversation import Conversation, Message +from fabledassistant.services.llm import pick_num_ctx, stream_chat_with_tools +from fabledassistant.services.settings import get_setting +from fabledassistant.services.tools import execute_tool, get_tools_for_user + +logger = logging.getLogger(__name__) + +# Tool-call iteration cap. The chat path uses 6; the curator should +# converge faster because its task is bounded (extract beats from a +# fixed transcript, not respond to evolving conversation). +_MAX_TOOL_ROUNDS = 4 + +_CURATOR_SYSTEM_PROMPT = """You are a curator reading a fragment of the user's journal conversation. Your job is to capture meaningful beats as structured records using the tools provided. You do NOT respond to the user — your only output is tool calls. + +Beats worth recording: +- Events that happened ("went grocery shopping", "finished the network restage") +- Encounters with people ("had coffee with Sarah", "called Mom") +- Decisions ("going to switch jobs", "won't pursue the contract") +- Observations about the user's state or world ("the new place is loud", "feeling tired") +- Plans and commitments ("watching a show tonight", "dentist Thursday") +- Small accomplishments or changes the user made ("installed the new AP", "shipped the migration") + +Rules: +- Use record_moment to capture each distinct beat. One tool call per beat — do not collapse multiple beats into one. +- When linking to entities (people, places, tasks, notes), use the *_names parameters and let the server resolve. Never invent ids. +- Before linking a task by title, call search_notes to confirm it exists. If you have not searched, do not pass task_titles. +- If the user explicitly references an existing task by name, prefer update_task to mark progress. If they describe finishing something, set status=done. +- If the user mentions a person or place you do not already know about, you may call save_person or save_place to create the entry. Otherwise skip new-entity creation — better to omit a link than to invent the wrong one. +- Skip meta-conversational fragments ("ok", "thanks", "got it") — those are not journal beats. +- Match the user's voice when writing moment content. First-person or imperative. Never "the user mentioned…" / "user reports…" framing. + +After the tool calls, you may emit one short summary sentence (≤ 20 words) describing what you captured. The summary is shown back to the chat model in subsequent turns so it stays aware of recent topics; it is NOT shown to the user directly. Examples: +- "Captured network restage progress and a coffee mention with Sarah." +- "Recorded plan for tonight; nothing else stood out." +- "" (empty if nothing was captured — perfectly fine). +""" + + +@dataclass +class CuratorToolCall: + """One tool call attempted by the curator.""" + + name: str + arguments: dict + result: dict | None = None + error: str | None = None + status: str = "pending" # success | error | pending + + +@dataclass +class CuratorRunResult: + """What the curator did in a single pass over a conversation.""" + + conv_id: int + user_id: int + model: str + messages_examined: int + tool_calls: list[CuratorToolCall] = field(default_factory=list) + summary: str = "" + duration_ms: int = 0 + error: str | None = None + + @property + def tools_attempted(self) -> int: + return len(self.tool_calls) + + @property + def tools_succeeded(self) -> int: + return sum(1 for tc in self.tool_calls if tc.status == "success") + + def to_dict(self) -> dict: + return { + "conv_id": self.conv_id, + "user_id": self.user_id, + "model": self.model, + "messages_examined": self.messages_examined, + "tool_calls": [ + { + "name": tc.name, + "arguments": tc.arguments, + "status": tc.status, + "error": tc.error, + } + for tc in self.tool_calls + ], + "tools_attempted": self.tools_attempted, + "tools_succeeded": self.tools_succeeded, + "summary": self.summary, + "duration_ms": self.duration_ms, + "error": self.error, + } + + +def _format_transcript(messages: list[Message]) -> str: + """Render a list of Message rows as a plain transcript the curator can read. + + Tool-call messages and previous assistant content are included so the + curator has full context, but the curator's own focus is on extracting + beats from user messages. + """ + lines: list[str] = [] + for m in messages: + if not m.content: + continue + ts = m.created_at.strftime("%H:%M") if m.created_at else "??:??" + role = m.role.capitalize() if m.role else "Unknown" + lines.append(f"[{ts}] {role}: {m.content.strip()}") + return "\n".join(lines) + + +async def _load_messages_since( + conv_id: int, since: datetime | None +) -> list[Message]: + """Load conversation messages since the cutoff (or all of today).""" + async with async_session() as session: + stmt = select(Message).where(Message.conversation_id == conv_id) + if since is not None: + stmt = stmt.where(Message.created_at > since) + stmt = stmt.order_by(Message.created_at.asc()) + result = await session.execute(stmt) + return list(result.scalars().all()) + + +async def run_curator_for_conversation( + conv_id: int, + *, + since: datetime | None = None, + user_id_override: int | None = None, +) -> CuratorRunResult: + """Run a single curator pass over the given journal conversation. + + Args: + conv_id: target conversation. + since: only consider messages after this timestamp. Defaults to + the last 24 hours so a first manual trigger gets the day's + worth of context without going back forever. + user_id_override: optional — used by the scheduler to attribute + the run to the conversation's owner without re-fetching. + Manual triggers from the route pass None and we read from + the conversation row. + + Returns a CuratorRunResult; never raises (errors land in result.error). + """ + started_at = time.monotonic() + + async with async_session() as session: + conv = await session.get(Conversation, conv_id) + if conv is None: + return CuratorRunResult( + conv_id=conv_id, user_id=0, model="", + messages_examined=0, error=f"Conversation {conv_id} not found", + ) + if conv.conversation_type != "journal": + return CuratorRunResult( + conv_id=conv_id, user_id=conv.user_id, model="", + messages_examined=0, + error=f"Curator only runs on journal conversations (got {conv.conversation_type!r})", + ) + + user_id = user_id_override or conv.user_id + # Default lookback: last 24h. Phase 2's scheduler will narrow this + # by passing the conversation's last_curator_run_at as `since`. + if since is None: + since = datetime.now(timezone.utc) - timedelta(hours=24) + + messages = await _load_messages_since(conv_id, since) + if not messages: + logger.info( + "Curator skipped conv %d: no new messages since %s", + conv_id, since.isoformat(), + ) + return CuratorRunResult( + conv_id=conv_id, user_id=user_id, + model="", messages_examined=0, + duration_ms=int((time.monotonic() - started_at) * 1000), + ) + + # Use the background model so the chat model's KV cache survives. + # Falls back to OLLAMA_MODEL if no background model is configured. + model = await get_setting(user_id, "background_model", "") or Config.OLLAMA_MODEL + + tools = await get_tools_for_user(user_id, conversation_type="journal") + transcript = _format_transcript(messages) + + user_prompt = ( + "Below is a fragment of the user's journal conversation. Extract " + "the captureable beats using the tools provided, then emit one " + "short summary line (or empty).\n\n" + "TRANSCRIPT:\n" + f"{transcript}\n" + ) + + llm_messages: list[dict] = [ + {"role": "system", "content": _CURATOR_SYSTEM_PROMPT}, + {"role": "user", "content": user_prompt}, + ] + + result = CuratorRunResult( + conv_id=conv_id, user_id=user_id, model=model, + messages_examined=len(messages), + ) + + try: + num_ctx = pick_num_ctx(llm_messages, tools=tools) + summary_chunks: list[str] = [] + + # Tool-call iteration loop — same shape as run_generation, but + # without SSE streaming since nothing is watching live. + for round_idx in range(_MAX_TOOL_ROUNDS): + tool_calls_this_round: list[dict] = [] + content_this_round: list[str] = [] + + async for chunk in stream_chat_with_tools( + llm_messages, model, tools=tools, think=False, num_ctx=num_ctx, + ): + if chunk.type == "content": + content_this_round.append(chunk.content) + elif chunk.type == "tool_calls": + tool_calls_this_round = chunk.tool_calls or [] + elif chunk.type == "done": + break + + # The model's content this round contributes to the summary + # only on the final round (after no more tool calls fire). + if not tool_calls_this_round: + summary_chunks.append("".join(content_this_round).strip()) + break + + # Execute each tool call, capture result, append back into the + # message list as a tool-role message so the model sees what + # happened on the next round. + llm_messages.append({ + "role": "assistant", + "content": "".join(content_this_round), + "tool_calls": tool_calls_this_round, + }) + + for tc in tool_calls_this_round: + fn = tc.get("function", {}) if isinstance(tc, dict) else {} + name = fn.get("name") or "" + args = fn.get("arguments") or {} + if isinstance(args, str): + try: + args = json.loads(args) + except Exception: + args = {} + call = CuratorToolCall(name=name, arguments=args) + try: + tool_result = await execute_tool( + user_id, name, args, conv_id=conv_id, + ) + call.result = tool_result + call.status = ( + "success" if (tool_result or {}).get("success", True) + and not (tool_result or {}).get("error") + else "error" + ) + if call.status == "error": + call.error = str((tool_result or {}).get("error", ""))[:500] + except Exception as e: + call.status = "error" + call.error = f"{type(e).__name__}: {e}"[:500] + tool_result = {"success": False, "error": call.error} + logger.exception("Curator tool %r failed for conv %d", name, conv_id) + result.tool_calls.append(call) + llm_messages.append({ + "role": "tool", + "content": json.dumps(tool_result)[:4000], + }) + else: + logger.warning( + "Curator hit _MAX_TOOL_ROUNDS=%d for conv %d (still emitting tool calls)", + _MAX_TOOL_ROUNDS, conv_id, + ) + + # Trim summary: at most one sentence, cap length aggressively. + summary = " ".join(summary_chunks).strip().splitlines() + result.summary = (summary[0][:240] if summary and summary[0] else "") + except Exception as e: + result.error = f"{type(e).__name__}: {e}" + logger.exception("Curator run failed for conv %d", conv_id) + + result.duration_ms = int((time.monotonic() - started_at) * 1000) + logger.info( + "Curator pass complete: conv=%d model=%s messages=%d " + "tool_calls=%d (ok=%d) duration=%dms summary=%r", + conv_id, model, result.messages_examined, + result.tools_attempted, result.tools_succeeded, + result.duration_ms, result.summary[:60], + ) + return result diff --git a/src/fabledassistant/services/generation_task.py b/src/fabledassistant/services/generation_task.py index 401ccb5..9d3d3bd 100644 --- a/src/fabledassistant/services/generation_task.py +++ b/src/fabledassistant/services/generation_task.py @@ -177,6 +177,13 @@ async def run_generation( buf.append_event("status", {"status": "Building context..."}) # Phase 1: Resolve the tools list for this user, scoped to conversation type. + # + # Journal conversations get NO tools (2026-05-22 architecture pivot): + # the chat model talks, a separate background curator does tool calls + # asynchronously. See services/curator.py. Removing tools here is the + # mechanical change that makes the architecture real — the chat model + # can no longer fire record_moment / create_task / etc. and therefore + # can no longer lie about firing them. from fabledassistant.models import async_session as _async_session from fabledassistant.models.conversation import Conversation as _Conversation async with _async_session() as _sess: @@ -184,7 +191,15 @@ async def run_generation( _conversation_type = ( _conv.conversation_type if _conv and _conv.conversation_type else "chat" ) - tools = await get_tools_for_user(user_id, conversation_type=_conversation_type) + if _conversation_type == "journal": + tools = [] + logger.info( + "Conv %d is journal: passing tools=[] to chat model " + "(curator handles tool calls async)", + conv_id, + ) + else: + tools = await get_tools_for_user(user_id, conversation_type=_conversation_type) logger.info( "Starting generation for conv %d: model=%s, tools=%d",