Spaces:
Sleeping
Sleeping
| """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 | |