Add the EDS calendar bridge

config/dot/quickshell/scripts/calendar-agenda

@@ -0,0 +1,370 @@
+#!/usr/bin/env python3
+
+"""Read GNOME's Evolution Data Server calendars for the Panama shell."""
+
+from __future__ import annotations
+
+import json
+import re
+import sys
+import time
+from collections.abc import Iterable
+from typing import Any
+from urllib.parse import urlsplit
+
+try:
+    import gi
+
+    gi.require_version("ECal", "2.0")
+    gi.require_version("EDataServer", "1.2")
+    gi.require_version("ICalGLib", "3.0")
+    from gi.repository import ECal, EDataServer, Gio, GLib, ICalGLib
+
+    EDS_READY = True
+except (ImportError, ValueError):
+    ECal = EDataServer = Gio = GLib = ICalGLib = None
+    EDS_READY = False
+
+
+MEETING_HOST_SUFFIXES = (
+    "meet.google.com",
+    "zoom.us",
+    "teams.microsoft.com",
+    "teams.live.com",
+    "webex.com",
+    "meet.jit.si",
+)
+
+URL_PATTERN = re.compile(r"https?://[^\s<>\"']+", re.IGNORECASE)
+COLOR_PATTERN = re.compile(r"^#[0-9a-fA-F]{6}(?:[0-9a-fA-F]{2})?$")
+FALLBACK_COLOR = "#82aaff"
+
+
+def stable_event_id(source_id: str, uid: str, recurrence_id: str) -> str:
+    return f"{source_id}:{uid}:{recurrence_id}"
+
+
+def sort_events(events: list[dict[str, object]]) -> list[dict[str, object]]:
+    return sorted(
+        events,
+        key=lambda event: (
+            not bool(event["allDay"]),
+            int(event["start"]),
+            str(event["id"]),
+        ),
+    )
+
+
+def is_allowed_meeting_host(host: str) -> bool:
+    normalized = host.lower().rstrip(".")
+    return any(
+        normalized == suffix or normalized.endswith("." + suffix)
+        for suffix in MEETING_HOST_SUFFIXES
+    )
+
+
+def find_join_url(values: Iterable[str | None]) -> str:
+    for value in values:
+        if not value:
+            continue
+        for match in URL_PATTERN.findall(value):
+            candidate = match.rstrip("),.;!?]}")
+            try:
+                parsed = urlsplit(candidate)
+            except ValueError:
+                continue
+            if parsed.scheme.lower() != "https" or not parsed.hostname:
+                continue
+            if is_allowed_meeting_host(parsed.hostname):
+                return candidate
+    return ""
+
+
+def _safe_component_value(component: Any, getter: str) -> str:
+    try:
+        return (getattr(component, getter)() or "").strip()
+    except (AttributeError, TypeError):
+        return ""
+
+
+def normalize_instance(
+    source_meta: dict[str, str],
+    component: Any,
+    instance_start: Any,
+    instance_end: Any,
+) -> dict[str, object]:
+    uid = (component.get_uid() or "").strip()
+    if not uid:
+        raise ValueError("calendar component has no UID")
+
+    start = int(instance_start.as_timet())
+    end = int(instance_end.as_timet())
+    if end < start:
+        raise ValueError("calendar component ends before it starts")
+
+    recurrence = component.get_recurrenceid()
+    recurrence_id = str(start)
+    if recurrence is not None and not recurrence.is_null_time():
+        recurrence_id = recurrence.as_ical_string()
+
+    summary = _safe_component_value(component, "get_summary") or "Untitled event"
+    location = _safe_component_value(component, "get_location")
+    description = _safe_component_value(component, "get_description")
+    component_url = ""
+    try:
+        url_property = component.get_first_property(ICalGLib.PropertyKind.URL_PROPERTY)
+        if url_property is not None:
+            component_url = (url_property.get_url() or "").strip()
+    except (AttributeError, TypeError):
+        component_url = ""
+
+    all_day = bool(instance_start.is_date() or instance_end.is_date())
+    return {
+        "id": stable_event_id(source_meta["id"], uid, recurrence_id),
+        "sourceId": source_meta["id"],
+        "uid": uid,
+        "summary": summary,
+        "start": start,
+        "end": end,
+        "allDay": all_day,
+        "location": location,
+        "joinUrl": find_join_url([component_url, location, description]),
+    }
+
+
+def _source_meta(source: Any) -> dict[str, str]:
+    extension = source.get_extension(EDataServer.SOURCE_EXTENSION_CALENDAR)
+    color = extension.get_color() or FALLBACK_COLOR
+    if not COLOR_PATTERN.fullmatch(color):
+        color = FALLBACK_COLOR
+    return {
+        "id": source.get_uid(),
+        "name": (source.get_display_name() or "Calendar").strip(),
+        "color": color.lower(),
+    }
+
+
+def _unavailable_snapshot() -> dict[str, object]:
+    return {
+        "ok": False,
+        "generatedAt": int(time.time()),
+        "sources": [],
+        "events": [],
+        "errors": [{"code": "eds-unavailable"}],
+    }
+
+
+def probe() -> dict[str, object]:
+    if not EDS_READY:
+        return {"eds": False, "sourceRegistry": False, "enabledSources": 0}
+    try:
+        registry = EDataServer.SourceRegistry.new_sync(None)
+        sources = registry.list_enabled(EDataServer.SOURCE_EXTENSION_CALENDAR)
+    except Exception:
+        return {"eds": True, "sourceRegistry": False, "enabledSources": 0}
+    return {
+        "eds": True,
+        "sourceRegistry": True,
+        "enabledSources": len(sources),
+    }
+
+
+def collect_snapshot(start: int, end: int) -> dict[str, object]:
+    if not EDS_READY:
+        return _unavailable_snapshot()
+    if start < 0 or end <= start:
+        return {
+            **_unavailable_snapshot(),
+            "errors": [{"code": "invalid-range"}],
+        }
+
+    try:
+        registry = EDataServer.SourceRegistry.new_sync(None)
+        enabled_sources = registry.list_enabled(EDataServer.SOURCE_EXTENSION_CALENDAR)
+    except Exception:
+        return _unavailable_snapshot()
+
+    sources: list[dict[str, str]] = []
+    events_by_id: dict[str, dict[str, object]] = {}
+    errors: list[dict[str, str]] = []
+
+    for source in enabled_sources:
+        source_meta = _source_meta(source)
+        sources.append(source_meta)
+        malformed_event = False
+
+        def on_instance(
+            component: Any,
+            instance_start: Any,
+            instance_end: Any,
+            _cancellable: Any,
+            _callback_data: Any,
+        ) -> bool:
+            nonlocal malformed_event
+            try:
+                event = normalize_instance(
+                    source_meta,
+                    component,
+                    instance_start,
+                    instance_end,
+                )
+                events_by_id[str(event["id"])] = event
+            except (AttributeError, TypeError, ValueError):
+                malformed_event = True
+            return True
+
+        try:
+            client = ECal.Client.connect_sync(
+                source,
+                ECal.ClientSourceType.EVENTS,
+                5,
+                None,
+            )
+            if client is None:
+                raise RuntimeError("calendar client unavailable")
+            client.generate_instances_sync(start, end, None, on_instance, None)
+        except Exception:
+            errors.append({"sourceId": source_meta["id"], "code": "unavailable"})
+            continue
+
+        if malformed_event:
+            errors.append({"sourceId": source_meta["id"], "code": "event-invalid"})
+
+    sources.sort(key=lambda source: (source["name"].casefold(), source["id"]))
+    return {
+        "ok": True,
+        "generatedAt": int(time.time()),
+        "sources": sources,
+        "events": sort_events(list(events_by_id.values())),
+        "errors": errors,
+    }
+
+
+def _write_snapshot(snapshot: dict[str, object]) -> bool:
+    try:
+        sys.stdout.write(json.dumps(snapshot, separators=(",", ":")) + "\n")
+        sys.stdout.flush()
+    except BrokenPipeError:
+        return False
+    return True
+
+
+def watch(start: int, end: int) -> int:
+    if not EDS_READY:
+        _write_snapshot(_unavailable_snapshot())
+        return 1
+
+    loop = GLib.MainLoop()
+    registry = EDataServer.SourceRegistry.new_sync(None)
+    debounce_source = 0
+    subscriptions: list[int] = []
+
+    def emit_snapshot() -> bool:
+        nonlocal debounce_source
+        debounce_source = 0
+        if not _write_snapshot(collect_snapshot(start, end)):
+            loop.quit()
+        return GLib.SOURCE_REMOVE
+
+    def schedule_refresh(*_args: Any) -> None:
+        nonlocal debounce_source
+        if debounce_source:
+            GLib.source_remove(debounce_source)
+        debounce_source = GLib.timeout_add(350, emit_snapshot)
+
+    def periodic_refresh() -> bool:
+        emit_snapshot()
+        return GLib.SOURCE_CONTINUE
+
+    for signal_name in ("source-added", "source-removed", "source-changed"):
+        try:
+            registry.connect(signal_name, schedule_refresh)
+        except TypeError:
+            pass
+
+    try:
+        bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)
+        for signal_name in (
+            "EventsAddedOrUpdated",
+            "EventsRemoved",
+            "ClientDisappeared",
+        ):
+            subscriptions.append(
+                bus.signal_subscribe(
+                    "org.gnome.Shell.CalendarServer",
+                    "org.gnome.Shell.CalendarServer",
+                    signal_name,
+                    "/org/gnome/Shell/CalendarServer",
+                    None,
+                    Gio.DBusSignalFlags.NONE,
+                    schedule_refresh,
+                    None,
+                )
+            )
+        proxy = Gio.DBusProxy.new_sync(
+            bus,
+            Gio.DBusProxyFlags.NONE,
+            None,
+            "org.gnome.Shell.CalendarServer",
+            "/org/gnome/Shell/CalendarServer",
+            "org.gnome.Shell.CalendarServer",
+            None,
+        )
+        proxy.call_sync(
+            "SetTimeRange",
+            GLib.Variant("(xxb)", (start, end, True)),
+            Gio.DBusCallFlags.NONE,
+            -1,
+            None,
+        )
+    except Exception:
+        bus = None
+
+    if not _write_snapshot(collect_snapshot(start, end)):
+        return 0
+    GLib.timeout_add_seconds(600, periodic_refresh)
+
+    try:
+        loop.run()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        if bus is not None:
+            for subscription in subscriptions:
+                bus.signal_unsubscribe(subscription)
+    return 0
+
+
+def _usage_error() -> dict[str, object]:
+    return {
+        "ok": False,
+        "generatedAt": int(time.time()),
+        "sources": [],
+        "events": [],
+        "errors": [{"code": "usage"}],
+    }
+
+
+def main(argv: list[str]) -> int:
+    command = argv[1] if len(argv) > 1 else ""
+    if command == "probe" and len(argv) == 2:
+        _write_snapshot(probe())
+        return 0
+    if command in {"query", "watch"} and len(argv) == 4:
+        try:
+            start = int(argv[2])
+            end = int(argv[3])
+        except ValueError:
+            _write_snapshot(_usage_error())
+            return 64
+        if command == "query":
+            snapshot = collect_snapshot(start, end)
+            _write_snapshot(snapshot)
+            return 0 if snapshot["ok"] else 1
+        return watch(start, end)
+    _write_snapshot(_usage_error())
+    return 64
+
+
+if __name__ == "__main__":
+    raise SystemExit(main(sys.argv))

