server/pipeline.py

"""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