"""The shared 'thread in -> ActionPlan out' pipeline. One implementation behind both the synchronous ``POST /agent`` endpoint and the autonomous ingest path, so the two can never drift. Stateless: it does not touch the feed or the dedup store (callers own statefulness). Importable without Gradio. Google Calendar is imported lazily so CI / stub mode (which exclude the google libs) stay clean. """ from __future__ import annotations import base64 from typing import Optional from dateutil import parser as dtparser from pydantic import BaseModel from calendar_out.freebusy import DEFAULT_DURATION, Busy, _as_dt, annotate_conflicts, load_ics_busy from calendar_out.ics import events_to_ics from server import events as bus from server.agent import run_agent from server.schema import ActionPlan, Event from server.threads import format_thread class AgentMessage(BaseModel): sender: str = "?" text: str = "" class AgentRequest(BaseModel): thread: Optional[str] = None messages: Optional[list[AgentMessage]] = None images: list[str] = [] # base64 data URIs existing_ics: Optional[str] = None # base64-encoded .ics bytes existing_events: list[Event] = [] now: Optional[str] = None # ISO 8601; defaults to datetime.now() push_gcal: bool = False return_ics: bool = False memory: Optional[str] = None # per-user recall block (else server global memory) class AgentResponse(BaseModel): plan: ActionPlan ics_base64: Optional[str] = None gcal_links: list[str] = [] def _busy_from_request(req: AgentRequest) -> list[Busy]: """Build busy intervals from an uploaded .ics or structured existing events.""" if req.existing_ics: try: return load_ics_busy(base64.b64decode(req.existing_ics)) except Exception: # noqa: BLE001 malformed .ics -> no conflict context return [] busy: list[Busy] = [] for ev in req.existing_events: start = _as_dt(ev.start) if start is None: continue end = _as_dt(ev.end) or (start + DEFAULT_DURATION) busy.append(Busy(start=start, end=end, title=ev.title)) return busy def _thread_text(req: AgentRequest) -> str: if req.thread: return req.thread if req.messages: return format_thread([m.model_dump() for m in req.messages]) return "" def run_pipeline(req: AgentRequest) -> AgentResponse: """thread/messages -> run_agent -> deterministic conflicts -> optional ics/gcal.""" thread = _thread_text(req) now = dtparser.isoparse(req.now) if req.now else None busy = _busy_from_request(req) plan = run_agent(thread, now=now, existing=req.existing_events, images=req.images, memory_block=req.memory) if busy: plan = annotate_conflicts(plan, busy) resp = AgentResponse(plan=plan) if req.return_ics: resp.ics_base64 = base64.b64encode(events_to_ics(plan.events)).decode("ascii") if req.push_gcal and plan.events: try: from calendar_out.gcal import push_events # lazy: google libs optional resp.gcal_links = push_events(plan.events) except Exception as e: # noqa: BLE001 no token.json / offline -> degrade bus.emit("calendar", f"Google Calendar push skipped: {e}", level="error") return resp