docs/superpowers/plans/2026-08-17-calendar-agenda.md

@@ -75,7 +75,7 @@
 - Produces: one-line JSON snapshots with `ok`, `generatedAt`, `sources`, `events`, and `errors`.
 - Produces event objects with exactly `id`, `sourceId`, `uid`, `summary`, `start`, `end`, `allDay`, `location`, and `joinUrl`.
 
-- [ ] **Step 1: Write failing pure bridge tests**
+- [x] **Step 1: Write failing pure bridge tests**
 
 Create `tests/quickshell/calendar_agenda_bridge_test.py` with import-by-path and synthetic policy cases:
 
@@ -148,7 +148,7 @@ if __name__ == "__main__":
     unittest.main()
 ```
 
-- [ ] **Step 2: Run the unit test and verify RED**
+- [x] **Step 2: Run the unit test and verify RED**
 
 Run:
 
@@ -158,7 +158,7 @@ python3 tests/quickshell/calendar_agenda_bridge_test.py
 
 Expected: FAIL because `config/dot/quickshell/scripts/calendar-agenda` does not exist.
 
-- [ ] **Step 3: Implement the bridge's pure contract and CLI boundary**
+- [x] **Step 3: Implement the bridge's pure contract and CLI boundary**
 
 Create `config/dot/quickshell/scripts/calendar-agenda` with a Python shebang and no top-level execution on import. Begin with these pure helpers:
 
@@ -217,7 +217,7 @@ Make the helper executable:
 chmod +x config/dot/quickshell/scripts/calendar-agenda
 ```
 
