feat: push interview events to connected calendar integrations (#19)

Implements idempotent calendar push for Apple Calendar (CalDAV) and
Google Calendar from the Interviews kanban.

- db: add calendar_event_id column (migration) + set_calendar_event_id helper
- integrations/apple_calendar: create_event / update_event via caldav + icalendar
- integrations/google_calendar: create_event / update_event via google-api-python-client;
  test() now makes a real API call instead of checking file existence
- scripts/calendar_push: orchestrates push/update, builds event title from stage +
  job title + company, attaches job URL and company brief to description,
  defaults to noon UTC / 1hr duration
- app/pages/5_Interviews: "Add to Calendar" / "Update Calendar" button shown
  when interview date is set and a calendar integration is configured
- environment.yml: pin caldav, icalendar, google-api-python-client, google-auth
- tests/test_calendar_push: 9 tests covering create, update, error handling,
  event timing, idempotency, and missing job/date guards

app/pages/5_Interviews.py

@@ -31,12 +31,19 @@ _name = _profile.name if _profile else "Job Seeker"
 from scripts.db import (
     DEFAULT_DB, init_db,
     get_interview_jobs, advance_to_stage, reject_at_stage,
-    set_interview_date, add_contact, get_contacts,
+    set_interview_date, set_calendar_event_id, add_contact, get_contacts,
     get_research, get_task_for_job, get_job_by_id,
     get_unread_stage_signals, dismiss_stage_signal,
 )
 from scripts.task_runner import submit_task
 
+_CONFIG_DIR = Path(__file__).parent.parent.parent / "config"
+_CALENDAR_INTEGRATIONS = ("apple_calendar", "google_calendar")
+_calendar_connected = any(
+    (_CONFIG_DIR / "integrations" / f"{n}.yaml").exists()
+    for n in _CALENDAR_INTEGRATIONS
+)
+
 st.title("🎯 Interviews")
 
 init_db(DEFAULT_DB)
@@ -275,6 +282,19 @@ def _render_card(job: dict, stage: str, compact: bool = False) -> None:
                     st.success("Saved!")
                     st.rerun()
 
+            # Calendar push — only shown when a date is saved and an integration is connected
+            if current_idate and _calendar_connected:
+                _has_event = bool(job.get("calendar_event_id"))
+                _cal_label = "🔄 Update Calendar" if _has_event else "📅 Add to Calendar"
+                if st.button(_cal_label, key=f"cal_push_{job_id}", use_container_width=True):
+                    from scripts.calendar_push import push_interview_event
+                    result = push_interview_event(DEFAULT_DB, job_id=job_id, config_dir=_CONFIG_DIR)
+                    if result["ok"]:
+                        st.success(f"Event {'updated' if _has_event else 'added'} ({result['provider'].replace('_', ' ').title()})")
+                        st.rerun()
+                    else:
+                        st.error(result["error"])
+
         if not compact:
             if stage in ("applied", "phone_screen", "interviewing"):
                 signals = get_unread_stage_signals(DEFAULT_DB, job_id=job_id)

environment.yml

@@ -48,6 +48,12 @@ dependencies:
     # ── Notion integration ────────────────────────────────────────────────────
     - notion-client>=3.0
 
+    # ── Calendar integrations ─────────────────────────────────────────────────
+    - caldav>=1.3
+    - icalendar>=5.0
+    - google-api-python-client>=2.0
+    - google-auth>=2.0
+
     # ── Document handling ─────────────────────────────────────────────────────
     - pypdf
     - pdfminer-six

scripts/calendar_push.py

@@ -0,0 +1,119 @@
+"""calendar_push.py — push interview events to connected calendar integrations.
+
+Supports Apple Calendar (CalDAV) and Google Calendar. Idempotent: a second
+push updates the existing event rather than creating a duplicate.
+"""
+from __future__ import annotations
+
+import uuid
+import yaml
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from typing import Optional
+
+from scripts.db import get_job_by_id, get_research, set_calendar_event_id, DEFAULT_DB
+
+_CALENDAR_INTEGRATIONS = ("apple_calendar", "google_calendar")
+
+# Stage label map matches 5_Interviews.py
+_STAGE_LABELS = {
+    "phone_screen":  "Phone Screen",
+    "interviewing":  "Interview",
+    "offer":         "Offer Review",
+}
+
+
+def _load_integration(name: str, config_dir: Path):
+    """Instantiate and connect an integration from its saved config file."""
+    config_file = config_dir / "integrations" / f"{name}.yaml"
+    if not config_file.exists():
+        return None
+    with open(config_file) as f:
+        config = yaml.safe_load(f) or {}
+    if name == "apple_calendar":
+        from scripts.integrations.apple_calendar import AppleCalendarIntegration
+        integration = AppleCalendarIntegration()
+    elif name == "google_calendar":
+        from scripts.integrations.google_calendar import GoogleCalendarIntegration
+        integration = GoogleCalendarIntegration()
+    else:
+        return None
+    integration.connect(config)
+    return integration
+
+
+def _build_event_details(job: dict, research: Optional[dict]) -> tuple[str, str]:
+    """Return (title, description) for the calendar event."""
+    stage_label = _STAGE_LABELS.get(job.get("status", ""), "Interview")
+    title = f"{stage_label}: {job.get('title', 'Interview')} @ {job.get('company', '')}"
+
+    lines = []
+    if job.get("url"):
+        lines.append(f"Job listing: {job['url']}")
+    if research and research.get("company_brief"):
+        brief = research["company_brief"].strip()
+        # Trim to first 3 sentences so the event description stays readable
+        sentences = brief.split(". ")
+        lines.append("\n" + ". ".join(sentences[:3]) + ("." if len(sentences) > 1 else ""))
+    lines.append("\n— Sent by Peregrine (CircuitForge)")
+
+    return title, "\n".join(lines)
+
+
+def push_interview_event(
+    db_path: Path = DEFAULT_DB,
+    job_id: int = None,
+    config_dir: Path = None,
+) -> dict:
+    """Push (or update) an interview event on the first connected calendar integration.
+
+    Returns:
+        {"ok": True,  "provider": "apple_calendar", "event_id": "..."}
+        {"ok": False, "error": "..."}
+    """
+    if config_dir is None:
+        config_dir = Path(__file__).parent.parent / "config"
+
+    job = get_job_by_id(db_path, job_id)
+    if not job:
+        return {"ok": False, "error": f"Job {job_id} not found"}
+
+    interview_date = job.get("interview_date")
+    if not interview_date:
+        return {"ok": False, "error": "No interview date set — save a date first"}
+
+    # Build datetimes: noon UTC, 1 hour duration
+    try:
+        base = datetime.fromisoformat(interview_date).replace(
+            hour=12, minute=0, second=0, microsecond=0, tzinfo=timezone.utc
+        )
+    except ValueError:
+        return {"ok": False, "error": f"Could not parse interview_date: {interview_date!r}"}
+    start_dt = base
+    end_dt = base + timedelta(hours=1)
+
+    research = get_research(db_path, job_id)
+    title, description = _build_event_details(job, research)
+
+    existing_event_id = job.get("calendar_event_id")
+
+    for name in _CALENDAR_INTEGRATIONS:
+        integration = _load_integration(name, config_dir)
+        if integration is None:
+            continue
+
+        try:
+            # Use a stable UID derived from job_id for CalDAV; gcal uses the returned event id
+            uid = existing_event_id or f"peregrine-job-{job_id}@circuitforge.tech"
+            if existing_event_id:
+                event_id = integration.update_event(uid, title, start_dt, end_dt, description)
+            else:
+                event_id = integration.create_event(uid, title, start_dt, end_dt, description)
+
+            set_calendar_event_id(db_path, job_id, event_id)
+            return {"ok": True, "provider": name, "event_id": event_id}
+
+        except Exception as exc:
+            return {"ok": False, "error": str(exc)}
+
+    return {"ok": False, "error": "No calendar integration configured — connect one in Settings → Integrations"}

scripts/db.py

@@ -138,15 +138,16 @@ CREATE TABLE IF NOT EXISTS survey_responses (
 """
 
 _MIGRATIONS = [
-    ("cover_letter",    "TEXT"),
+    ("cover_letter",       "TEXT"),
-    ("applied_at",      "TEXT"),
+    ("applied_at",         "TEXT"),
-    ("interview_date",  "TEXT"),
+    ("interview_date",     "TEXT"),
-    ("rejection_stage", "TEXT"),
+    ("rejection_stage",    "TEXT"),
-    ("phone_screen_at", "TEXT"),
+    ("phone_screen_at",    "TEXT"),
-    ("interviewing_at", "TEXT"),
+    ("interviewing_at",    "TEXT"),
-    ("offer_at",        "TEXT"),
+    ("offer_at",           "TEXT"),
-    ("hired_at",        "TEXT"),
+    ("hired_at",           "TEXT"),
-    ("survey_at",       "TEXT"),
+    ("survey_at",          "TEXT"),
+    ("calendar_event_id",  "TEXT"),
 ]
 
 
@@ -508,6 +509,15 @@ def set_interview_date(db_path: Path = DEFAULT_DB, job_id: int = None,
     conn.close()
 
 
+def set_calendar_event_id(db_path: Path = DEFAULT_DB, job_id: int = None,
+                          event_id: str = "") -> None:
+    """Persist the calendar event ID returned after a successful push."""
+    conn = sqlite3.connect(db_path)
+    conn.execute("UPDATE jobs SET calendar_event_id = ? WHERE id = ?", (event_id, job_id))
+    conn.commit()
+    conn.close()
+
+
 # ── Contact log helpers ───────────────────────────────────────────────────────
 
 def add_contact(db_path: Path = DEFAULT_DB, job_id: int = None,

scripts/integrations/apple_calendar.py

@@ -1,4 +1,5 @@
 from __future__ import annotations
+from datetime import datetime, timedelta, timezone
 from scripts.integrations.base import IntegrationBase
 
 
@@ -46,3 +47,60 @@ class AppleCalendarIntegration(IntegrationBase):
             return principal is not None
         except Exception:
             return False
+
+    def _get_calendar(self):
+        """Return the configured caldav Calendar object."""
+        import caldav
+        client = caldav.DAVClient(
+            url=self._config["caldav_url"],
+            username=self._config["username"],
+            password=self._config["app_password"],
+        )
+        principal = client.principal()
+        cal_name = self._config.get("calendar_name", "Interviews")
+        for cal in principal.calendars():
+            if cal.name == cal_name:
+                return cal
+        # Calendar not found — create it
+        return principal.make_calendar(name=cal_name)
+
+    def create_event(self, uid: str, title: str, start_dt: datetime,
+                     end_dt: datetime, description: str = "") -> str:
+        """Create a calendar event. Returns the UID (used as calendar_event_id)."""
+        from icalendar import Calendar, Event
+        cal = Calendar()
+        cal.add("prodid", "-//CircuitForge Peregrine//EN")
+        cal.add("version", "2.0")
+        event = Event()
+        event.add("uid", uid)
+        event.add("summary", title)
+        event.add("dtstart", start_dt)
+        event.add("dtend", end_dt)
+        event.add("description", description)
+        cal.add_component(event)
+        dav_cal = self._get_calendar()
+        dav_cal.add_event(cal.to_ical().decode())
+        return uid
+
+    def update_event(self, uid: str, title: str, start_dt: datetime,
+                     end_dt: datetime, description: str = "") -> str:
+        """Update an existing event by UID, or create it if not found."""
+        from icalendar import Calendar, Event
+        dav_cal = self._get_calendar()
+        try:
+            existing = dav_cal.event_by_uid(uid)
+            cal = Calendar()
+            cal.add("prodid", "-//CircuitForge Peregrine//EN")
+            cal.add("version", "2.0")
+            event = Event()
+            event.add("uid", uid)
+            event.add("summary", title)
+            event.add("dtstart", start_dt)
+            event.add("dtend", end_dt)
+            event.add("description", description)
+            cal.add_component(event)
+            existing.data = cal.to_ical().decode()
+            existing.save()
+        except Exception:
+            return self.create_event(uid, title, start_dt, end_dt, description)
+        return uid

scripts/integrations/google_calendar.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 import os
+from datetime import datetime
 from scripts.integrations.base import IntegrationBase
 
 
@@ -26,6 +27,53 @@ class GoogleCalendarIntegration(IntegrationBase):
         return bool(config.get("calendar_id") and config.get("credentials_json"))
 
     def test(self) -> bool:
-        # TODO: use google-api-python-client calendars().get()
+        try:
-        creds = os.path.expanduser(self._config.get("credentials_json", ""))
+            service = self._build_service()
-        return os.path.exists(creds)
+            service.calendars().get(calendarId=self._config["calendar_id"]).execute()
+            return True
+        except Exception:
+            return False
+
+    def _build_service(self):
+        from google.oauth2 import service_account
+        from googleapiclient.discovery import build
+        creds_path = os.path.expanduser(self._config["credentials_json"])
+        creds = service_account.Credentials.from_service_account_file(
+            creds_path,
+            scopes=["https://www.googleapis.com/auth/calendar"],
+        )
+        return build("calendar", "v3", credentials=creds)
+
+    def _fmt(self, dt: datetime) -> str:
+        return dt.strftime("%Y-%m-%dT%H:%M:%S") + "Z"
+
+    def create_event(self, uid: str, title: str, start_dt: datetime,
+                     end_dt: datetime, description: str = "") -> str:
+        """Create a Google Calendar event. Returns the Google event ID."""
+        service = self._build_service()
+        body = {
+            "summary": title,
+            "description": description,
+            "start": {"dateTime": self._fmt(start_dt), "timeZone": "UTC"},
+            "end":   {"dateTime": self._fmt(end_dt),   "timeZone": "UTC"},
+            "extendedProperties": {"private": {"peregrine_uid": uid}},
+        }
+        result = service.events().insert(
+            calendarId=self._config["calendar_id"], body=body
+        ).execute()
+        return result["id"]
+
+    def update_event(self, uid: str, title: str, start_dt: datetime,
+                     end_dt: datetime, description: str = "") -> str:
+        """Update an existing Google Calendar event by its stored event ID (uid is the gcal id)."""
+        service = self._build_service()
+        body = {
+            "summary": title,
+            "description": description,
+            "start": {"dateTime": self._fmt(start_dt), "timeZone": "UTC"},
+            "end":   {"dateTime": self._fmt(end_dt),   "timeZone": "UTC"},
+        }
+        result = service.events().update(
+            calendarId=self._config["calendar_id"], eventId=uid, body=body
+        ).execute()
+        return result["id"]

tests/test_calendar_push.py

@@ -0,0 +1,193 @@
+# tests/test_calendar_push.py
+"""Unit tests for scripts/calendar_push.py.
+
+Integration classes are mocked — no real CalDAV or Google API calls.
+"""
+import sys
+from datetime import timezone
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+
+# ── Fixtures ──────────────────────────────────────────────────────────────────
+
+def _make_db(tmp_path, interview_date="2026-04-15", calendar_event_id=None):
+    from scripts.db import init_db, insert_job, set_interview_date, set_calendar_event_id
+    db = tmp_path / "test.db"
+    init_db(db)
+    job_id = insert_job(db, {
+        "title": "Customer Success Manager", "company": "Acme Corp",
+        "url": "https://example.com/job/1", "source": "linkedin",
+        "location": "Remote", "is_remote": True,
+        "salary": "", "description": "Great role.", "date_found": "2026-04-01",
+        "status": "phone_screen",
+    })
+    if interview_date:
+        set_interview_date(db, job_id=job_id, date_str=interview_date)
+    if calendar_event_id:
+        set_calendar_event_id(db, job_id=job_id, event_id=calendar_event_id)
+    return db, job_id
+
+
+def _config_dir_with(tmp_path, integration_name: str) -> Path:
+    """Create a minimal integration config file and return the config dir."""
+    integrations_dir = tmp_path / "config" / "integrations"
+    integrations_dir.mkdir(parents=True)
+    (integrations_dir / f"{integration_name}.yaml").write_text(
+        "caldav_url: https://caldav.example.com/\n"
+        "username: user@example.com\n"
+        "app_password: test-password\n"
+        "calendar_name: Interviews\n"
+    )
+    return tmp_path / "config"
+
+
+# ── No integration configured ─────────────────────────────────────────────────
+
+def test_push_returns_error_when_no_integration_configured(tmp_path):
+    db, job_id = _make_db(tmp_path)
+    config_dir = tmp_path / "config"
+    config_dir.mkdir()
+
+    from scripts.calendar_push import push_interview_event
+    result = push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    assert result["ok"] is False
+    assert "No calendar integration" in result["error"]
+
+
+# ── No interview date ─────────────────────────────────────────────────────────
+
+def test_push_returns_error_when_no_interview_date(tmp_path):
+    db, job_id = _make_db(tmp_path, interview_date=None)
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    from scripts.calendar_push import push_interview_event
+    result = push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    assert result["ok"] is False
+    assert "No interview date" in result["error"]
+
+
+# ── Successful create ─────────────────────────────────────────────────────────
+
+def test_push_creates_event_and_stores_event_id(tmp_path):
+    db, job_id = _make_db(tmp_path)
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    mock_integration = MagicMock()
+    mock_integration.create_event.return_value = "peregrine-job-1@circuitforge.tech"
+
+    with patch("scripts.calendar_push._load_integration", return_value=mock_integration):
+        from scripts.calendar_push import push_interview_event
+        result = push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    assert result["ok"] is True
+    assert result["event_id"] == "peregrine-job-1@circuitforge.tech"
+    mock_integration.create_event.assert_called_once()
+
+
+def test_push_event_title_includes_stage_and_company(tmp_path):
+    db, job_id = _make_db(tmp_path)
+    from scripts.db import advance_to_stage
+    advance_to_stage(db, job_id=job_id, stage="phone_screen")
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    mock_integration = MagicMock()
+    mock_integration.create_event.return_value = "uid-123"
+
+    with patch("scripts.calendar_push._load_integration", return_value=mock_integration):
+        from scripts.calendar_push import push_interview_event
+        push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    call_kwargs = mock_integration.create_event.call_args
+    title = call_kwargs.args[1] if call_kwargs.args else call_kwargs.kwargs.get("title", "")
+    assert "Acme Corp" in title
+    assert "Phone Screen" in title
+
+
+def test_push_event_start_is_noon_utc(tmp_path):
+    db, job_id = _make_db(tmp_path, interview_date="2026-04-15")
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    mock_integration = MagicMock()
+    mock_integration.create_event.return_value = "uid-abc"
+
+    with patch("scripts.calendar_push._load_integration", return_value=mock_integration):
+        from scripts.calendar_push import push_interview_event
+        push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    call_args = mock_integration.create_event.call_args.args
+    start_dt = call_args[2]
+    assert start_dt.hour == 12
+    assert start_dt.tzinfo == timezone.utc
+
+
+def test_push_event_duration_is_one_hour(tmp_path):
+    db, job_id = _make_db(tmp_path, interview_date="2026-04-15")
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    mock_integration = MagicMock()
+    mock_integration.create_event.return_value = "uid-abc"
+
+    with patch("scripts.calendar_push._load_integration", return_value=mock_integration):
+        from scripts.calendar_push import push_interview_event
+        push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    call_args = mock_integration.create_event.call_args.args
+    start_dt, end_dt = call_args[2], call_args[3]
+    assert (end_dt - start_dt).seconds == 3600
+
+
+# ── Idempotent update ─────────────────────────────────────────────────────────
+
+def test_push_calls_update_when_event_id_already_exists(tmp_path):
+    db, job_id = _make_db(tmp_path, calendar_event_id="existing-event-id")
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    mock_integration = MagicMock()
+    mock_integration.update_event.return_value = "existing-event-id"
+
+    with patch("scripts.calendar_push._load_integration", return_value=mock_integration):
+        from scripts.calendar_push import push_interview_event
+        result = push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    assert result["ok"] is True
+    mock_integration.update_event.assert_called_once()
+    mock_integration.create_event.assert_not_called()
+
+
+# ── Integration error handling ────────────────────────────────────────────────
+
+def test_push_returns_error_on_integration_exception(tmp_path):
+    db, job_id = _make_db(tmp_path)
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    mock_integration = MagicMock()
+    mock_integration.create_event.side_effect = RuntimeError("CalDAV server unreachable")
+
+    with patch("scripts.calendar_push._load_integration", return_value=mock_integration):
+        from scripts.calendar_push import push_interview_event
+        result = push_interview_event(db, job_id=job_id, config_dir=config_dir)
+
+    assert result["ok"] is False
+    assert "CalDAV server unreachable" in result["error"]
+
+
+# ── Missing job ───────────────────────────────────────────────────────────────
+
+def test_push_returns_error_for_unknown_job_id(tmp_path):
+    from scripts.db import init_db
+    db = tmp_path / "test.db"
+    init_db(db)
+    config_dir = _config_dir_with(tmp_path, "apple_calendar")
+
+    from scripts.calendar_push import push_interview_event
+    result = push_interview_event(db, job_id=9999, config_dir=config_dir)
+
+    assert result["ok"] is False
+    assert "9999" in result["error"]