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]})
|
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")
|
@journal_bp.post("/trigger-prep")
|
||||||
@login_required
|
@login_required
|
||||||
async def trigger_prep():
|
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..."})
|
buf.append_event("status", {"status": "Building context..."})
|
||||||
|
|
||||||
# Phase 1: Resolve the tools list for this user, scoped to conversation type.
|
# 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 import async_session as _async_session
|
||||||
from fabledassistant.models.conversation import Conversation as _Conversation
|
from fabledassistant.models.conversation import Conversation as _Conversation
|
||||||
async with _async_session() as _sess:
|
async with _async_session() as _sess:
|
||||||
@@ -184,7 +191,15 @@ async def run_generation(
|
|||||||
_conversation_type = (
|
_conversation_type = (
|
||||||
_conv.conversation_type if _conv and _conv.conversation_type else "chat"
|
_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(
|
logger.info(
|
||||||
"Starting generation for conv %d: model=%s, tools=%d",
|
"Starting generation for conv %d: model=%s, tools=%d",
|
||||||
|
|||||||
Reference in New Issue
Block a user