-- [ ] **Step 4: Run pure tests and verify GREEN**
+- [x] **Step 4: Run pure tests and verify GREEN**
 
 Run:
 
@@ -227,7 +227,7 @@ python3 tests/quickshell/calendar_agenda_bridge_test.py
 
 Expected: four tests PASS.
 
-- [ ] **Step 5: Add the redacted live helper contract**
+- [x] **Step 5: Add the redacted live helper contract**
 
 Create `tests/quickshell/calendar-agenda-helper-contract.sh` with:
 
@@ -260,7 +260,7 @@ printf 'calendar agenda helper contract: PASS (%s sources, %s events; contents r
 
 Make it executable and run it. Expected: PASS with counts only; no source names, summaries, locations, or URLs in terminal output.
 
-- [ ] **Step 6: Commit the bridge**
+- [x] **Step 6: Commit the bridge**
 
 ```bash
 git add config/dot/quickshell/scripts/calendar-agenda \

tests/quickshell/calendar-agenda-helper-contract.sh

@@ -0,0 +1,33 @@
+#!/usr/bin/env bash
+
+set -euo pipefail
+
+fail() {
+    printf 'calendar agenda helper contract: %s\n' "$1" >&2
+    exit 1
+}
+
+project_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+helper="$project_root/config/dot/quickshell/scripts/calendar-agenda"
+
+[[ -x "$helper" ]] || fail 'helper is missing or not executable'
+
+probe="$($helper probe)"
+jq -e '.eds == true and .sourceRegistry == true and .enabledSources > 0' <<<"$probe" >/dev/null \
+    || fail 'EDS readiness probe failed'
+
+range_start="$(date +%s)"
+range_end="$((range_start + 1209600))"
+snapshot="$($helper query "$range_start" "$range_end")"
+
+jq -e '
+    .ok == true and
+    (.sources | type == "array" and length > 0) and
+    (.events | type == "array") and
+    ([.sources[] | (keys | sort) == (["id", "name", "color"] | sort)] | all) and
+    ([.events[] | (keys | sort) == (["id", "sourceId", "uid", "summary", "start", "end", "allDay", "location", "joinUrl"] | sort)] | all)
+' <<<"$snapshot" >/dev/null || fail 'live snapshot shape is invalid'
+
+printf 'calendar agenda helper contract: PASS (%s sources, %s events; contents redacted)\n' \
+    "$(jq '.sources | length' <<<"$snapshot")" \
+    "$(jq '.events | length' <<<"$snapshot")"

tests/quickshell/calendar_agenda_bridge_test.py

@@ -0,0 +1,73 @@
+#!/usr/bin/env python3
+
+import importlib.machinery
+import importlib.util
+import pathlib
+import sys
+import unittest
+from unittest.mock import Mock
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+HELPER = ROOT / "config/dot/quickshell/scripts/calendar-agenda"
+sys.dont_write_bytecode = True
+
+loader = importlib.machinery.SourceFileLoader("calendar_agenda", str(HELPER))
+spec = importlib.util.spec_from_loader(loader.name, loader)
+calendar_agenda = importlib.util.module_from_spec(spec)
+loader.exec_module(calendar_agenda)
+
+
+class CalendarAgendaBridgeTest(unittest.TestCase):
+    def test_meeting_url_allowlist(self):
+        meeting_url = "https://meet.google.com/abc-defg-hij"
+
+        self.assertEqual(calendar_agenda.find_join_url([f"Planning: {meeting_url}"]), meeting_url)
+        self.assertEqual(calendar_agenda.find_join_url(["https://example.com/private"]), "")
+        self.assertEqual(calendar_agenda.find_join_url(["http://zoom.us/j/123"]), "")
+
+    def test_stable_instance_id_includes_recurrence(self):
+        self.assertEqual(
+            calendar_agenda.stable_event_id("source", "uid", "20260817T133000Z"),
+            "source:uid:20260817T133000Z",
+        )
+
+    def test_snapshot_sort_order(self):
+        events = [
+            {"id": "b", "start": 20, "end": 30, "allDay": False},
+            {"id": "a", "start": 10, "end": 20, "allDay": True},
+        ]
+
+        self.assertEqual([event["id"] for event in calendar_agenda.sort_events(events)], ["a", "b"])
+
+    def test_normalizes_an_all_day_recurrence_instance(self):
+        recurrence_id = Mock()
+        recurrence_id.is_null_time.return_value = False
+        recurrence_id.as_ical_string.return_value = "20260817"
+        component = Mock()
+        component.get_uid.return_value = "uid"
+        component.get_recurrenceid.return_value = recurrence_id
+        component.get_summary.return_value = "Synthetic event"
+        component.get_location.return_value = ""
+        component.get_description.return_value = ""
+        component.get_first_property.return_value = None
+        start = Mock()
+        start.as_timet.return_value = 1786924800
+        start.is_date.return_value = True
+        end = Mock()
+        end.as_timet.return_value = 1787011200
+        end.is_date.return_value = True
+
+        event = calendar_agenda.normalize_instance(
+            {"id": "source", "name": "Synthetic", "color": "#82aaff"},
+            component,
+            start,
+            end,
+        )
+
+        self.assertTrue(event["allDay"])
+        self.assertEqual(event["id"], "source:uid:20260817")
+
+
+if __name__ == "__main__":
+    unittest.main()