Merge main into spending report totals and preserve artifact output

This commit is contained in:
2026-10-12 19:47:52 -04:00
55 changed files with 6641 additions and 208 deletions
+180 -14
View File
@@ -5,6 +5,7 @@ import os
import re
import time
import uuid
from urllib.parse import quote, unquote
from quart import Blueprint, jsonify, make_response, request
from quart_jwt_extended import get_jwt_identity
@@ -16,9 +17,19 @@ from blueprints.users.decorators import jwt_access_required
from metrics import count_result, observe_response, time_response
from utils.image_process import image_content_block
from utils.strip_markdown import extract_agent_text
from utils.file_extract import (
FILE_TYPES,
FileExtractionError,
attachment_block,
extract_text,
file_extension,
)
from utils.image_upload import ALLOWED_TYPES as IMAGE_TYPES
from utils.image_upload import ImageValidationError, process_image
from utils.s3_client import generate_presigned_url as s3_presigned_url
from utils.s3_client import get_file as s3_get_file
from utils.s3_client import get_image as s3_get_image
from utils.s3_client import upload_file as s3_upload_file
from utils.s3_client import upload_image as s3_upload_image
from .logic import (
@@ -37,7 +48,35 @@ from .models import (
from .tool_events import tool_end_payload, tool_name_from_message, tool_start_payload
from .tool_labels import tool_label
TTFT_TIMEOUT = float(os.getenv("TTFT_TIMEOUT_SECONDS", "30"))
# Longest an in-flight model call may go without producing any output (prompt
# prefill included) before the turn is abandoned as stalled. Thinking tokens and
# empty chunks count as output; time spent running tools does not count at all.
MODEL_STALL_TIMEOUT = float(os.getenv("TTFT_TIMEOUT_SECONDS", "30"))
def stall_timeout(last_model_activity: float | None, now: float) -> float | None:
"""Seconds left before the in-flight model call counts as stalled.
None when no model call is in flight (e.g. while tools run): the httpx read
timeout and the tools' own timeouts cover those phases.
"""
if last_model_activity is None:
return None
return max(MODEL_STALL_TIMEOUT - (now - last_model_activity), 1.0)
def _model_call_summary(call: dict) -> dict:
"""Per-LLM-call timing and token counts for the stream_request log line."""
start = call["start"]
first_chunk = call.get("first_chunk")
end = call.get("end")
return {
"prefill_ms": round((first_chunk - start) * 1000) if first_chunk else None,
"ms": round((end - start) * 1000) if end else None,
"prompt_tokens": call.get("prompt_tokens"),
"output_tokens": call.get("output_tokens"),
}
_THINK_OPEN = "<think>"
_THINK_CLOSE = "</think>"
@@ -203,6 +242,87 @@ async def upload_image():
return jsonify({"image_key": key})
def _file_key_belongs_to(file_key: str, conversation_id) -> bool:
"""Only allow attachments stored under this conversation's prefix."""
return (
isinstance(file_key, str)
and ".." not in file_key
and file_key.startswith(f"conversations/{conversation_id}/files/")
)
async def _load_attached_file(file_key: str) -> tuple[str, str]:
"""Fetch an attached file from S3 and return (original name, extracted text)."""
data, metadata = await s3_get_file(file_key)
name = unquote(metadata.get("filename", "")) or file_key.rsplit("/", 1)[-1]
ext = file_extension(file_key, None) or ""
# PyMuPDF is CPU-bound; keep it off the event loop so other streams don't stall.
return name, await asyncio.to_thread(extract_text, data, ext)
@conversation_blueprint.post("/upload")
@jwt_access_required
async def upload_attachment():
current_user_uuid = get_jwt_identity()
files = await request.files
form = await request.form
file = files.get("file")
conversation_id = form.get("conversation_id")
if not file or not conversation_id:
return jsonify({"error": "file and conversation_id are required"}), 400
conversation = await _get_owned_conversation(conversation_id, current_user_uuid)
if conversation is None:
return jsonify({"error": "conversation not found"}), 404
file_bytes = file.read()
content_type = (file.content_type or "").split(";")[0].strip().lower()
name = os.path.basename(file.filename or "") or "attachment"
if content_type in IMAGE_TYPES:
try:
processed_bytes, output_content_type = process_image(
file_bytes, content_type
)
except Exception as e:
return jsonify({"error": f"Could not process image: {e}"}), 400
ext = output_content_type.split("/")[-1]
if ext == "jpeg":
ext = "jpg"
key = f"conversations/{conversation.id}/{uuid.uuid4()}.{ext}"
await s3_upload_image(processed_bytes, key, output_content_type)
return jsonify({"kind": "image", "image_key": key})
ext = file_extension(name, content_type)
if ext is None:
return (
jsonify(
{
"error": f"Unsupported file type: {content_type or name}. "
"Allowed: images, PDF, TXT, MD, CSV"
}
),
415,
)
try:
text = await asyncio.to_thread(extract_text, file_bytes, ext)
except FileExtractionError as e:
return jsonify({"error": str(e)}), 400
if not text.strip():
message = (
"This PDF has no text layer (scanned?), so Simba can't read it yet."
if ext == "pdf"
else "This file is empty, so there's nothing for Simba to read."
)
return jsonify({"error": message}), 422
key = f"conversations/{conversation.id}/files/{uuid.uuid4()}.{ext}"
await s3_upload_file(file_bytes, key, FILE_TYPES[ext], {"filename": quote(name)})
return jsonify({"kind": "file", "file_key": key, "name": name, "chars": len(text)})
@conversation_blueprint.get("/image/<path:image_key>")
@jwt_access_required
async def serve_image(image_key: str):
@@ -219,12 +339,30 @@ async def stream_query():
query_text = data.get("query")
conversation_id = data.get("conversation_id")
image_key = data.get("image_key")
file_key = data.get("file_key")
conversation = await _get_owned_conversation(conversation_id, current_user_uuid)
if conversation is None:
return jsonify({"error": "conversation not found"}), 404
# An attached file is extracted and appended to the agent's user turn; the
# saved message only records its name.
saved_text = query_text or ""
if file_key:
if not _file_key_belongs_to(file_key, conversation.id):
return jsonify({"error": "invalid file_key"}), 400
try:
file_name, file_text = await _load_attached_file(file_key)
block = attachment_block(file_name, file_text)
except Exception as e:
logging.error(f"Failed to load file {file_key}: {e}")
file_name = file_key.rsplit("/", 1)[-1]
block = "[A file was attached but could not be loaded.]"
query_text = f"{query_text or 'The user attached this file.'}\n\n{block}"
saved_text = f"{saved_text}\n\n📎 {file_name}".strip()
await add_message_to_conversation(
conversation=conversation,
message=query_text or "",
message=saved_text,
speaker="user",
user=user,
image_key=image_key,
@@ -271,6 +409,9 @@ async def stream_query():
first_content_at: float | None = None
tool_timers: dict[str, float] = {} # run_id -> start time
tool_spans: list[dict] = []
# Last output from the in-flight model call; None between calls.
last_model_activity: float | None = None
model_calls: dict[str, dict] = {} # run_id -> timing / token usage
n_tokens = 0
model_name: str | None = None
cancelled = False
@@ -293,6 +434,8 @@ async def stream_query():
ttft_ms = (
round((first_content_at - t0) * 1000) if first_content_at else None
)
calls = [_model_call_summary(c) for c in model_calls.values()]
prompt_tokens = [c["prompt_tokens"] for c in calls if c["prompt_tokens"]]
logging.info(
json.dumps(
{
@@ -304,6 +447,8 @@ async def stream_query():
"ttft_ms": ttft_ms,
"tokens": n_tokens,
"tools": tool_spans,
"model_calls": calls,
"max_prompt_tokens": max(prompt_tokens, default=None),
"model": model_name,
"cancelled": cancelled,
**({"error": error} if error else {}),
@@ -316,15 +461,10 @@ async def stream_query():
payload, version="v2", config=agent_config
).__aiter__()
while True:
# Apply a short TTFT timeout only during the window between
# on_chat_model_start and the first content token. Before the
# LLM call starts (tool phase) and after the first token, let
# the httpx read timeout handle stalls instead.
if llm_start_at is not None and first_content_at is None:
ttft_wait = TTFT_TIMEOUT - (time.monotonic() - llm_start_at)
timeout = max(ttft_wait, 1.0)
else:
timeout = None
# Only a silent model call counts as a stall. A model that is
# thinking (or streaming empty chunks) keeps resetting the clock,
# and each new call starts a fresh one.
timeout = stall_timeout(last_model_activity, time.monotonic())
try:
if timeout is not None:
@@ -336,7 +476,9 @@ async def stream_query():
except StopAsyncIteration:
break
except asyncio.TimeoutError:
_log_summary(error=f"TTFT timeout after {TTFT_TIMEOUT:.0f}s")
_log_summary(
error=f"model stalled: no output for {MODEL_STALL_TIMEOUT:.0f}s"
)
yield f"data: {json.dumps({'type': 'error', 'message': 'Simba took too long to respond. Please try again.', 'stalled': True})}\n\n"
return
@@ -388,14 +530,34 @@ async def stream_query():
elif event_type == "on_chat_model_start":
turn_has_content = False
now = time.monotonic()
last_model_activity = now
model_calls[event["run_id"]] = {"start": now}
if llm_start_at is None:
llm_start_at = time.monotonic()
llm_start_at = now
if model_name is None:
model_name = event.get("metadata", {}).get(
"ls_model_name"
) or event.get("name")
elif event_type == "on_chat_model_end":
last_model_activity = None
call = model_calls.get(event["run_id"])
if call is not None:
call["end"] = time.monotonic()
usage = getattr(
event.get("data", {}).get("output"), "usage_metadata", None
)
if usage:
call["prompt_tokens"] = usage.get("input_tokens")
call["output_tokens"] = usage.get("output_tokens")
elif event_type == "on_chat_model_stream":
now = time.monotonic()
last_model_activity = now
call = model_calls.get(event.get("run_id"))
if call is not None and "first_chunk" not in call:
call["first_chunk"] = now
chunk = event.get("data", {}).get("chunk")
raw = getattr(chunk, "content", None)
if isinstance(raw, str) and raw:
@@ -457,7 +619,7 @@ async def stream_query():
_log_summary()
yield "data: [DONE]\n\n"
return await make_response(
response = await make_response(
event_generator(),
200,
{
@@ -466,6 +628,10 @@ async def stream_query():
"X-Accel-Buffering": "no",
},
)
# Quart's default 60s response deadline cancels even active SSE streams.
# Silent model calls are bounded separately by MODEL_STALL_TIMEOUT.
response.timeout = None
return response
@conversation_blueprint.route("/<conversation_id>")
+229 -52
View File
@@ -1,4 +1,5 @@
import os
from functools import partial
from typing import cast
import httpx
@@ -8,9 +9,24 @@ from langchain.tools import tool
from langgraph.prebuilt import create_react_agent
from langchain_core.messages import SystemMessage
from langchain_core.runnables import RunnableConfig, RunnableLambda
from langchain_core.utils.function_calling import convert_to_openai_tool
from langchain_openai import ChatOpenAI
from tavily import AsyncTavilyClient
from blueprints.conversation.artifacts_util import (
NOTE_MAX_CHARS,
format_date_range,
format_day,
json_safe,
make_artifact,
mealie_source_ref,
note_title,
obsidian_source_ref,
search_documents_artifact,
truncate,
unique,
)
from blueprints.conversation.context import RESPONSE_RESERVE_TOKENS, build_context
from blueprints.conversation.memory import save_memory
from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
from blueprints.rag.logic import query_vector_store
@@ -41,11 +57,14 @@ if llama_url:
timeout=httpx.Timeout(LLAMA_READ_TIMEOUT, connect=10.0),
max_retries=1,
model_kwargs={"reasoning_effort": "low"},
# Ask llama-server for token usage on the final chunk (prompt size logging).
stream_usage=True,
max_tokens=RESPONSE_RESERVE_TOKENS,
)
else:
llama_chat = None
openai_fallback = ChatOpenAI(model="gpt-5-mini")
openai_fallback = ChatOpenAI(model="gpt-5-mini", max_tokens=RESPONSE_RESERVE_TOKENS)
model_with_fallback = cast(
BaseChatModel,
llama_chat.with_fallbacks([openai_fallback]) if llama_chat else openai_fallback,
@@ -131,20 +150,36 @@ async def simba_search(query: str):
print(f"[SIMBA SEARCH] Found {len(docs)} documents")
print(f"[SIMBA SEARCH] Serialized result length: {len(serialized)}")
print(f"[SIMBA SEARCH] First 200 chars: {serialized[:200]}")
return serialized, docs
return serialized, search_documents_artifact(query, "paperless", docs)
@tool
async def mealie_todays_meals() -> str:
MEALIE_NOT_CONFIGURED = "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
def _meal_entry(meal: dict) -> dict:
"""Meal plan entry in the `mealie.meal_plan` artifact shape."""
entry = {
"entry_type": meal["entry_type"],
"name": meal["recipe_name"] or meal["title"] or "Untitled",
}
if meal.get("recipe_slug"):
entry["recipe_slug"] = meal["recipe_slug"]
if meal.get("note"):
entry["note"] = meal["note"]
return entry
@tool(response_format="content_and_artifact")
async def mealie_todays_meals():
"""Get today's planned meals from Mealie."""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
result = await mealie_service.get_todays_meals()
if result["total_meals"] == 0:
return f"No meals planned for today ({result['today']})."
return f"No meals planned for today ({result['today']}).", None
lines = [f"Meals planned for today ({result['today']}):"]
for meal in result["meals"]:
@@ -152,13 +187,29 @@ async def mealie_todays_meals() -> str:
lines.append(f"- {meal['entry_type'].capitalize()}: {name}")
if meal["note"]:
lines.append(f" Note: {meal['note']}")
return "\n".join(lines)
today = result["today"]
artifact = make_artifact(
"mealie.meal_plan",
f"Today's meals · {format_day(today)}",
{
"start_date": today,
"end_date": today,
"days": [
{
"date": today,
"meals": [_meal_entry(m) for m in result["meals"]],
}
],
},
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching today's meals: {str(e)}"
return f"Error fetching today's meals: {str(e)}", None
@tool
async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
@tool(response_format="content_and_artifact")
async def mealie_meal_plans(start_date: str = "", end_date: str = ""):
"""Get Mealie meal plans for a date range.
Args:
@@ -166,7 +217,7 @@ async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
end_date: YYYY-MM-DD (default: 7 days out)
"""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
result = await mealie_service.get_meal_plans(
@@ -175,11 +226,15 @@ async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
)
if result["total_plans"] == 0:
return f"No meal plans found from {result['start_date']} to {result['end_date']}."
return (
f"No meal plans found from {result['start_date']} to {result['end_date']}.",
None,
)
lines = [
f"Meal plans from {result['start_date']} to {result['end_date']} ({result['total_plans']} meals):"
]
days = []
for date, plans in sorted(result["plans_by_date"].items()):
lines.append(f"\n{date}:")
for plan in plans:
@@ -187,20 +242,31 @@ async def mealie_meal_plans(start_date: str = "", end_date: str = "") -> str:
lines.append(f" - {plan['entry_type'].capitalize()}: {name}")
if plan["note"]:
lines.append(f" Note: {plan['note']}")
return "\n".join(lines)
days.append({"date": date, "meals": [_meal_entry(p) for p in plans]})
artifact = make_artifact(
"mealie.meal_plan",
f"Meal plan · {format_date_range(result['start_date'], result['end_date'])}",
{
"start_date": result["start_date"],
"end_date": result["end_date"],
"days": days,
},
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching meal plans: {str(e)}"
return f"Error fetching meal plans: {str(e)}", None
@tool
async def mealie_get_recipe(recipe_slug: str) -> str:
@tool(response_format="content_and_artifact")
async def mealie_get_recipe(recipe_slug: str):
"""Get full recipe details (ingredients, instructions, times) from Mealie.
Args:
recipe_slug: Recipe slug from meal plan data
"""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
recipe = await mealie_service.get_recipe(recipe_slug)
@@ -236,22 +302,49 @@ async def mealie_get_recipe(recipe_slug: str) -> str:
if recipe["tags"]:
lines.append(f"\nTags: {', '.join(recipe['tags'])}")
return "\n".join(lines)
instructions = []
for step in recipe["instructions"]:
instruction = {"text": step.get("text", "")}
if step.get("title"):
instruction["title"] = step["title"]
instructions.append(instruction)
slug = recipe.get("slug") or recipe_slug
artifact = make_artifact(
"mealie.recipe",
recipe["name"],
json_safe(
{
"slug": slug,
"name": recipe["name"],
"description": recipe["description"],
"servings": recipe["servings"],
"prep_time": recipe["prep_time"],
"cook_time": recipe["cook_time"],
"total_time": recipe["total_time"],
"ingredients": [ing["display"] for ing in recipe["ingredients"]],
"instructions": instructions,
"tags": recipe["tags"],
}
),
mealie_source_ref(slug),
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching recipe: {str(e)}"
return f"Error fetching recipe: {str(e)}", None
@tool
async def mealie_shopping_lists() -> str:
@tool(response_format="content_and_artifact")
async def mealie_shopping_lists():
"""Get Mealie shopping lists with needed and checked-off items."""
if not mealie_enabled:
return "Mealie integration is not configured. Please set MEALIE_BASE_URL and MEALIE_API_TOKEN environment variables."
return MEALIE_NOT_CONFIGURED, None
try:
result = await mealie_service.get_shopping_lists()
if result["total_lists"] == 0:
return "No shopping lists found."
return "No shopping lists found.", None
lines = [f"Shopping lists ({result['total_unchecked_items']} items needed):"]
for sl in result["shopping_lists"]:
@@ -270,9 +363,37 @@ async def mealie_shopping_lists() -> str:
lines.append(
f" ({len(sl['checked_items'])} items already checked off)"
)
return "\n".join(lines)
lists = [
{
"name": sl["name"],
"items": [
{
"name": item["name"],
"quantity": item["quantity"],
"checked": False,
}
for item in sl["unchecked_items"]
]
+ [
{
"name": item["name"],
"quantity": item["quantity"],
"checked": True,
}
for item in sl["checked_items"]
],
}
for sl in result["shopping_lists"]
]
artifact = make_artifact(
"mealie.shopping_lists",
"Shopping lists",
json_safe({"lists": lists}),
)
return "\n".join(lines), artifact
except Exception as e:
return f"Error fetching shopping lists: {str(e)}"
return f"Error fetching shopping lists: {str(e)}", None
@tool
@@ -309,34 +430,63 @@ async def mealie_create_meal_plan(
return f"Error creating meal plan: {str(e)}"
@tool
async def obsidian_search_notes(query: str) -> str:
OBSIDIAN_NOT_CONFIGURED = "Obsidian integration is not configured. Please set OBSIDIAN_VAULT_PATH environment variable."
def _note_artifact(path: str, parsed: dict, title: str | None = None) -> dict:
"""`obsidian.note` artifact from ObsidianService.parse_markdown output."""
frontmatter = json_safe(parsed["metadata"])
title = title or note_title(path, frontmatter)
return make_artifact(
"obsidian.note",
title,
{
"path": path,
"title": title,
"markdown": truncate(parsed["content"], NOTE_MAX_CHARS),
"tags": unique(parsed["tags"]),
"frontmatter": frontmatter,
},
obsidian_source_ref(path),
)
@tool(response_format="content_and_artifact")
async def obsidian_search_notes(query: str):
"""Search personal Obsidian vault notes.
Args:
query: Search query
"""
if not obsidian_enabled:
return "Obsidian integration is not configured. Please set OBSIDIAN_VAULT_PATH environment variable."
return OBSIDIAN_NOT_CONFIGURED, None
try:
# Query vector store filtered to obsidian source only
serialized, docs = await query_vector_store(query=query, source="obsidian")
return serialized
return serialized, search_documents_artifact(query, "obsidian", docs)
except Exception as e:
return f"Error searching Obsidian notes: {str(e)}"
return f"Error searching Obsidian notes: {str(e)}", None
@tool
async def obsidian_read_note(relative_path: str) -> str:
"""Read the full content of a specific Obsidian note by path.
@tool(response_format="content_and_artifact")
async def obsidian_read_note(
relative_path: str, start_line: int = 1, max_lines: int | None = None
):
"""Read an Obsidian note, optionally selecting a small range of content lines.
Use line ranges to retrieve omitted sections of a large note.
Args:
relative_path: Path relative to vault root (e.g., "notes/my-note.md")
start_line: First content line to read, starting at 1 (excludes frontmatter).
max_lines: Number of lines to read, or None for all remaining lines.
"""
if not obsidian_enabled:
return "Obsidian integration is not configured. Please set OBSIDIAN_VAULT_PATH environment variable."
return OBSIDIAN_NOT_CONFIGURED, None
if start_line < 1 or (max_lines is not None and max_lines < 1):
return "start_line and max_lines must be positive integers.", None
try:
note = obsidian_service.read_note(relative_path)
@@ -344,16 +494,37 @@ async def obsidian_read_note(relative_path: str) -> str:
result = f"File: {note['path']}\n\n"
result += f"Frontmatter:\n{content_data['metadata']}\n\n"
result += f"Content:\n{content_data['content']}\n\n"
content = content_data["content"]
if start_line != 1 or max_lines is not None:
lines = content.splitlines()
if start_line > len(lines):
return (
f"File: {note['path']} has {len(lines)} content lines; start_line is beyond the end.",
None,
)
end_line = (
min(len(lines), start_line - 1 + max_lines)
if max_lines is not None
else len(lines)
)
content = "\n".join(lines[start_line - 1 : end_line])
result += (
f"Content lines {start_line}-{end_line} of {len(lines)}:\n{content}\n\n"
)
else:
result += f"Content:\n{content}\n\n"
result += f"Tags: {', '.join(content_data['tags'])}\n"
result += f"Contains {len(content_data['wikilinks'])} wikilinks and {len(content_data['embeds'])} embeds"
return result
return result, _note_artifact(note["path"], content_data)
except FileNotFoundError:
return f"Note not found at '{relative_path}'. Please check the path is correct."
return (
f"Note not found at '{relative_path}'. Please check the path is correct.",
None,
)
except Exception as e:
return f"Error reading note: {str(e)}"
return f"Error reading note: {str(e)}", None
@tool
@@ -391,19 +562,27 @@ async def obsidian_create_note(
return f"Error creating note: {str(e)}"
@tool
def journal_get_today() -> str:
@tool(response_format="content_and_artifact")
def journal_get_today():
"""Get today's full daily journal note including tasks and log entries."""
if not obsidian_enabled:
return "Obsidian integration is not configured."
return "Obsidian integration is not configured.", None
try:
note = obsidian_service.get_daily_note()
if not note["found"]:
return f"No daily note found for {note['date']}. Use journal_add_task to create one."
return f"Daily note for {note['date']}:\n\n{note['content']}"
return (
f"No daily note found for {note['date']}. Use journal_add_task to create one.",
None,
)
artifact = _note_artifact(
note["path"],
obsidian_service.parse_markdown(note["content"]),
title=f"Daily note · {format_day(note['date'])}",
)
return f"Daily note for {note['date']}:\n\n{note['content']}", artifact
except Exception as e:
return f"Error reading daily note: {str(e)}"
return f"Error reading daily note: {str(e)}", None
@tool
@@ -950,8 +1129,7 @@ async def beli_search_restaurants(
tools.append(beli_search_restaurants)
@RunnableLambda
def _build_prompt(state, config: RunnableConfig = None) -> list:
def _build_prompt(state, config: RunnableConfig = None, *, tool_schemas=()) -> list:
"""Inject the per-request system prompt, replacing any prior system messages."""
system_prompt = (
(config or {}).get("configurable", {}).get("system_prompt", SIMBA_SYSTEM_PROMPT)
@@ -959,10 +1137,7 @@ def _build_prompt(state, config: RunnableConfig = None) -> list:
# LangGraph passes the full state dict {"messages": [...]}, not just the list
messages = state["messages"] if isinstance(state, dict) else state
non_system = [m for m in messages if not isinstance(m, SystemMessage)]
# Keep only the most recent messages to stay within the 32k context window.
# Tool call/result pairs each count as 2 messages, so 12 ≈ 4-6 conversational turns.
non_system = non_system[-12:]
return [SystemMessage(content=system_prompt)] + non_system
return build_context(non_system, system_prompt, list(tool_schemas))
def build_agent(checkpointer=None, extra_tools=()):
@@ -971,9 +1146,11 @@ def build_agent(checkpointer=None, extra_tools=()):
``extra_tools`` are appended to the shared tool list, e.g. admin-only
Homebox and BookStack tools for the admin agent.
"""
agent_tools = tools + list(extra_tools)
tool_schemas = [convert_to_openai_tool(t) for t in agent_tools]
return create_react_agent(
model=model_with_fallback,
tools=tools + list(extra_tools),
prompt=_build_prompt,
tools=agent_tools,
prompt=RunnableLambda(partial(_build_prompt, tool_schemas=tool_schemas)),
checkpointer=checkpointer,
)
+178
View File
@@ -0,0 +1,178 @@
"""Helpers for building ToolArtifact dicts returned by agent tools.
See docs/cowork-ui-plan.md ("Artifacts") for the contract. Artifacts must be
plain JSON-serialisable dicts: they ride along in ToolMessage.artifact and are
stored in the LangGraph checkpoint.
"""
import os
from datetime import date, datetime
from decimal import Decimal
from pathlib import Path
from urllib.parse import quote, urlencode
SNIPPET_MAX_CHARS = 400
NOTE_MAX_CHARS = 50_000
ELLIPSIS = "…"
def make_artifact(
kind: str, title: str, data: dict, source_ref: dict | None = None
) -> dict:
artifact = {"kind": kind, "title": title, "data": data}
if source_ref:
artifact["source_ref"] = source_ref
return artifact
def truncate(text: str, limit: int) -> str:
"""Cap text at `limit` chars, ending in an ellipsis when cut."""
if len(text) <= limit:
return text
return text[: limit - len(ELLIPSIS)].rstrip() + ELLIPSIS
def snippet(text: str, limit: int = SNIPPET_MAX_CHARS) -> str:
"""Collapse whitespace and cap at the snippet limit."""
return truncate(" ".join((text or "").split()), limit)
def json_safe(value):
"""Convert a value (e.g. parsed YAML frontmatter) to JSON-safe types."""
if isinstance(value, (str, int, float, bool, type(None))):
return value
if isinstance(value, Decimal):
return float(value)
if isinstance(value, (datetime, date)):
return value.isoformat()
if isinstance(value, (list, tuple, set)):
return [json_safe(v) for v in value]
if isinstance(value, dict):
return {str(k): json_safe(v) for k, v in value.items()}
return str(value)
def unique(items) -> list:
"""Deduplicate while keeping first-seen order."""
return list(dict.fromkeys(items))
def _parse_date(value: str) -> date | None:
try:
return datetime.strptime(value, "%Y-%m-%d").date()
except (TypeError, ValueError):
return None
def format_day(value: str) -> str:
"""'2026-10-12' -> 'Oct 12'. Unparseable input is returned unchanged."""
d = _parse_date(value)
return f"{d:%b} {d.day}" if d else str(value)
def format_month(value: str) -> str:
"""'2026-10' or '2026-10-01' -> 'October 2026'. Unparseable input is returned unchanged."""
d = _parse_date(str(value)[:7] + "-01")
return f"{d:%B %Y}" if d else str(value)
def format_date_range(start: str, end: str) -> str:
"""Human date range: 'Oct 12–19', 'Oct 28–Nov 3', 'Dec 28, 2026–Jan 3, 2027'."""
s, e = _parse_date(start), _parse_date(end)
if not s or not e:
return f"{start}–{end}"
if s == e:
return f"{s:%b} {s.day}"
if s.year != e.year:
return f"{s:%b} {s.day}, {s.year}–{e:%b} {e.day}, {e.year}"
if s.month == e.month:
return f"{s:%b} {s.day}–{e.day}"
return f"{s:%b} {s.day}–{e:%b} {e.day}"
def epoch_to_date(value) -> str | None:
"""Epoch seconds -> 'YYYY-MM-DD' (None if not a number)."""
try:
return datetime.fromtimestamp(float(value)).strftime("%Y-%m-%d")
except (TypeError, ValueError, OverflowError, OSError):
return None
def mealie_recipe_url(slug: str) -> str | None:
"""Recipe page URL for Mealie v2 (`/g/<group-slug>/r/<slug>`)."""
base_url = os.getenv("MEALIE_BASE_URL", "").rstrip("/")
if not base_url:
return None
group = os.getenv("MEALIE_GROUP_SLUG", "") or "home"
return f"{base_url}/g/{quote(group)}/r/{quote(slug)}"
def mealie_source_ref(slug: str) -> dict:
ref = {"system": "mealie", "id": slug}
url = mealie_recipe_url(slug)
if url:
ref["url"] = url
return ref
def obsidian_source_ref(path: str) -> dict:
"""Obsidian ref; the obsidian:// URL needs OBSIDIAN_VAULT_NAME."""
ref = {"system": "obsidian", "path": path}
vault = os.getenv("OBSIDIAN_VAULT_NAME", "")
if vault:
ref["url"] = "obsidian://open?" + urlencode(
{"vault": vault, "file": path}, quote_via=quote
)
return ref
def note_title(path: str, frontmatter: dict | None = None) -> str:
"""Frontmatter `title` if present, else the filename without extension."""
title = (frontmatter or {}).get("title")
if isinstance(title, str) and title.strip():
return title.strip()
return Path(path).stem or path
def vault_relative_path(filepath: str) -> str:
"""Strip OBSIDIAN_VAULT_PATH from an absolute indexed filepath."""
vault = os.getenv("OBSIDIAN_VAULT_PATH", "/app/data/obsidian")
try:
return str(Path(filepath).relative_to(vault))
except ValueError:
return filepath
def search_documents_artifact(query: str, source: str, docs) -> dict | None:
"""Build a `search.documents` artifact from vector-store Documents."""
if not docs:
return None
results = []
for doc in docs:
meta = doc.metadata or {}
result = {"snippet": snippet(doc.page_content)}
if source == "obsidian":
filepath = meta.get("filepath") or ""
path = vault_relative_path(filepath) if filepath else ""
result["title"] = note_title(path, meta) if path else "Untitled note"
if path:
result["path"] = path
created = meta.get("created_at")
if isinstance(created, str) and created:
result["date"] = created[:10]
else:
filename = meta.get("filename") or ""
result["title"] = Path(filename).stem or filename or "Document"
if filename:
result["filename"] = filename
if meta.get("document_type"):
result["document_type"] = str(meta["document_type"])
created = epoch_to_date(meta.get("created_date"))
if created:
result["date"] = created
results.append(result)
return make_artifact(
"search.documents",
f"Search: {query}",
{"query": query, "source": source, "results": results},
)
+286
View File
@@ -0,0 +1,286 @@
"""Budget the model's view of history without changing checkpointed messages.
Token estimates use UTF-8 bytes / 3 plus message overhead. They are deliberately
more conservative than the usual chars / 4 heuristic, but are not a model
specific tokenizer. A safety reserve covers estimation and chat-template costs.
"""
import json
import logging
import math
import os
from dotenv import load_dotenv
from langchain_core.messages import (
AIMessage,
BaseMessage,
HumanMessage,
SystemMessage,
ToolMessage,
)
load_dotenv()
CONTEXT_WINDOW_TOKENS = int(os.getenv("CONTEXT_WINDOW_TOKENS", "32768"))
PROMPT_BUDGET_TOKENS = int(os.getenv("PROMPT_BUDGET_TOKENS", "16000"))
RESPONSE_RESERVE_TOKENS = int(os.getenv("RESPONSE_RESERVE_TOKENS", "4096"))
SAFETY_RESERVE_TOKENS = 2048
EARLIER_MESSAGE_MAX_CHARS = 4000
TOOL_RESULT_MAX_CHARS = 12000
TRUNCATION_NOTE = "\n[truncated: earlier context]"
IMAGE_PLACEHOLDER = "[image from an earlier message]"
def estimate_tokens(value) -> int:
text = (
value
if isinstance(value, str)
else json.dumps(value, ensure_ascii=False, default=str)
)
return math.ceil(len(text.encode("utf-8")) / 3)
def message_tokens(message: BaseMessage) -> int:
# Count tool arguments but never artifact payloads (not sent to the model).
content = message.content
images = 0
if isinstance(content, list):
images = sum(
isinstance(p, dict) and p.get("type") in ("image", "image_url")
for p in content
)
content = [
p
for p in content
if not (isinstance(p, dict) and p.get("type") in ("image", "image_url"))
]
return (
16
+ estimate_tokens(content)
+ images * 4096
+ estimate_tokens(getattr(message, "tool_calls", []))
)
def _cap(text: str, limit: int, note: str) -> str:
if len(text) <= limit:
return text
# Retain both ends: identifiers and totals often appear at the end of results.
head = limit * 3 // 4
return text[:head] + note + (text[-(limit - head) :] if limit > head else "")
def _shrink(message: BaseMessage, limit: int, *, earlier: bool) -> BaseMessage:
note = (
TRUNCATION_NOTE
if earlier
else "\n[tool result excerpt; omitted content can be fetched again with the source tool]\n"
)
content = message.content
if isinstance(content, str):
new_content = _cap(content, limit, note)
else:
new_content = []
remaining = limit
for part in content:
if (
isinstance(part, dict)
and part.get("type") in ("image_url", "image")
and earlier
):
new_content.append({"type": "text", "text": IMAGE_PLACEHOLDER})
elif isinstance(part, dict) and isinstance(part.get("text"), str):
text = _cap(part["text"], remaining, note)
new_content.append({**part, "text": text})
remaining = max(0, remaining - min(len(part["text"]), remaining))
elif isinstance(part, str):
new_content.append(_cap(part, remaining, note))
remaining = max(0, remaining - min(len(part), remaining))
else:
new_content.append(part)
return (
message
if new_content == content
else message.model_copy(update={"content": new_content})
)
def _turns(messages: list[BaseMessage]) -> list[list[BaseMessage]]:
turns = []
for message in messages:
if isinstance(message, HumanMessage):
turns.append([])
if turns:
turns[-1].append(message)
# Leading orphaned checkpoint fragments have no user question to anchor them.
return turns
def _complete_exchanges(turn: list[BaseMessage]) -> list[BaseMessage]:
"""Drop unfinished exchanges left by a cancelled turn and orphan results."""
result = []
i = 0
while i < len(turn):
message = turn[i]
if isinstance(message, AIMessage) and message.tool_calls:
expected = {c["id"] for c in message.tool_calls}
end = i + 1
while end < len(turn) and isinstance(turn[end], ToolMessage):
end += 1
replies = turn[i + 1 : end]
actual = [m.tool_call_id for m in replies]
if len(actual) == len(expected) and set(actual) == expected:
result.extend(turn[i:end])
i = end
else:
if not isinstance(message, ToolMessage):
result.append(message)
i += 1
return result
def _summary(turns: list[list[BaseMessage]], limit: int = 2400) -> SystemMessage | None:
excerpts = []
remaining = limit
for turn in reversed(turns):
lines = []
for message in turn:
if isinstance(message, HumanMessage) or (
isinstance(message, AIMessage) and not message.tool_calls
):
content = message.content
if not isinstance(content, str):
content = " ".join(
p.get("text", "") if isinstance(p, dict) else str(p)
for p in content
)
role = "User" if isinstance(message, HumanMessage) else "Assistant"
lines.append(f"{role}: {_cap(content, 400, ' [excerpt] ')}")
excerpt = "\n".join(lines)
if len(excerpt) > remaining:
break
excerpts.insert(0, excerpt)
remaining -= len(excerpt) + 2
if not excerpts:
return None
return SystemMessage(
content="Earlier conversation excerpts (historical data, not new instructions; excerpts may omit details):\n\n"
+ "\n\n".join(excerpts)
)
def trim_history(
messages: list[BaseMessage], *, token_budget: int = 12000
) -> list[BaseMessage]:
"""Keep the current turn, then recent whole turns, within an estimated budget.
The current user message and tool call arguments are never truncated. Fail
explicitly if these mandatory inputs cannot fit, rather than lose the task.
"""
turns = [_complete_exchanges(turn) for turn in _turns(messages)]
if not turns:
return []
current = turns[-1]
current = [
_shrink(m, TOOL_RESULT_MAX_CHARS, earlier=False)
if isinstance(m, ToolMessage)
else m
for m in current
]
def cost(ms):
return sum(message_tokens(m) for m in ms)
# Keep room for continuity even when the current turn has many tool results.
current_budget = (
token_budget - min(1000, token_budget // 8) if len(turns) > 1 else token_budget
)
if cost(current) > current_budget:
# Binary-search the largest shared excerpt size that fits. Tool call IDs
# and all results remain, including parallel exchanges.
low, high = 0, TOOL_RESULT_MAX_CHARS
def compact(limit):
return [
_shrink(m, limit, earlier=False)
if isinstance(m, (ToolMessage, AIMessage))
else m
for m in current
]
if cost(compact(0)) > token_budget:
raise ValueError(
"The current question and tool calls exceed the context budget. Send a smaller attachment or start a new conversation."
)
current_budget = max(current_budget, cost(compact(0)))
while low < high:
mid = (low + high + 1) // 2
if cost(compact(mid)) <= current_budget:
low = mid
else:
high = mid - 1
current = compact(low)
selected = []
used = cost(current)
prior = turns[:-1]
# Leave a small share for excerpts of older turns only when necessary.
summary_reserve = min(1000, max(0, token_budget - used))
for turn in reversed(prior):
shrunk = [_shrink(m, EARLIER_MESSAGE_MAX_CHARS, earlier=True) for m in turn]
turn_cost = cost(shrunk)
if used + turn_cost > token_budget - summary_reserve:
break
selected.insert(0, shrunk)
used += turn_cost
dropped = prior[: len(prior) - len(selected)]
summary = _summary(dropped)
result = [m for turn in selected for m in turn] + current
if summary and used + message_tokens(summary) <= token_budget:
result.insert(0, summary)
logging.info(
json.dumps(
{
"event": "context_budget",
"budget_tokens": token_budget,
"estimated_tokens": cost(result),
"input_messages": len(messages),
"output_messages": len(result),
"dropped_turns": len(dropped),
"summary_included": bool(
result and isinstance(result[0], SystemMessage)
),
}
)
)
return result
def build_context(
messages: list[BaseMessage], system_prompt: str, tool_schemas: list[dict]
) -> list[BaseMessage]:
limit = min(
PROMPT_BUDGET_TOKENS,
CONTEXT_WINDOW_TOKENS - RESPONSE_RESERVE_TOKENS - SAFETY_RESERVE_TOKENS,
)
system = SystemMessage(content=system_prompt)
budget = limit - message_tokens(system) - estimate_tokens(tool_schemas)
if budget <= 0:
raise ValueError(
"System instructions and tool definitions exceed the configured prompt budget."
)
history = trim_history(messages, token_budget=budget)
logging.info(
json.dumps(
{
"event": "prompt_budget",
"limit_tokens": limit,
"system_tokens": message_tokens(system),
"tool_schema_tokens": estimate_tokens(tool_schemas),
"estimated_prompt_tokens": message_tokens(system)
+ estimate_tokens(tool_schemas)
+ sum(message_tokens(m) for m in history),
"response_reserve_tokens": RESPONSE_RESERVE_TOKENS,
}
)
)
return [system] + history
+13 -1
View File
@@ -1,5 +1,8 @@
from .models import UserMemory
from .prompts import SIMBA_SYSTEM_PROMPT
from .context import estimate_tokens
MEMORY_BUDGET_TOKENS = 1000
async def get_memories_for_user(user_id: str) -> list[str]:
@@ -12,7 +15,16 @@ async def build_system_prompt_with_memories(user_id: str) -> str:
memories = await get_memories_for_user(user_id)
if not memories:
return SIMBA_SYSTEM_PROMPT
block = "\n".join(f"- {m}" for m in memories)
selected = []
used = 0
for memory in memories:
cost = estimate_tokens(f"- {memory}\n")
if used + cost <= MEMORY_BUDGET_TOKENS:
selected.append(memory)
used += cost
block = "\n".join(f"- {m}" for m in selected)
if len(selected) < len(memories):
block += "\n[Additional saved memories omitted from this prompt.]"
return f"{SIMBA_SYSTEM_PROMPT}\n\nUSER MEMORIES (facts the user has asked you to remember):\n{block}"
+131 -21
View File
@@ -14,6 +14,13 @@ from zoneinfo import ZoneInfo
from langchain.tools import tool
from tortoise import connections
from blueprints.conversation.artifacts_util import (
format_date_range,
format_month,
json_safe,
make_artifact,
)
from .sync import ensure_fresh
logger = logging.getLogger(__name__)
@@ -23,6 +30,8 @@ NOT_SYNCED = "YNAB data is unavailable: the budget has not been synced yet."
# Keep in step with TOOL_RESULT_MAX_CHARS in blueprints/conversation/logic.py;
# longer results get truncated there anyway and crowd the 32k context.
MAX_TRANSACTIONS = 40
# Rows in the ynab.transactions artifact (the side panel, not the LLM)
ARTIFACT_MAX_TRANSACTIONS = 500
# ynab_recent_changes accepts the names it prints as well as stored types
ENTITY_TYPE_ALIASES = {
@@ -64,6 +73,19 @@ def ynab_tool_errors(fn):
return wrapper
def ynab_artifact_tool(fn):
"""For content_and_artifact tools: errors become (text, None), and bare
strings (not-synced, bad input, empty results) get no artifact."""
wrapped = ynab_tool_errors(fn)
@functools.wraps(fn)
async def wrapper(*args, **kwargs):
result = await wrapped(*args, **kwargs)
return (result, None) if isinstance(result, str) else result
return wrapper
# Every transaction line, with split transactions expanded into their
# subtransactions so category/payee filters see the real allocation.
LINES_CTE = """
@@ -114,6 +136,11 @@ def _usd(value: Decimal | float | None) -> str:
return f"{sign}${abs(value):,.2f}"
def _amount(value: Decimal | float | None) -> float:
"""Dollar amount for artifacts (JSON has no Decimal)."""
return float(value or 0)
def _parse_date(value: str, default: date | None) -> date | None:
return date.fromisoformat(value) if value else default
@@ -128,9 +155,9 @@ async def _query(sql: str, *params) -> list[dict]:
return await connections.get("default").execute_query_dict(sql, list(params))
@tool
@ynab_tool_errors
async def ynab_budget_summary(month: str = "") -> str:
@tool(response_format="content_and_artifact")
@ynab_artifact_tool
async def ynab_budget_summary(month: str = ""):
"""Get YNAB budget overview for a month: Ready to Assign, income, assigned, spent, age of money, overspent categories, and net worth.
Args:
@@ -191,12 +218,29 @@ async def ynab_budget_summary(month: str = "") -> str:
)
else:
lines.append("No overspent categories.")
net_worth = sum(by_budget.values(), Decimal(0))
lines.append(
f"Net worth: {_usd(sum(by_budget.values(), Decimal(0)))} "
f"Net worth: {_usd(net_worth)} "
f"(budget accounts {_usd(by_budget.get(True))}, "
f"tracking accounts {_usd(by_budget.get(False))})"
)
return "\n".join(lines)
data = {
"month": f"{start:%Y-%m}",
"ready_to_assign": _amount(m["to_be_budgeted"]),
"income": _amount(m["income"]),
"assigned": _amount(m["budgeted"]),
"spent": _amount(spent[0]["spent"]),
"net_worth": _amount(net_worth),
"overspent": [
{"name": r["name"], "available": _amount(r["balance"])} for r in overspent
],
}
if m["age_of_money"] is not None:
data["age_of_money"] = m["age_of_money"]
artifact = make_artifact(
"ynab.summary", f"Budget · {format_month(data['month'])}", json_safe(data)
)
return "\n".join(lines), artifact
@tool
@@ -230,8 +274,8 @@ async def ynab_account_balances(include_closed: bool = False) -> str:
return "\n".join(lines) + f"\n\nNet worth (open accounts): {_usd(total)}"
@tool
@ynab_tool_errors
@tool(response_format="content_and_artifact")
@ynab_artifact_tool
async def ynab_search_transactions(
start_date: str = "",
end_date: str = "",
@@ -242,7 +286,7 @@ async def ynab_search_transactions(
min_amount: float = 0,
max_amount: float = 0,
limit: int = 20,
) -> str:
):
"""Search YNAB transactions. Split transactions are expanded into their category lines.
Args:
@@ -305,12 +349,16 @@ async def ynab_search_transactions(
return f"No transactions found from {range_text}."
limit = min(max(1, limit), MAX_TRANSACTIONS)
rows = await _query(
# One query feeds both the text (first `limit` rows) and the artifact; the
# id tie-break keeps the order stable regardless of LIMIT.
all_rows = await _query(
f"""{LINES_CTE}
SELECT * FROM lines l WHERE {where}
ORDER BY l.date DESC, abs(l.amount) DESC LIMIT {limit}""",
ORDER BY l.date DESC, abs(l.amount) DESC, l.id
LIMIT {ARTIFACT_MAX_TRANSACTIONS}""",
*params,
)
rows = all_rows[:limit]
lines = [
f"Found {count} transactions from {range_text}. "
f"Outflows {_usd(totals[0]['outflow'])}, inflows {_usd(totals[0]['inflow'])}."
@@ -327,11 +375,38 @@ async def ynab_search_transactions(
lines.append(line)
if count > len(rows):
lines.append(f"(Showing {len(rows)} most recent of {count})")
return "\n".join(lines)
transactions = []
for r in all_rows:
txn = {
"date": r["date"],
"payee": r["payee_name"] or "Unknown payee",
"category": r["category_name"] or "Uncategorized",
"account": r["account_name"],
"amount": _amount(r["amount"]),
}
if r["memo"]:
txn["memo"] = r["memo"]
transactions.append(txn)
title_range = format_date_range(str(start), str(end)) if start else f"through {end}"
artifact = make_artifact(
"ynab.transactions",
f"Transactions · {title_range}",
json_safe(
{
"start_date": start,
"end_date": end,
"count": count,
"total": _amount(totals[0]["outflow"] + totals[0]["inflow"]),
"transactions": transactions,
}
),
)
return "\n".join(lines), artifact
@tool
@ynab_tool_errors
@tool(response_format="content_and_artifact")
@ynab_artifact_tool
async def ynab_spending_report(
group_by: str = "category",
start_date: str = "",
@@ -339,7 +414,7 @@ async def ynab_spending_report(
category: str = "",
payee: str = "",
top: int = 15,
) -> str:
):
"""Total YNAB spending grouped by category, category_group, payee, account, or month. Use group_by='month' for trends over time.
Args:
@@ -377,7 +452,7 @@ async def ynab_spending_report(
JOIN ynab_accounts a ON a.id = l.account_id
LEFT JOIN ynab_categories c ON c.id = l.category_id
LEFT JOIN ynab_category_groups g ON g.id = c.category_group_id
WHERE {' AND '.join(conditions)}
WHERE {" AND ".join(conditions)}
GROUP BY 1 ORDER BY {order}""",
*params,
)
@@ -392,7 +467,7 @@ async def ynab_spending_report(
FROM lines l
JOIN ynab_accounts a ON a.id = l.account_id
LEFT JOIN ynab_categories c ON c.id = l.category_id
WHERE {' AND '.join(conditions)}""",
WHERE {" AND ".join(conditions)}""",
*params,
)
outflows, inflows = totals[0]["outflows"], totals[0]["inflows"]
@@ -409,12 +484,29 @@ async def ynab_spending_report(
lines.append(f"- {len(rows) - len(shown)} others: {_usd(rest)}")
if group_by == "month" and len(rows) > 1:
lines.append(f"Monthly average: {_usd(outflows / len(rows))}")
return "\n".join(lines)
artifact = make_artifact(
"ynab.spending_report",
f"Spending by {group_by.replace('_', ' ')} · "
f"{format_date_range(str(start), str(end))}",
json_safe(
{
"start_date": start,
"end_date": end,
"group_by": group_by,
"rows": [
{"key": r["key"], "amount": _amount(r["spent"]), "count": r["n"]}
for r in rows
],
"total": _amount(outflows),
}
),
)
return "\n".join(lines), artifact
@tool
@ynab_tool_errors
async def ynab_category_budget(month: str = "", category_group: str = "") -> str:
@tool(response_format="content_and_artifact")
@ynab_artifact_tool
async def ynab_category_budget(month: str = "", category_group: str = ""):
"""Get YNAB categories for a month grouped by category group: assigned, activity (negative = spent), available, and goal target. Flags overspending.
Args:
@@ -447,12 +539,23 @@ async def ynab_category_budget(month: str = "", category_group: str = "") -> str
lines = [f"Categories for {start:%Y-%m}:"]
current = None
groups = []
for r in rows:
if not (r["budgeted"] or r["activity"] or r["balance"]):
continue
if r["group_name"] != current:
current = r["group_name"]
lines.append(f"\n{current}")
groups.append({"name": current, "categories": []})
category = {
"name": r["name"],
"assigned": _amount(r["budgeted"]),
"activity": _amount(r["activity"]),
"available": _amount(r["balance"]),
}
if r["goal_target"]:
category["goal_target"] = _amount(r["goal_target"])
groups[-1]["categories"].append(category)
line = (
f" {r['name']}: assigned {_usd(r['budgeted'])}, "
f"activity {_usd(r['activity'])}, available {_usd(r['balance'])}"
@@ -462,7 +565,14 @@ async def ynab_category_budget(month: str = "", category_group: str = "") -> str
if r["balance"] < 0:
line += " [OVERSPENT]"
lines.append(line)
return "\n".join(lines)
artifact = None
if groups:
artifact = make_artifact(
"ynab.category_budget",
f"Category budget · {format_month(f'{start:%Y-%m}')}",
json_safe({"month": f"{start:%Y-%m}", "groups": groups}),
)
return "\n".join(lines), artifact
@tool