diff --git a/config/dot/quickshell/scripts/calendar-agenda b/config/dot/quickshell/scripts/calendar-agenda new file mode 100755 index 0000000..b22e8ca --- /dev/null +++ b/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)) diff --git a/docs/superpowers/plans/2026-08-17-calendar-agenda.md b/docs/superpowers/plans/2026-08-17-calendar-agenda.md index 6b88e00..e8f45e0 100644 --- a/docs/superpowers/plans/2026-08-17-calendar-agenda.md +++ b/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 \ diff --git a/tests/quickshell/calendar-agenda-helper-contract.sh b/tests/quickshell/calendar-agenda-helper-contract.sh new file mode 100755 index 0000000..7aa434c --- /dev/null +++ b/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")" diff --git a/tests/quickshell/calendar_agenda_bridge_test.py b/tests/quickshell/calendar_agenda_bridge_test.py new file mode 100644 index 0000000..8dee13b --- /dev/null +++ b/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()