feat(journal): chat model has no tools; curator runs them async (Phase 1a)
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/<conv_id> 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) <noreply@anthropic.com>
This commit is contained in:
@@ -171,6 +171,39 @@ async def list_days():
|
||||
return jsonify({"days": [d.isoformat() for d in rows]})
|
||||
|
||||
|
||||
@journal_bp.post("/curator/run/<int:conv_id>")
|
||||
@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():
|
||||
|
||||
@@ -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
|
||||
@@ -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",
|
||||
|
||||
Reference in New Issue
Block a user