ryan
5809ad3adb
Shared runtime - One agent entry point (runtime.invoke_agent_for_user / prepare_agent_turn) for web, email, iMessage, WhatsApp, scheduled and cron, so identity, permissions, memories, summaries and metrics match across channels. - Serialize turns per checkpoint thread; /query validates input and returns 409 while a durable run owns the conversation. Reconnectable chat (durable runs) - POST /runs creates an idempotent, owner-scoped run; GET .../events replays an ordered SSE log by cursor, so generation survives refresh and device switches. Stop cancels; disconnect only detaches. /stream-query is a compatibility adapter. - Coalesce streamed deltas (150 ms) and wake same-process subscribers instead of polling; keep buffered text when a run is stopped. - Drain runs on shutdown; serve via Hypercorn with a shutdown timeout covering the drain (and no longer in debug mode). CHAT_RUN_* settings are forwarded. Scheduled messages and cron - Atomic delivery claims, attempt history and explicit retries; ambiguous sends become "uncertain" and are never resent automatically. - Cron triggers need a run_id/Idempotency-Key; legacy jobs need an explicit owner. - Cancelling never rewrites a started delivery; series stop via tombstones. Legacy sent rows are not forked into new series. Malformed bodies return 400. History, integrations and retrieval - Message-content search, pinned conversations, JSON/Markdown export and a memory editor. - Integration status panel and YNAB data-freshness provenance. - Date-filtered document/note search with stable citations and a source viewer. Migration 20 is generated by aerich. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
60 lines
2.3 KiB
Python
60 lines
2.3 KiB
Python
"""One durable delivery per caller-supplied cron run identifier."""
|
|
|
|
from tortoise.transactions import in_transaction
|
|
|
|
from blueprints.scheduled_messages import delivery
|
|
from .models import CronJob
|
|
|
|
|
|
async def execute_job(job, run_id, *, retry=False):
|
|
await delivery.expire_claims("cron")
|
|
if not job.enabled:
|
|
return {"skipped": True, "status": "disabled"}
|
|
claimed = await delivery.claim("cron", job.id, run_id, retry=retry)
|
|
if not claimed:
|
|
from blueprints.scheduled_messages.models import JobDelivery
|
|
|
|
existing = await JobDelivery.get(kind="cron", job_id=str(job.id), run_id=run_id)
|
|
return {"skipped": True, "status": existing.status}
|
|
sending = False
|
|
try:
|
|
if not job.created_by_id:
|
|
raise ValueError(
|
|
"Cron job requires an explicit created_by owner before execution"
|
|
)
|
|
user = await job.created_by
|
|
if user is None:
|
|
raise ValueError("Cron job owner no longer exists")
|
|
content = await delivery.agent_content(
|
|
user, job.prompt, f"cron-{claimed.id}", "cron"
|
|
)
|
|
async with in_transaction() as db:
|
|
active = (
|
|
await CronJob.filter(id=job.id, enabled=True)
|
|
.using_db(db)
|
|
.update(updated_at=delivery.now())
|
|
)
|
|
if not active:
|
|
await delivery.transition(claimed, "cancelled")
|
|
return {"skipped": True, "status": "cancelled"}
|
|
if not await delivery.transition(claimed, "sending"):
|
|
return {"skipped": True, "status": "uncertain"}
|
|
sending = True
|
|
await delivery.send(job.channel, job.recipient, content, job.name)
|
|
await delivery.transition(claimed, "sent")
|
|
status, error = "sent", None
|
|
except Exception as exc:
|
|
status = (
|
|
"uncertain"
|
|
if sending and not isinstance(exc, delivery.SafeDeliveryError)
|
|
else "failed"
|
|
)
|
|
error = str(exc)
|
|
await delivery.transition(claimed, status, error)
|
|
await CronJob.filter(id=job.id).update(
|
|
last_run_at=delivery.now(),
|
|
last_status="success" if status == "sent" else status,
|
|
last_error=error,
|
|
)
|
|
return {"ok": status == "sent", "status": status, "error": error}
|