ryan/simbarag
1
0
Fork 0
Code Issues Pull Requests 4 Actions Packages Projects Releases Wiki Activity

Cowork B: Mealie, Obsidian and search tools return artifacts #61

Merged
ryan merged 2 commits from feat/cowork-tool-artifacts into main 2026-10-12 18:31:35 -04:00
Conversation 1 Commits 2 Files Changed 7
+1520 -61
7 changed files with 1520 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+4
View File
@@ -70,6 +70,8 @@ YNAB_SYNC_INTERVAL=900
MEALIE_BASE_URL=http://192.168.1.5:9000
# Get your API token from Mealie's user settings page
MEALIE_API_TOKEN=your-mealie-api-token
# Group slug used in recipe page links (/g/<slug>/r/<recipe>); defaults to "home"
MEALIE_GROUP_SLUG=home
# Email Integration
# Email Encryption Key (32-byte URL-safe base64)
@@ -167,6 +169,8 @@ OBSIDIAN_DEVICE_NAME=simbarag
OBSIDIAN_CONTINUOUS_SYNC=false
# Local path to Obsidian vault (where files are synced)
OBSIDIAN_VAULT_PATH=/app/data/obsidian
# Vault name for obsidian://open links in the artifact panel (optional; no link if unset)
OBSIDIAN_VAULT_NAME=
# LangGraph checkpoint store (SQLite). Must be on a persistent volume.
# Both blue and green containers share this volume via docker-compose.
blueprints/conversation/agents.py
+188 -41
View File
@@ -11,6 +11,19 @@ from langchain_core.runnables import RunnableConfig, RunnableLambda
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.memory import save_memory
from blueprints.conversation.prompts import SIMBA_SYSTEM_PROMPT
from blueprints.rag.logic import query_vector_store
@@ -131,20 +144,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 +181,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 +211,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 +220,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 +236,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 +296,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 +357,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 +424,55 @@ 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:
@tool(response_format="content_and_artifact")
async def obsidian_read_note(relative_path: str):
"""Read the full content of a specific Obsidian note by path.
Args:
relative_path: Path relative to vault root (e.g., "notes/my-note.md")
"""
if not obsidian_enabled:
return "Obsidian integration is not configured. Please set OBSIDIAN_VAULT_PATH environment variable."
return OBSIDIAN_NOT_CONFIGURED, None
try:
note = obsidian_service.read_note(relative_path)
@@ -348,12 +484,15 @@ async def obsidian_read_note(relative_path: str) -> str:
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 +530,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
blueprints/conversation/artifacts_util.py
+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},
)
blueprints/ynab/tools.py
+130 -20
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,
)
@@ -394,12 +469,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(total / 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(total),
}
),
)
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:
@@ -432,12 +524,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'])}"
@@ -447,7 +550,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
docker-compose.yml
+2
View File
@@ -27,6 +27,7 @@ x-raggr-common: &raggr-common
- YNAB_SYNC_INTERVAL=${YNAB_SYNC_INTERVAL:-900}
- MEALIE_BASE_URL=${MEALIE_BASE_URL}
- MEALIE_API_TOKEN=${MEALIE_API_TOKEN}
- MEALIE_GROUP_SLUG=${MEALIE_GROUP_SLUG:-home}
- TWILIO_ACCOUNT_SID=${TWILIO_ACCOUNT_SID}
- TWILIO_AUTH_TOKEN=${TWILIO_AUTH_TOKEN}
- TWILIO_WHATSAPP_NUMBER=${TWILIO_WHATSAPP_NUMBER}
@@ -40,6 +41,7 @@ x-raggr-common: &raggr-common
- OBSIDIAN_DEVICE_NAME=${OBSIDIAN_DEVICE_NAME}
- OBSIDIAN_CONTINUOUS_SYNC=${OBSIDIAN_CONTINUOUS_SYNC:-false}
- OBSIDIAN_VAULT_PATH=${OBSIDIAN_VAULT_PATH:-/app/data/obsidian}
- OBSIDIAN_VAULT_NAME=${OBSIDIAN_VAULT_NAME:-}
- S3_ENDPOINT_URL=${S3_ENDPOINT_URL}
- S3_ACCESS_KEY_ID=${S3_ACCESS_KEY_ID}
- S3_SECRET_ACCESS_KEY=${S3_SECRET_ACCESS_KEY}
tests/unit/test_tool_artifacts.py
+591
View File
@@ -0,0 +1,591 @@
"""Tests for tools returning (text, artifact) via content_and_artifact.
The text half must stay identical to the pre-artifact output (the LLM's view),
and the artifact must match the contract in docs/cowork-ui-plan.md.
"""
import datetime
import json
import os
from unittest.mock import AsyncMock, MagicMock
import pytest
from langchain_core.documents import Document
from langchain_core.messages import ToolMessage
os.environ.setdefault("TAVILY_API_KEY", "dummy")
from blueprints.conversation import agents # noqa: E402
from blueprints.conversation.artifacts_util import ( # noqa: E402
NOTE_MAX_CHARS,
SNIPPET_MAX_CHARS,
format_date_range,
search_documents_artifact,
)
from utils.obsidian_service import ObsidianService # noqa: E402
KIND_DATA_KEYS = {
"mealie.recipe": {
"slug",
"name",
"description",
"servings",
"prep_time",
"cook_time",
"total_time",
"ingredients",
"instructions",
"tags",
},
"mealie.meal_plan": {"start_date", "end_date", "days"},
"mealie.shopping_lists": {"lists"},
"obsidian.note": {"path", "title", "markdown", "tags", "frontmatter"},
"search.documents": {"query", "source", "results"},
}
async def call(tool, args=None) -> ToolMessage:
"""Invoke a tool the way the agent does: with a ToolCall dict."""
msg = await tool.ainvoke(
{"type": "tool_call", "id": "call-1", "name": tool.name, "args": args or {}}
)
assert isinstance(msg, ToolMessage)
return msg
def assert_artifact(artifact, kind):
assert artifact is not None
assert artifact["kind"] == kind
assert isinstance(artifact["title"], str) and artifact["title"]
assert set(artifact["data"]) == KIND_DATA_KEYS[kind]
json.dumps(artifact)
@pytest.fixture
def mealie(monkeypatch):
service = MagicMock()
monkeypatch.setattr(agents, "mealie_enabled", True)
monkeypatch.setattr(agents, "mealie_service", service, raising=False)
monkeypatch.setenv("MEALIE_BASE_URL", "http://mealie.local:9000/")
monkeypatch.delenv("MEALIE_GROUP_SLUG", raising=False)
return service
@pytest.fixture
def obsidian(monkeypatch, tmp_path):
(tmp_path / "seed.md").write_text("# seed")
monkeypatch.setenv("OBSIDIAN_VAULT_PATH", str(tmp_path))
monkeypatch.delenv("OBSIDIAN_VAULT_NAME", raising=False)
service = ObsidianService()
monkeypatch.setattr(agents, "obsidian_enabled", True)
monkeypatch.setattr(agents, "obsidian_service", service, raising=False)
return service
# --- Mealie -----------------------------------------------------------------
RECIPE = {
"name": "Chicken Tikka Masala",
"slug": "chicken-tikka-masala",
"description": "Creamy and spiced.",
"ingredients": [{"display": "1 lb chicken"}, {"display": "1 cup yogurt"}],
"instructions": [
{"title": "Marinate", "text": "Mix chicken and yogurt."},
{"title": "", "text": "Cook it."},
],
"prep_time": "20 minutes",
"cook_time": "30 minutes",
"total_time": "",
"servings": "4",
"tags": ["Indian", "Dinner"],
}
RECIPE_TEXT = """**Chicken Tikka Masala**
Creamy and spiced.
Servings: 4
Prep: 20 minutes | Cook: 30 minutes
Ingredients:
- 1 lb chicken
- 1 cup yogurt
Instructions:
1. Mix chicken and yogurt.
2. Cook it.
Tags: Indian, Dinner"""
async def test_mealie_get_recipe(mealie):
mealie.get_recipe = AsyncMock(return_value=RECIPE)
msg = await call(agents.mealie_get_recipe, {"recipe_slug": "chicken-tikka-masala"})
assert msg.content == RECIPE_TEXT
art = msg.artifact
assert_artifact(art, "mealie.recipe")
assert art["title"] == "Chicken Tikka Masala"
assert art["data"]["ingredients"] == ["1 lb chicken", "1 cup yogurt"]
assert art["data"]["instructions"] == [
{"title": "Marinate", "text": "Mix chicken and yogurt."},
{"text": "Cook it."},
]
assert art["source_ref"] == {
"system": "mealie",
"id": "chicken-tikka-masala",
"url": "http://mealie.local:9000/g/home/r/chicken-tikka-masala",
}
async def test_mealie_recipe_url_uses_group_slug(mealie, monkeypatch):
monkeypatch.setenv("MEALIE_GROUP_SLUG", "family")
mealie.get_recipe = AsyncMock(return_value=RECIPE)
msg = await call(agents.mealie_get_recipe, {"recipe_slug": "chicken-tikka-masala"})
assert msg.artifact["source_ref"]["url"].endswith(
"/g/family/r/chicken-tikka-masala"
)
async def test_mealie_get_recipe_error(mealie):
mealie.get_recipe = AsyncMock(side_effect=Exception("404"))
msg = await call(agents.mealie_get_recipe, {"recipe_slug": "nope"})
assert msg.content == "Error fetching recipe: 404"
assert msg.artifact is None
MEAL_PLANS = {
"start_date": "2026-10-12",
"end_date": "2026-10-19",
"total_plans": 3,
"plans_by_date": {
"2026-10-13": [
{
"entry_type": "dinner",
"title": "",
"recipe_name": "Tacos",
"recipe_slug": "tacos",
"note": "",
}
],
"2026-10-12": [
{
"entry_type": "lunch",
"title": "Leftovers",
"recipe_name": "",
"recipe_slug": "",
"note": "from Sunday",
},
{
"entry_type": "dinner",
"title": "",
"recipe_name": "Chicken Tikka Masala",
"recipe_slug": "chicken-tikka-masala",
"note": "",
},
],
},
}
MEAL_PLANS_TEXT = """Meal plans from 2026-10-12 to 2026-10-19 (3 meals):
2026-10-12:
- Lunch: Leftovers
Note: from Sunday
- Dinner: Chicken Tikka Masala
2026-10-13:
- Dinner: Tacos"""
async def test_mealie_meal_plans(mealie):
mealie.get_meal_plans = AsyncMock(return_value=MEAL_PLANS)
msg = await call(agents.mealie_meal_plans, {})
assert msg.content == MEAL_PLANS_TEXT
art = msg.artifact
assert_artifact(art, "mealie.meal_plan")
assert art["title"] == "Meal plan · Oct 12–19"
assert [d["date"] for d in art["data"]["days"]] == ["2026-10-12", "2026-10-13"]
assert art["data"]["days"][0]["meals"] == [
{"entry_type": "lunch", "name": "Leftovers", "note": "from Sunday"},
{
"entry_type": "dinner",
"name": "Chicken Tikka Masala",
"recipe_slug": "chicken-tikka-masala",
},
]
async def test_mealie_meal_plans_empty(mealie):
mealie.get_meal_plans = AsyncMock(
return_value={
"start_date": "2026-10-12",
"end_date": "2026-10-19",
"total_plans": 0,
"plans_by_date": {},
}
)
msg = await call(agents.mealie_meal_plans, {})
assert msg.content == "No meal plans found from 2026-10-12 to 2026-10-19."
assert msg.artifact is None
async def test_mealie_todays_meals(mealie):
mealie.get_todays_meals = AsyncMock(
return_value={
"today": "2026-10-12",
"total_meals": 1,
"meals": [
{
"entry_type": "dinner",
"title": "",
"recipe_name": "Tacos",
"recipe_slug": "tacos",
"note": "extra salsa",
}
],
}
)
msg = await call(agents.mealie_todays_meals)
assert msg.content == (
"Meals planned for today (2026-10-12):\n- Dinner: Tacos\n Note: extra salsa"
)
art = msg.artifact
assert_artifact(art, "mealie.meal_plan")
assert art["title"] == "Today's meals · Oct 12"
assert art["data"] == {
"start_date": "2026-10-12",
"end_date": "2026-10-12",
"days": [
{
"date": "2026-10-12",
"meals": [
{
"entry_type": "dinner",
"name": "Tacos",
"recipe_slug": "tacos",
"note": "extra salsa",
}
],
}
],
}
async def test_mealie_todays_meals_empty(mealie):
mealie.get_todays_meals = AsyncMock(
return_value={"today": "2026-10-12", "total_meals": 0, "meals": []}
)
msg = await call(agents.mealie_todays_meals)
assert msg.content == "No meals planned for today (2026-10-12)."
assert msg.artifact is None
async def test_mealie_shopping_lists(mealie):
mealie.get_shopping_lists = AsyncMock(
return_value={
"total_lists": 1,
"total_unchecked_items": 2,
"shopping_lists": [
{
"name": "Groceries",
"unchecked_items": [
{"name": "Milk", "quantity": 1, "note": "", "food": None},
{"name": "Eggs", "quantity": 12.0, "note": "", "food": None},
],
"checked_items": [
{"name": "Bread", "quantity": 1, "note": "", "food": None}
],
}
],
}
)
msg = await call(agents.mealie_shopping_lists)
assert msg.content == (
"Shopping lists (2 items needed):\n"
"\n**Groceries** (2 items needed)\n"
" - [ ] Milk\n"
" - [ ] Eggs x12.0\n"
" (1 items already checked off)"
)
art = msg.artifact
assert_artifact(art, "mealie.shopping_lists")
assert art["data"]["lists"] == [
{
"name": "Groceries",
"items": [
{"name": "Milk", "quantity": 1, "checked": False},
{"name": "Eggs", "quantity": 12.0, "checked": False},
{"name": "Bread", "quantity": 1, "checked": True},
],
}
]
async def test_mealie_shopping_lists_empty(mealie):
mealie.get_shopping_lists = AsyncMock(
return_value={
"total_lists": 0,
"total_unchecked_items": 0,
"shopping_lists": [],
}
)
msg = await call(agents.mealie_shopping_lists)
assert msg.content == "No shopping lists found."
assert msg.artifact is None
@pytest.mark.parametrize(
"tool,args",
[
(agents.mealie_get_recipe, {"recipe_slug": "x"}),
(agents.mealie_meal_plans, {}),
(agents.mealie_todays_meals, {}),
(agents.mealie_shopping_lists, {}),
],
)
async def test_mealie_not_configured(monkeypatch, tool, args):
monkeypatch.setattr(agents, "mealie_enabled", False)
msg = await call(tool, args)
assert msg.content == (
"Mealie integration is not configured. Please set MEALIE_BASE_URL and "
"MEALIE_API_TOKEN environment variables."
)
assert msg.artifact is None
# --- Obsidian ---------------------------------------------------------------
NOTE = """---
title: Vet visit
date: 2026-10-01
tags: [simba]
---
Simba saw the [[Dr Smith|vet]] today. #simba #health #simba
"""
async def test_obsidian_read_note(obsidian, monkeypatch):
monkeypatch.setenv("OBSIDIAN_VAULT_NAME", "Home Vault")
os.makedirs(os.path.join(obsidian.vault_path, "notes"))
with open(os.path.join(obsidian.vault_path, "notes", "vet visit.md"), "w") as f:
f.write(NOTE)
msg = await call(agents.obsidian_read_note, {"relative_path": "notes/vet visit.md"})
assert msg.content == (
"File: notes/vet visit.md\n\n"
"Frontmatter:\n{'title': 'Vet visit', 'date': datetime.date(2026, 10, 1), "
"'tags': ['simba']}\n\n"
"Content:\nSimba saw the vet today. #simba #health #simba\n\n"
"Tags: simba, health, simba\n"
"Contains 1 wikilinks and 0 embeds"
)
art = msg.artifact
assert_artifact(art, "obsidian.note")
assert art["title"] == "Vet visit"
assert art["data"]["frontmatter"] == {
"title": "Vet visit",
"date": "2026-10-01",
"tags": ["simba"],
}
assert art["data"]["tags"] == ["simba", "health"]
assert art["data"]["markdown"] == "Simba saw the vet today. #simba #health #simba"
assert art["source_ref"] == {
"system": "obsidian",
"path": "notes/vet visit.md",
"url": "obsidian://open?vault=Home%20Vault&file=notes%2Fvet%20visit.md",
}
async def test_obsidian_read_note_caps_markdown_and_omits_url(obsidian):
with open(os.path.join(obsidian.vault_path, "big.md"), "w") as f:
f.write("x" * (NOTE_MAX_CHARS + 500))
msg = await call(agents.obsidian_read_note, {"relative_path": "big.md"})
art = msg.artifact
assert art["title"] == "big"
assert len(art["data"]["markdown"]) <= NOTE_MAX_CHARS
assert art["source_ref"] == {"system": "obsidian", "path": "big.md"}
async def test_obsidian_read_note_missing(obsidian):
msg = await call(agents.obsidian_read_note, {"relative_path": "nope.md"})
assert msg.content == (
"Note not found at 'nope.md'. Please check the path is correct."
)
assert msg.artifact is None
async def test_journal_get_today(obsidian):
path = obsidian.get_daily_note_path()
full = os.path.join(obsidian.vault_path, path)
os.makedirs(os.path.dirname(full), exist_ok=True)
raw = "## Tasks\n- [ ] Feed Simba\n"
with open(full, "w") as f:
f.write(raw)
today = datetime.datetime.now().strftime("%Y-%m-%d")
msg = await call(agents.journal_get_today)
assert msg.content == f"Daily note for {today}:\n\n{raw}"
art = msg.artifact
assert_artifact(art, "obsidian.note")
assert art["title"].startswith("Daily note · ")
assert art["data"]["path"] == path
assert art["data"]["markdown"] == "## Tasks\n- [ ] Feed Simba"
async def test_journal_get_today_missing(obsidian):
today = datetime.datetime.now().strftime("%Y-%m-%d")
msg = await call(agents.journal_get_today)
assert msg.content == (
f"No daily note found for {today}. Use journal_add_task to create one."
)
assert msg.artifact is None
async def test_obsidian_not_configured(monkeypatch):
monkeypatch.setattr(agents, "obsidian_enabled", False)
for tool, args in [
(agents.obsidian_read_note, {"relative_path": "a.md"}),
(agents.obsidian_search_notes, {"query": "a"}),
]:
msg = await call(tool, args)
assert msg.content == (
"Obsidian integration is not configured. Please set "
"OBSIDIAN_VAULT_PATH environment variable."
)
assert msg.artifact is None
msg = await call(agents.journal_get_today)
assert msg.content == "Obsidian integration is not configured."
assert msg.artifact is None
# --- Search -----------------------------------------------------------------
PAPERLESS_DOCS = [
Document(
page_content="Rabies vaccine\nadministered. " + "y" * 600,
metadata={
"source": "paperless",
"created_date": 1760227200.0,
"filename": "vet_invoice.pdf",
"document_type": "Invoice",
},
)
]
async def test_simba_search(monkeypatch):
query_mock = AsyncMock(return_value=("Source: {...}\nContent: x", PAPERLESS_DOCS))
monkeypatch.setattr(agents, "query_vector_store", query_mock)
msg = await call(agents.simba_search, {"query": "vet records"})
assert msg.content == "Source: {...}\nContent: x"
query_mock.assert_awaited_once_with(query="vet records", source="paperless")
art = msg.artifact
assert_artifact(art, "search.documents")
assert art["title"] == "Search: vet records"
assert art["data"]["source"] == "paperless"
(result,) = art["data"]["results"]
assert result["title"] == "vet_invoice"
assert result["filename"] == "vet_invoice.pdf"
assert result["document_type"] == "Invoice"
assert len(result["date"]) == 10
assert result["snippet"].startswith("Rabies vaccine administered. ")
assert len(result["snippet"]) <= SNIPPET_MAX_CHARS
async def test_simba_search_empty(monkeypatch):
monkeypatch.setattr(agents, "query_vector_store", AsyncMock(return_value=("", [])))
msg = await call(agents.simba_search, {"query": "nothing"})
assert msg.content == ""
assert msg.artifact is None
async def test_obsidian_search_notes(obsidian, monkeypatch):
docs = [
Document(
page_content="Simba's vet is Dr Smith.",
metadata={
"source": "obsidian",
"filepath": os.path.join(obsidian.vault_path, "pets", "simba.md"),
"tags": ["simba"],
"created_at": "2026-01-02T10:00:00",
},
)
]
monkeypatch.setattr(
agents, "query_vector_store", AsyncMock(return_value=("serialized", docs))
)
msg = await call(agents.obsidian_search_notes, {"query": "vet"})
assert msg.content == "serialized"
art = msg.artifact
assert_artifact(art, "search.documents")
assert art["data"]["results"] == [
{
"snippet": "Simba's vet is Dr Smith.",
"title": "simba",
"path": "pets/simba.md",
"date": "2026-01-02",
}
]
async def test_obsidian_search_notes_error(obsidian, monkeypatch):
monkeypatch.setattr(
agents, "query_vector_store", AsyncMock(side_effect=Exception("db down"))
)
msg = await call(agents.obsidian_search_notes, {"query": "vet"})
assert msg.content == "Error searching Obsidian notes: db down"
assert msg.artifact is None
def test_search_documents_artifact_empty():
assert search_documents_artifact("q", "paperless", []) is None
@pytest.mark.parametrize(
"start,end,expected",
[
("2026-10-12", "2026-10-19", "Oct 12–19"),
("2026-10-28", "2026-11-03", "Oct 28–Nov 3"),
("2026-12-28", "2027-01-03", "Dec 28, 2026–Jan 3, 2027"),
("2026-10-12", "2026-10-12", "Oct 12"),
("bad", "2026-10-12", "bad–2026-10-12"),
],
)
def test_format_date_range(start, end, expected):
assert format_date_range(start, end) == expected
tests/unit/test_ynab_tool_artifacts.py
+427
View File
@@ -0,0 +1,427 @@
"""Tests for the YNAB tools' content_and_artifact output.
`ensure_fresh` and the `_query` DB helper are mocked; the fake answers each
query by matching a fragment of its SQL.
"""
import json
import re
from datetime import date
from decimal import Decimal
from types import SimpleNamespace
import pytest
from langchain_core.messages import ToolMessage
from blueprints.ynab import tools
KIND_DATA_KEYS = {
"ynab.summary": {
"month",
"ready_to_assign",
"income",
"assigned",
"spent",
"net_worth",
"overspent",
},
"ynab.transactions": {"start_date", "end_date", "count", "total", "transactions"},
"ynab.spending_report": {"start_date", "end_date", "group_by", "rows", "total"},
"ynab.category_budget": {"month", "groups"},
}
async def call(tool, args=None) -> ToolMessage:
"""Invoke a tool the way the agent does: with a ToolCall dict."""
msg = await tool.ainvoke(
{"type": "tool_call", "id": "call-1", "name": tool.name, "args": args or {}}
)
assert isinstance(msg, ToolMessage)
return msg
def assert_artifact(artifact, kind, optional=()):
assert artifact is not None
assert artifact["kind"] == kind
assert isinstance(artifact["title"], str) and artifact["title"]
assert KIND_DATA_KEYS[kind] <= set(artifact["data"])
assert set(artifact["data"]) <= KIND_DATA_KEYS[kind] | set(optional)
json.dumps(artifact)
def fake_db(monkeypatch, responses: dict, synced=True):
"""Route `_query` calls to `responses` by SQL fragment; honour LIMIT."""
calls = []
async def fake_query(sql, *params):
calls.append((sql, params))
for fragment, rows in responses.items():
if fragment in sql:
limit = re.search(r"LIMIT (\d+)", sql)
return rows[: int(limit.group(1))] if limit else rows
raise AssertionError(f"unexpected query: {sql}")
async def fake_ensure_fresh():
return SimpleNamespace(budget_name="Home") if synced else None
monkeypatch.setattr(tools, "_query", fake_query)
monkeypatch.setattr(tools, "ensure_fresh", fake_ensure_fresh)
monkeypatch.setattr(tools, "_today", lambda: date(2026, 10, 12))
return calls
# --- ynab_budget_summary ----------------------------------------------------
SUMMARY_DB = {
"FROM ynab_months": [
{
"to_be_budgeted": Decimal("120.50"),
"income": Decimal("5000.00"),
"budgeted": Decimal("4800.00"),
"age_of_money": 42,
}
],
"mc.balance < 0": [{"name": "Dining Out", "balance": Decimal("-23.10")}],
"AS spent": [{"spent": Decimal("3100.25")}],
"GROUP BY on_budget": [
{"on_budget": True, "total": Decimal("10000.00")},
{"on_budget": False, "total": Decimal("-2500.00")},
],
}
async def test_budget_summary(monkeypatch):
fake_db(monkeypatch, SUMMARY_DB)
msg = await call(tools.ynab_budget_summary, {"month": "2026-10"})
assert msg.content == (
"YNAB budget 'Home' for 2026-10:\n"
"Ready to Assign: $120.50\n"
"Income: $5,000.00\n"
"Assigned: $4,800.00\n"
"Spent: $3,100.25\n"
"Age of money: 42 days\n"
"Overspent: Dining Out (-$23.10)\n"
"Net worth: $7,500.00 (budget accounts $10,000.00, "
"tracking accounts -$2,500.00)"
)
art = msg.artifact
assert_artifact(art, "ynab.summary", optional={"age_of_money"})
assert art["title"] == "Budget · October 2026"
assert art["data"] == {
"month": "2026-10",
"ready_to_assign": 120.5,
"income": 5000.0,
"assigned": 4800.0,
"spent": 3100.25,
"age_of_money": 42,
"net_worth": 7500.0,
"overspent": [{"name": "Dining Out", "available": -23.1}],
}
async def test_budget_summary_without_age_of_money(monkeypatch):
db = dict(SUMMARY_DB)
db["FROM ynab_months"] = [
{**SUMMARY_DB["FROM ynab_months"][0], "age_of_money": None}
]
fake_db(monkeypatch, db)
msg = await call(tools.ynab_budget_summary, {"month": "2026-10"})
assert "Age of money" not in msg.content
assert "age_of_money" not in msg.artifact["data"]
async def test_budget_summary_no_month_data(monkeypatch):
fake_db(monkeypatch, {"FROM ynab_months": []})
msg = await call(tools.ynab_budget_summary, {"month": "2026-10"})
assert msg.content == "No YNAB budget data for 2026-10."
assert msg.artifact is None
# --- ynab_search_transactions -----------------------------------------------
def _txn(i, amount, **extra):
return {
"id": f"t{i}",
"date": date(2026, 10, 12 - i % 10),
"amount": Decimal(amount),
"memo": None,
"approved": True,
"account_name": "Checking",
"payee_name": "Costco",
"category_name": "Groceries",
**extra,
}
TXNS = [
_txn(0, "-54.20", memo="milk"),
_txn(1, "-12.00", approved=False, payee_name=None, category_name=None),
_txn(2, "100.00"),
]
TXN_TOTALS = {
"count(*) AS n": [
{"n": 3, "outflow": Decimal("-66.20"), "inflow": Decimal("100.00")}
],
}
async def test_search_transactions(monkeypatch):
calls = fake_db(monkeypatch, {**TXN_TOTALS, "SELECT * FROM lines": TXNS})
msg = await call(tools.ynab_search_transactions, {"limit": 2})
assert msg.content == (
"Found 3 transactions from 2026-09-12 to 2026-10-12. "
"Outflows -$66.20, inflows $100.00.\n"
"- 2026-10-12: Costco -$54.20 (Groceries; Checking) — milk\n"
"- 2026-10-11: Unknown payee -$12.00 (Uncategorized; Checking) [unapproved]\n"
"(Showing 2 most recent of 3)"
)
assert "LIMIT 500" in calls[-1][0]
art = msg.artifact
assert_artifact(art, "ynab.transactions")
assert art["title"] == "Transactions · Sep 12–Oct 12"
data = art["data"]
assert data["start_date"] == "2026-09-12"
assert data["end_date"] == "2026-10-12"
assert data["count"] == 3
assert data["total"] == pytest.approx(33.8)
# the artifact keeps every row, not just the `limit` shown to the model
assert len(data["transactions"]) == 3
assert data["transactions"][0] == {
"date": "2026-10-12",
"payee": "Costco",
"category": "Groceries",
"account": "Checking",
"memo": "milk",
"amount": -54.2,
}
assert "memo" not in data["transactions"][1]
async def test_search_transactions_caps_artifact_rows(monkeypatch):
many = [_txn(i, "-1.00") for i in range(600)]
totals = {"count(*) AS n": [{"n": 600, "outflow": Decimal("-600"), "inflow": 0}]}
fake_db(monkeypatch, {**totals, "SELECT * FROM lines": many})
msg = await call(tools.ynab_search_transactions, {"payee": "costco"})
assert msg.content.endswith("(Showing 20 most recent of 600)")
assert len(msg.artifact["data"]["transactions"]) == tools.ARTIFACT_MAX_TRANSACTIONS
assert msg.artifact["data"]["start_date"] is None
assert msg.artifact["title"] == "Transactions · through 2026-10-12"
async def test_search_transactions_empty(monkeypatch):
fake_db(monkeypatch, {"count(*) AS n": [{"n": 0, "outflow": 0, "inflow": 0}]})
msg = await call(tools.ynab_search_transactions, {})
assert msg.content == "No transactions found from 2026-09-12 to 2026-10-12."
assert msg.artifact is None
async def test_search_transactions_bad_date(monkeypatch):
fake_db(monkeypatch, {})
msg = await call(tools.ynab_search_transactions, {"start_date": "nope"})
assert msg.content == "Dates must be YYYY-MM-DD."
assert msg.artifact is None
# --- ynab_spending_report ---------------------------------------------------
SPENDING_ROWS = [
{"key": "Groceries", "spent": Decimal("400.00"), "n": 10},
{"key": "Dining Out", "spent": Decimal("150.50"), "n": 5},
{"key": "Gas", "spent": Decimal("60.00"), "n": 2},
]
async def test_spending_report(monkeypatch):
fake_db(monkeypatch, {"GROUP BY 1": SPENDING_ROWS})
msg = await call(tools.ynab_spending_report, {"top": 2})
assert msg.content == (
"Spending by category from 2026-10-01 to 2026-10-12: total $610.50\n"
"- Groceries: $400.00 (10 txns)\n"
"- Dining Out: $150.50 (5 txns)\n"
"- 1 others: $60.00"
)
art = msg.artifact
assert_artifact(art, "ynab.spending_report")
assert art["title"] == "Spending by category · Oct 1–12"
assert art["data"] == {
"start_date": "2026-10-01",
"end_date": "2026-10-12",
"group_by": "category",
"rows": [
{"key": "Groceries", "amount": 400.0, "count": 10},
{"key": "Dining Out", "amount": 150.5, "count": 5},
{"key": "Gas", "amount": 60.0, "count": 2},
],
"total": 610.5,
}
async def test_spending_report_by_month(monkeypatch):
rows = [
{"key": "2026-09", "spent": Decimal("100"), "n": 1},
{"key": "2026-10", "spent": Decimal("200"), "n": 2},
]
fake_db(monkeypatch, {"GROUP BY 1": rows})
msg = await call(
tools.ynab_spending_report,
{"group_by": "month", "start_date": "2026-09-01"},
)
assert msg.content.endswith("Monthly average: $150.00")
assert [r["key"] for r in msg.artifact["data"]["rows"]] == ["2026-09", "2026-10"]
async def test_spending_report_empty_and_invalid(monkeypatch):
fake_db(monkeypatch, {"GROUP BY 1": []})
msg = await call(tools.ynab_spending_report, {})
assert msg.content == "No spending found from 2026-10-01 to 2026-10-12."
assert msg.artifact is None
msg = await call(tools.ynab_spending_report, {"group_by": "vibes"})
assert msg.content == (
"group_by must be one of: category, category_group, payee, account, month."
)
assert msg.artifact is None
# --- ynab_category_budget ---------------------------------------------------
def _cat(group, name, budgeted, activity, balance, goal=None):
return {
"group_name": group,
"name": name,
"goal_target": Decimal(goal) if goal else None,
"budgeted": Decimal(budgeted),
"activity": Decimal(activity),
"balance": Decimal(balance),
}
async def test_category_budget(monkeypatch):
rows = [
_cat("Bills", "Rent", "2000", "-2000", "0"),
_cat("Bills", "Unused", "0", "0", "0"),
_cat("Food", "Dining Out", "100", "-123.10", "-23.10", goal="150"),
]
fake_db(monkeypatch, {"FROM ynab_month_categories mc": rows})
msg = await call(tools.ynab_category_budget, {"month": "2026-10"})
assert msg.content == (
"Categories for 2026-10:\n"
"\nBills\n"
" Rent: assigned $2,000.00, activity -$2,000.00, available $0.00\n"
"\nFood\n"
" Dining Out: assigned $100.00, activity -$123.10, available -$23.10, "
"goal $150.00 [OVERSPENT]"
)
art = msg.artifact
assert_artifact(art, "ynab.category_budget")
assert art["title"] == "Category budget · October 2026"
assert art["data"] == {
"month": "2026-10",
"groups": [
{
"name": "Bills",
"categories": [
{
"name": "Rent",
"assigned": 2000.0,
"activity": -2000.0,
"available": 0.0,
}
],
},
{
"name": "Food",
"categories": [
{
"name": "Dining Out",
"assigned": 100.0,
"activity": -123.1,
"available": -23.1,
"goal_target": 150.0,
}
],
},
],
}
async def test_category_budget_all_zero_rows(monkeypatch):
rows = [_cat("Bills", "Unused", "0", "0", "0")]
fake_db(monkeypatch, {"FROM ynab_month_categories mc": rows})
msg = await call(tools.ynab_category_budget, {"month": "2026-10"})
assert msg.content == "Categories for 2026-10:"
assert msg.artifact is None
async def test_category_budget_empty(monkeypatch):
fake_db(monkeypatch, {"FROM ynab_month_categories mc": []})
msg = await call(tools.ynab_category_budget, {"month": "2026-10"})
assert msg.content == "No YNAB category data for 2026-10."
assert msg.artifact is None
# --- shared paths -----------------------------------------------------------
ARTIFACT_TOOLS = [
tools.ynab_budget_summary,
tools.ynab_search_transactions,
tools.ynab_spending_report,
tools.ynab_category_budget,
]
@pytest.mark.parametrize("tool", ARTIFACT_TOOLS, ids=lambda t: t.name)
async def test_not_synced(monkeypatch, tool):
fake_db(monkeypatch, {}, synced=False)
msg = await call(tool)
assert msg.content == tools.NOT_SYNCED
assert msg.artifact is None
@pytest.mark.parametrize("tool", ARTIFACT_TOOLS, ids=lambda t: t.name)
async def test_db_error(monkeypatch, tool):
fake_db(monkeypatch, {})
async def boom(sql, *params):
raise RuntimeError("db down")
monkeypatch.setattr(tools, "_query", boom)
msg = await call(tool)
assert msg.content == f"YNAB error in {tool.name}: RuntimeError: db down"
assert msg.artifact is None
def test_text_only_tools_unchanged():
assert tools.ynab_account_balances.response_format == "content"
assert tools.ynab_recent_changes.response_format == "content"