snapshot of https://github.com/ActivityWatch/aw-watcher-window.git (master, 2026-09-16; blobs > 131072B omitted)

aw_watcher_window/__init__.py

@@ -0,0 +1,6 @@
+def main():
+    from .main import main as _main
+
+    return _main()
+
+__all__ = ["main"]

aw_watcher_window/__main__.py

@@ -0,0 +1,4 @@
+from aw_watcher_window import main
+
+if __name__ == "__main__":
+    main()

aw_watcher_window/config.py

@@ -0,0 +1,97 @@
+import argparse
+
+import tomlkit
+
+from aw_core.config import load_config_toml
+
+# Only end-user options belong here. load_config_toml() writes this template into
+# the user's config file on first run; it comments out keys but NOT table headers,
+# so anything listed here is authored into every fresh install's config.
+default_config = """
+[aw-watcher-window]
+exclude_title = false
+exclude_titles = []
+poll_time = 1.0
+strategy_macos = "swift"
+""".strip()
+
+# Research Edition defaults, deliberately kept out of default_config: these are
+# study/dev knobs, not end-user options, and should not be persisted into the
+# config of users who are not part of a study. They are still read normally when
+# a user does set them (see load_config), which is how the Research Edition build
+# and study participants configure them.
+#
+# The Research Edition release build rewrites the flag below with
+#   sed -i 's/^research_enabled = false$/research_enabled = true/'
+# (ActivityWatch/activitywatch, .github/workflows/release.yml). Keep that line at
+# column 0 and byte-identical.
+research_defaults = """
+research_enabled = false
+""".strip()
+
+
+def load_config():
+    config = load_config_toml("aw-watcher-window", default_config)["aw-watcher-window"]
+    # Research defaults fill in only what the user's config didn't set.
+    for key, value in tomlkit.parse(research_defaults).items():
+        config.setdefault(key, value)
+    return config
+
+
+def parse_args():
+    config = load_config()
+
+    default_poll_time = config["poll_time"]
+    default_exclude_title = config["exclude_title"]
+    default_exclude_titles = config["exclude_titles"]
+    default_strategy_macos = config["strategy_macos"]
+    default_research_enabled = config.get("research_enabled", False)
+
+    parser = argparse.ArgumentParser(
+        description="A cross platform window watcher for Activitywatch.\nSupported on: Linux (X11), macOS and Windows."
+    )
+    parser.add_argument("--host", dest="host")
+    parser.add_argument("--port", dest="port")
+    parser.add_argument("--testing", dest="testing", action="store_true")
+    parser.add_argument(
+        "--exclude-title",
+        dest="exclude_title",
+        action="store_true",
+        default=default_exclude_title,
+    )
+    parser.add_argument(
+        "--exclude-titles",
+        dest="exclude_titles",
+        nargs='+',
+        default=default_exclude_titles,
+        help="Exclude window titles by regular expression. Can specify multiple times."
+    )
+    parser.add_argument("--verbose", dest="verbose", action="store_true")
+    parser.add_argument(
+        "--poll-time", dest="poll_time", type=float, default=default_poll_time
+    )
+    parser.add_argument(
+        "--strategy",
+        dest="strategy",
+        default=default_strategy_macos,
+        choices=["jxa", "applescript", "swift"],
+        help="(macOS only) strategy to use for retrieving the active window",
+    )
+    research_group = parser.add_mutually_exclusive_group()
+    research_group.add_argument(
+        "--research",
+        dest="research_enabled",
+        action="store_true",
+        default=default_research_enabled,
+        help="Enable Research Edition mode: browser titles are classified into study categories, non-browser titles are dropped. Category map must be set in the config file.",
+    )
+    research_group.add_argument(
+        "--no-research",
+        dest="research_enabled",
+        action="store_false",
+        help="Disable Research Edition mode, even when enabled in the config file.",
+    )
+    parsed_args = parser.parse_args()
+    parsed_args.research_category_map = dict(config.get("research_category_map", {}))
+    parsed_args.research_app_category_map = dict(config.get("research_app_category_map", {}))
+    return parsed_args

aw_watcher_window/exceptions.py

@@ -0,0 +1,2 @@
+class FatalError(Exception):
+    """Exception raised when the program should exit with a fatal error."""

aw_watcher_window/lib.py

@@ -0,0 +1,81 @@
+import sys
+from typing import Optional
+
+from .exceptions import FatalError
+
+
+def get_current_window_linux() -> Optional[dict]:
+    from . import xlib
+
+    window = xlib.get_current_window()
+
+    if window is None:
+        cls = "unknown"
+        name = "unknown"
+    else:
+        cls = xlib.get_window_class(window)
+        name = xlib.get_window_name(window)
+
+    return {"app": cls, "title": name}
+
+
+def get_current_window_macos(strategy: str) -> Optional[dict]:
+    # TODO should we use unknown when the title is blank like the other platforms?
+
+    # `jxa` is the default & preferred strategy. It includes the url + incognito status
+    if strategy == "jxa":
+        from . import macos_jxa
+
+        return macos_jxa.getInfo()
+    elif strategy == "applescript":
+        from . import macos_applescript
+
+        return macos_applescript.getInfo()
+    else:
+        raise FatalError(f"invalid strategy '{strategy}'")
+
+
+def get_current_window_windows() -> Optional[dict]:
+    from . import windows
+
+    window_handle = windows.get_active_window_handle()
+
+    # hwnd 0 means no foreground window (e.g. during UAC prompt, lock screen,
+    # or secure desktop). Return None so heartbeat_loop skips this poll cycle.
+    if not window_handle:
+        return None
+
+    try:
+        app = windows.get_app_name(window_handle)
+    except Exception:
+        # Fall back to WMI for elevated/admin processes where OpenProcess fails
+        try:
+            app = windows.get_app_name_wmi(window_handle)
+        except Exception:
+            app = None
+
+    title = windows.get_window_title(window_handle)
+
+    if app is None:
+        app = "unknown"
+    if title is None:
+        title = "unknown"
+
+    return {"app": app, "title": title}
+
+
+def get_current_window(strategy: Optional[str] = None) -> Optional[dict]:
+    """
+    :raises FatalError: if a fatal error occurs (e.g. unsupported platform, X server closed)
+    """
+
+    if sys.platform.startswith("linux"):
+        return get_current_window_linux()
+    elif sys.platform == "darwin":
+        if strategy is None:
+            raise FatalError("macOS strategy not specified")
+        return get_current_window_macos(strategy)
+    elif sys.platform in ["win32", "cygwin"]:
+        return get_current_window_windows()
+    else:
+        raise FatalError(f"Unknown platform: {sys.platform}")

aw_watcher_window/macos.swift

@@ -0,0 +1,751 @@
+import Cocoa
+import ScriptingBridge
+
+@objc protocol ChromeTab {
+  @objc optional var URL: String { get }
+  @objc optional var title: String { get }
+}
+
+@objc protocol ChromeWindow {
+  @objc optional var activeTab: ChromeTab { get }
+  @objc optional var mode: String { get }
+}
+
+extension SBObject: ChromeWindow, ChromeTab {}
+
+@objc protocol ChromeProtocol {
+  @objc optional func windows() -> [ChromeWindow]
+}
+
+extension SBApplication: ChromeProtocol {}
+
+// https://github.com/tingraldi/SwiftScripting/blob/4346eba0f47e806943601f5fb2fe978e2066b310/Frameworks/SafariScripting/SafariScripting/Safari.swift#L37
+
+@objc public protocol SafariDocument {
+    @objc optional var name: String { get } // Its name.
+    @objc optional var modified: Bool { get } // Has it been modified since the last save?
+    @objc optional var file: URL { get } // Its location on disk, if it has one.
+    @objc optional var source: String { get } // The HTML source of the web page currently loaded in the document.
+    @objc optional var URL: String { get } // The current URL of the document.
+    @objc optional var text: String { get } // The text of the web page currently loaded in the document. Modifications to text aren't reflected on the web page.
+    @objc optional func setURL(_ URL: String!) // The current URL of the document.
+}
+
+@objc public protocol SafariTab {
+    @objc optional var source: String { get } // The HTML source of the web page currently loaded in the tab.
+    @objc optional var URL: String { get } // The current URL of the tab.
+    @objc optional var index: NSNumber { get } // The index of the tab, ordered left to right.
+    @objc optional var text: String { get } // The text of the web page currently loaded in the tab. Modifications to text aren't reflected on the web page.
+    @objc optional var visible: Bool { get } // Whether the tab is currently visible.
+    @objc optional var name: String { get } // The name of the tab.
+    @objc optional func setURL(_ URL: String!) // The current URL of the tab.
+}
+
+@objc public protocol SafariWindow {
+    @objc optional var name: String { get } // The title of the window.
+    @objc optional func id() -> Int // The unique identifier of the window.
+    @objc optional var index: Int { get } // The index of the window, ordered front to back.
+    @objc optional var document: SafariDocument { get } // The document whose contents are displayed in the window.
+    @objc optional func tabs() -> SBElementArray
+    @objc optional var currentTab: SafariTab { get } // The current tab.
+}
+extension SBObject: SafariWindow {}
+
+@objc public protocol SafariApplication {
+    @objc optional func documents() -> SBElementArray
+    @objc optional func windows() -> [SafariWindow]
+    @objc optional var name: String { get } // The name of the application.
+    @objc optional var frontmost: Bool { get } // Is this the active application?
+}
+extension SBApplication: SafariApplication {}
+
+// AW-specific structs
+
+struct NetworkMessage: Codable, Equatable {
+  var app: String
+  var title: String?
+  var url: String?
+}
+
+struct Heartbeat: Codable {
+  var timestamp: Date
+  var data: NetworkMessage
+}
+
+enum HeartbeatError: Error {
+  case error(msg: String)
+}
+
+struct Bucket: Codable {
+  var client: String
+  var type: String
+  var hostname: String
+}
+
+// there's no builtin logging library on macos which has levels & hits stdout, so we build our own simple one
+// there a complex open source one, but it makes it harder to compile this simple one-file swift application
+let dateFormatter =  DateFormatter();
+
+func logTimestamp() -> String {
+  let now = Date()
+  dateFormatter.timeZone = TimeZone.current
+  dateFormatter.dateFormat = "yyyy-MM-dd HH:mm:ss"
+  return dateFormatter.string(from: now)
+}
+
+// generate log prefix based on level
+func logPrefix(_ level: String) -> String {
+  return "\(logTimestamp()) [aw-watcher-window-macos] [\(level)]"
+}
+
+let logLevel = ProcessInfo.processInfo.environment["LOG_LEVEL"]?.uppercased() ?? "INFO"
+
+func debug(_ msg: String) {
+  if (logLevel == "DEBUG") {
+    print("\(logPrefix("DEBUG")) \(msg)")
+    fflush(stdout)
+  }
+}
+
+func log(_ msg: String) {
+  print("\(logPrefix("INFO")) \(msg)")
+  fflush(stdout)
+}
+
+func error(_ msg: String) {
+  print("\(logPrefix("ERROR")) \(msg)")
+  fflush(stdout)
+}
+
+// Placeholder values, set in start() from CLI arguments
+var baseurl = "http://localhost:5600"
+// NOTE: this differs from the hostname we get from Python, here we get `.local`, but in Python we get `.localdomain`
+var clientHostname = ProcessInfo.processInfo.hostName
+var clientName = "aw-watcher-window"
+var bucketName = "\(clientName)_\(clientHostname)"
+var excludeTitle = false
+var excludeTitlePatterns: [NSRegularExpression] = []
+var researchEnabled = false
+var researchCategoryMap: [(pattern: String, category: String)] = []
+var researchAppCategoryMap: [(app: String, category: String)] = []
+
+let researchBrowserApps = Set([
+  "chrome",
+  "google chrome",
+  "google chrome canary",
+  "google-chrome",
+  "google-chrome-beta",
+  "google-chrome-unstable",
+  "chromium",
+  "chromium-browser",
+  "chromium.exe",
+  "brave browser",
+  "brave",
+  "brave-browser",
+  "arc",
+  "arc browser",
+  "firefox",
+  "firefox developer edition",
+  "firefox-esr",
+  "safari",
+  "edge",
+  "microsoft edge",
+  "microsoft-edge",
+  "microsoft-edge-beta",
+  "microsoft-edge-dev",
+  "opera",
+  "vivaldi",
+  "vivaldi.exe",
+  "chrome.exe",
+  "brave.exe",
+  "firefox.exe",
+  "msedge.exe",
+  "opera.exe",
+])
+
+let main = MainThing()
+var oldHeartbeat: Heartbeat?
+
+let encoder = JSONEncoder()
+let formatter = ISO8601DateFormatter()
+formatter.formatOptions = [.withInternetDateTime, .withFractionalSeconds]
+
+encoder.dateEncodingStrategy = .custom({ date, encoder in
+  var container = encoder.singleValueContainer()
+  let dateString = formatter.string(from: date)
+  try container.encode(dateString)
+})
+
+start()
+RunLoop.main.run()
+
+func compileExcludeTitlePattern(_ pattern: String) -> NSRegularExpression {
+  do {
+    return try NSRegularExpression(pattern: pattern, options: [.caseInsensitive])
+  } catch let regexError {
+    error("Invalid regex pattern: \(pattern) — \(regexError.localizedDescription)")
+    exit(1)
+  }
+}
+
+func parseOptionalArguments(_ arguments: ArraySlice<String>) {
+  var index = arguments.startIndex
+  while index < arguments.endIndex {
+    let argument = arguments[index]
+
+    if argument == "--exclude-title" {
+      excludeTitle = true
+      index = arguments.index(after: index)
+      continue
+    }
+
+    if argument == "--exclude-titles" {
+      let nextIndex = arguments.index(after: index)
+      guard nextIndex < arguments.endIndex else {
+        error("Missing value for --exclude-titles")
+        exit(1)
+      }
+      excludeTitlePatterns.append(compileExcludeTitlePattern(arguments[nextIndex]))
+      index = arguments.index(after: nextIndex)
+      continue
+    }
+
+    if argument == "--research" {
+      researchEnabled = true
+      index = arguments.index(after: index)
+      continue
+    }
+
+    if argument == "--research-category" {
+      let patternIndex = arguments.index(after: index)
+      let categoryIndex = patternIndex < arguments.endIndex ? arguments.index(after: patternIndex) : arguments.endIndex
+      guard patternIndex < arguments.endIndex, categoryIndex < arguments.endIndex else {
+        error("Missing pattern/category values for --research-category")
+        exit(1)
+      }
+      researchCategoryMap.append((pattern: arguments[patternIndex], category: arguments[categoryIndex]))
+      index = arguments.index(after: categoryIndex)
+      continue
+    }
+
+    if argument == "--research-app-category" {
+      let appIndex = arguments.index(after: index)
+      let categoryIndex = appIndex < arguments.endIndex ? arguments.index(after: appIndex) : arguments.endIndex
+      guard appIndex < arguments.endIndex, categoryIndex < arguments.endIndex else {
+        error("Missing app/category values for --research-app-category")
+        exit(1)
+      }
+      researchAppCategoryMap.append((app: arguments[appIndex], category: arguments[categoryIndex]))
+      index = arguments.index(after: categoryIndex)
+      continue
+    }
+
+    error("Unknown argument: \(argument)")
+    exit(1)
+  }
+}
+
+func titleShouldBeExcluded(_ title: String) -> Bool {
+  let range = NSRange(title.startIndex..<title.endIndex, in: title)
+  return excludeTitlePatterns.contains { pattern in
+    pattern.firstMatch(in: title, options: [], range: range) != nil
+  }
+}
+
+func isResearchBrowser(_ app: String) -> Bool {
+  return researchBrowserApps.contains(app.trimmingCharacters(in: .whitespacesAndNewlines).lowercased())
+}
+
+// Return the category of the *longest* pattern contained in `haystack`.
+// Longest-match beats map order because study maps routinely contain both a
+// generic and a specific form of the same host ("google.com" vs
+// "docs.google.com"); first-match-wins would file every Google Docs/Drive/
+// Calendar/Meet URL under the generic entry. Ties resolve to map order.
+// Twin of `_longest_match` in research_filter.py — keep the two in sync.
+func longestResearchMatch(_ haystack: String) -> String? {
+  var bestCategory: String?
+  var bestLength = -1
+  for item in researchCategoryMap {
+    if item.pattern.isEmpty { continue }
+    // Compare by Unicode scalar count, not Character count: Swift's `count`
+    // measures grapheme clusters while Python's `len()` measures code points,
+    // which would let the two twins pick different winners for non-ASCII
+    // patterns of equal visual length.
+    let patternLength = item.pattern.unicodeScalars.count
+    if haystack.range(of: item.pattern, options: [.caseInsensitive]) != nil,
+      patternLength > bestLength
+    {
+      bestCategory = item.category
+      bestLength = patternLength
+    }
+  }
+  return bestCategory
+}
+
+func classifyResearch(_ title: String, url: String?) -> String {
+  // Try URL first — more reliable than page title, which can change mid-load
+  if let url = url, !url.isEmpty {
+    if let category = longestResearchMatch(url) {
+      return category
+    }
+  }
+  if let category = longestResearchMatch(title) {
+    return category
+  }
+  return "excluded"
+}
+
+func classifyApp(_ app: String) -> String {
+  // Case-insensitive exact lookup of app name in the app category map.
+  // Returns the mapped category, or "Excluded" when the app is not in the map.
+  // Both sides are trimmed and lowercased so a configured key carrying stray
+  // whitespace matches identically here and in the Python path
+  // (research_filter.classify_app). Without trimming the configured key, a map
+  // entry like " Microsoft Outlook" would classify on Linux/Windows but fall
+  // through to "Excluded" on the macOS Swift path for the same config.
+  let appLower = app.trimmingCharacters(in: .whitespacesAndNewlines).lowercased()
+  for item in researchAppCategoryMap {
+    if item.app.trimmingCharacters(in: .whitespacesAndNewlines).lowercased() == appLower {
+      return item.category
+    }
+  }
+  return "Excluded"
+}
+
+func applyResearchFilter(_ data: NetworkMessage) -> NetworkMessage {
+  if !researchEnabled {
+    return data
+  }
+
+  if isResearchBrowser(data.app) {
+    return NetworkMessage(app: data.app, title: classifyResearch(data.title ?? "", url: data.url), url: nil)
+  }
+
+  // Non-browser: map app to study category when a map is provided;
+  // otherwise keep app name and drop title (legacy behaviour).
+  if researchAppCategoryMap.isEmpty {
+    return NetworkMessage(app: data.app, title: nil, url: nil)
+  }
+  // Replace the raw app identity with its category — the app name is the
+  // sensitive identifier for non-browser apps, so it must not be retained.
+  return NetworkMessage(app: classifyApp(data.app), title: nil, url: nil)
+}
+
+func start() {
+  // Arguments should be:
+  //  - url + port
+  //  - bucket_id
+  //  - hostname
+  //  - client_id
+  let arguments = CommandLine.arguments
+
+  // Check that we get the 4 required arguments plus any optional flags
+  if arguments.count < 5 {
+    print("Usage: aw-watcher-window <url> <bucket> <hostname> <client> [--exclude-title] [--exclude-titles <pattern> ...] [--research] [--research-category <pattern> <category> ...] [--research-app-category <app_name> <category> ...]")
+    exit(1)
+  }
+
+  baseurl = arguments[1]
+  bucketName = arguments[2]
+  clientHostname = arguments[3]
+  clientName = arguments[4]
+  parseOptionalArguments(arguments.dropFirst(5))
+
+  guard checkAccess() else {
+    DispatchQueue.main.asyncAfter(deadline: .now() + 10) {
+      start()
+    }
+    return
+  }
+
+  createBucket()
+
+  // listen for changes in focused application
+  NSWorkspace.shared.notificationCenter.addObserver(
+    main,
+    selector: #selector(main.focusedAppChanged),
+    name: NSWorkspace.didActivateApplicationNotification,
+    object: nil
+  )
+
+  main.focusedAppChanged()
+
+  // Start the polling timer
+  main.pollingTimer = Timer.scheduledTimer(timeInterval: 10.0, target: main, selector: #selector(main.pollActiveWindow), userInfo: nil, repeats: true)
+}
+
+// TODO might be better to have the python wrapper create this before launching the swift application
+func createBucket() {
+  let payload = try! encoder.encode(
+    Bucket(client: clientName, type: "currentwindow", hostname: clientHostname))
+
+  let url = URL(string: "\(baseurl)/api/0/buckets/\(bucketName)")!
+  Task {
+    var urlRequest = URLRequest(url: url)
+    urlRequest.httpMethod = "POST"
+    urlRequest.addValue("application/json", forHTTPHeaderField: "Content-Type")
+    let (_, response) = try await URLSession.shared.upload(for: urlRequest, from: payload)
+    guard (200...299).contains((response as! HTTPURLResponse).statusCode) else {
+      log("Failed to create bucket")
+      return
+    }
+  }
+}
+
+func sendHeartbeat(_ heartbeat: Heartbeat) {
+  let oldPayloadDifferent = oldHeartbeat != nil && oldHeartbeat!.data != heartbeat.data
+  let timeSinceLastHeartbeat = oldHeartbeat != nil ? heartbeat.timestamp.timeIntervalSince(oldHeartbeat!.timestamp) : -1.0
+
+  // if you resize a window a ton of events (subsecond) will be fired
+  // we enforce a 1s minimum gap between events to avoid this
+  if timeSinceLastHeartbeat != -1.0 && timeSinceLastHeartbeat <= 0.5 {
+    debug("skipping heartbeat, last heartbeat was sent 1s ago")
+    return
+  }
+
+  // TODO running these async could cause weird state issues since the observer stuff can send a log of heartbeats
+  //      in a short time under certain circumstances, and we don't want to send them all
+  Task {
+    if oldPayloadDifferent {
+      debug("sending old heartbeat for merging")
+
+      do {
+        // unlike the python aw-client library, we do not enforce a `commit_interval` and instead send the old event (which is not invalid)
+        // at the current time with a pulse value equal to the time since this event was originally sent. The aw-server will then merge
+        // this new event with the original event, extending the recorded time spent on this particular window/application.
+
+        let refreshedOldHeartbeat = Heartbeat(
+          // it is important to refresh the hearbeat using the timestamp where the user stopped working on the previous application
+          // more info: https://github.com/ActivityWatch/aw-watcher-window/pull/69
+          // we don't *think* this millisecond subtraction is necessary, but it may be:
+          // https://github.com/ActivityWatch/aw-watcher-window/pull/69#discussion_r987064282
+          timestamp: heartbeat.timestamp - 0.001,
+          data: oldHeartbeat!.data
+        )
+
+        try await sendHeartbeatSingle(refreshedOldHeartbeat, pulsetime: timeSinceLastHeartbeat + 1)
+      } catch {
+        log("Failed to send old heartbeat: \(error)")
+        return
+      }
+    }
+
+    do {
+      let since_last_seconds = oldHeartbeat != nil ? heartbeat.timestamp.timeIntervalSince(oldHeartbeat!.timestamp) : 0
+      try await sendHeartbeatSingle(heartbeat, pulsetime: since_last_seconds + 1)
+    } catch {
+      log("Failed to send heartbeat: \(error)")
+      return
+    }
+
+    oldHeartbeat = heartbeat
+  }
+}
+
+func sendHeartbeatSingle(_ heartbeat: Heartbeat, pulsetime: Double) async throws {
+  let url = URL(string: "\(baseurl)/api/0/buckets/\(bucketName)/heartbeat?pulsetime=\(pulsetime)")!
+
+  var urlRequest = URLRequest(url: url)
+  urlRequest.httpMethod = "POST"
+  urlRequest.addValue("application/json", forHTTPHeaderField: "Content-Type")
+
+  let payload = try! encoder.encode(heartbeat)
+  let (_, response) = try await URLSession.shared.upload(for: urlRequest, from: payload)
+
+  guard (200...299).contains((response as! HTTPURLResponse).statusCode) else {
+    throw HeartbeatError.error(msg: "Failed to send heartbeat: \(response)")
+  }
+
+  debug("[heartbeat] bucket: \(bucketName), timestamp: \(heartbeat.timestamp), pulsetime: \(round(pulsetime * 10) / 10), app: \(heartbeat.data.app), title: \(heartbeat.data.title ?? ""), url: \(heartbeat.data.url ?? "")")
+}
+
+class MainThing {
+  var observer: AXObserver?
+  var oldWindow: AXUIElement?
+  var pollingTimer: Timer?
+
+  // list of chrome equivalent browsers
+  let CHROME_BROWSERS = [
+    "Google Chrome",
+    "Google Chrome Canary",
+    "Chromium",
+    "Brave Browser",
+  ]
+
+  // Gecko-based browsers have no scripting interface for tabs, but expose the
+  // current page URL on their accessibility tree's AXWebArea node
+  let FIREFOX_BROWSERS = [
+    "Firefox",
+    "Firefox Developer Edition",
+    "Firefox Nightly",
+    "Zen",
+    "Zen Browser",
+    "LibreWolf",
+    "Waterfox",
+    "Floorp",
+  ]
+
+  // upper bound on accessibility elements examined per lookup, so a
+  // pathological tree can't stall the watcher (the web area is typically
+  // found within a few dozen elements)
+  let AX_TRAVERSAL_LIMIT = 384
+
+  // Search the window's accessibility tree breadth-first for an AXWebArea node
+  // and return its AXURL, the URL of the loaded page. The top-level document's
+  // web area sits shallow in Gecko's tree and is reached before any web areas
+  // of nested iframes, so the first hit is the current page.
+  // (there is no kAXWebAreaRole constant in HIServices; the role string is
+  // defined by the browsers themselves)
+  func geckoURL(window: AXUIElement) -> String? {
+    var queue: [AXUIElement] = [window]
+    var index = 0
+    while index < queue.count && index < AX_TRAVERSAL_LIMIT {
+      let element = queue[index]
+      index += 1
+
+      var roleRef: AnyObject?
+      AXUIElementCopyAttributeValue(element, kAXRoleAttribute as CFString, &roleRef)
+      if roleRef as? String == "AXWebArea" {
+        var urlRef: AnyObject?
+        AXUIElementCopyAttributeValue(element, kAXURLAttribute as CFString, &urlRef)
+        if let url = urlRef as? NSURL {
+          return url.absoluteString
+        }
+        // no URL on the web area (e.g. page still loading); stop rather than
+        // keep searching, since a deeper hit would be an iframe's web area
+        return urlRef as? String
+      }
+
+      var childrenRef: AnyObject?
+      AXUIElementCopyAttributeValue(element, kAXChildrenAttribute as CFString, &childrenRef)
+      if let children = childrenRef as? [AXUIElement] {
+        queue.append(contentsOf: children)
+      }
+    }
+    return nil
+  }
+
+  @objc func pollActiveWindow() {
+    debug("Polling active window")
+
+    guard let frontmost = NSWorkspace.shared.frontmostApplication else {
+      log("Failed to get frontmost application from polling")
+      return
+    }
+
+    let pid = frontmost.processIdentifier
+    let focusedApp = AXUIElementCreateApplication(pid)
+
+    var focusedWindow: AnyObject?
+    AXUIElementCopyAttributeValue(focusedApp, kAXFocusedWindowAttribute as CFString, &focusedWindow)
+
+    if focusedWindow != nil {
+      focusedWindowChanged(observer!, window: focusedWindow as! AXUIElement)
+    }
+  }
+
+  deinit {
+    pollingTimer?.invalidate()
+  }
+
+  func windowTitleChanged(
+    _ axObserver: AXObserver,
+    axElement: AXUIElement,
+    notification: CFString
+  ) {
+    guard let frontmost = NSWorkspace.shared.frontmostApplication else {
+      log("Failed to get frontmost application from window title notification")
+      return
+    }
+
+    // calculate now before executing any scripting since that can take some time
+    let nowTime = Date.now
+
+    var windowTitle: AnyObject?
+    AXUIElementCopyAttributeValue(axElement, kAXTitleAttribute as CFString, &windowTitle)
+
+    let applicationName = frontmost.localizedName ?? frontmost.bundleIdentifier ?? ""
+    var data = NetworkMessage(app: applicationName, title: windowTitle as? String ?? "")
+
+    if CHROME_BROWSERS.contains(applicationName) {
+      debug("Chrome browser detected, extracting URL and title")
+
+      guard let bundleIdentifier = frontmost.bundleIdentifier else {
+        log("Failed to get bundle identifier from frontmost application, which was recognized to be Chrome")
+        return
+      }
+      let chromeObject: ChromeProtocol = SBApplication.init(bundleIdentifier: bundleIdentifier)!
+
+      guard let windows = chromeObject.windows,
+            let frontWindow = windows().first else {
+        log("Failed to get chrome front window")
+        return
+      }
+      guard let activeTab = frontWindow.activeTab else {
+        log("Failed to get chrome active tab")
+        return
+      }
+
+      if frontWindow.mode == "incognito" {
+        data = NetworkMessage(app: "", title: "")
+      } else {
+        data.url = activeTab.URL
+
+        // the tab title is more accurate and often different than the window title
+        // however, in some cases the binary does not have the right permissions to read
+        // the title properly and will return a blank string
+
+        if let tabTitle = activeTab.title {
+          if(tabTitle != "" && data.title != tabTitle) {
+            error("tab title diff: \(tabTitle), window title: \(data.title)")
+            data.title = tabTitle
+          }
+        }
+      }
+    } else if frontmost.localizedName == "Safari" {
+      debug("Safari browser detected, extracting URL and title")
+
+      guard let bundleIdentifier = frontmost.bundleIdentifier else {
+        log("Failed to get bundle identifier from frontmost application, which was recognized to be Safari")
+        return
+      }
+      let safariObject: SafariApplication = SBApplication.init(bundleIdentifier: bundleIdentifier)!
+
+      guard let windows = safariObject.windows,
+            let frontWindow = windows().first else {
+        log("Failed to get safari front window")
+        return
+      }
+      guard let activeTab = frontWindow.currentTab else {
+        log("Failed to get safari active tab")
+        return
+      }
+
+      // Safari doesn't allow incognito mode to be inspected, so we do not know if we should hide the url
+      data.url = activeTab.URL
+
+      // comment above applies here as well
+      if let tabTitle = activeTab.name {
+        if tabTitle != "" && data.title != tabTitle {
+          error("tab title diff: \(tabTitle), window title: \(data.title)")
+          data.title = tabTitle
+        }
+      }
+    } else if FIREFOX_BROWSERS.contains(applicationName) {
+      debug("Firefox-based browser detected, extracting URL from accessibility tree")
+
+      // note: private windows are not hidden here (unlike the Chrome incognito
+      // branch) — Gecko does not mark them in the accessibility tree, and their
+      // window titles carry a "Private Browsing" suffix for rules to match
+      data.url = geckoURL(window: axElement)
+
+      if data.url == nil {
+        // Newer Gecko builds instantiate their accessibility engine lazily and
+        // no longer treat plain tree walks as an assistive client, leaving the
+        // window's AX tree without any web content. Requesting
+        // AXEnhancedUserInterface (as VoiceOver does) turns the engine on; the
+        // call may report an error while the engine spins up, but the tree is
+        // populated for subsequent polls and stays on for the browser session.
+        let axApp = AXUIElementCreateApplication(frontmost.processIdentifier)
+        AXUIElementSetAttributeValue(axApp, "AXEnhancedUserInterface" as CFString, kCFBooleanTrue)
+      }
+    }
+
+    if researchEnabled {
+      data = applyResearchFilter(data)
+    } else if excludeTitle || titleShouldBeExcluded(data.title ?? "") {
+      data.title = "excluded"
+      // the URL identifies the page at least as precisely as the title does,
+      // so an excluded window must not report it either
+      data.url = nil
+    }
+
+    let heartbeat = Heartbeat(timestamp: nowTime, data: data)
+    sendHeartbeat(heartbeat)
+  }
+
+  @objc func focusedWindowChanged(_ observer: AXObserver, window: AXUIElement) {
+    debug("Focused window changed")
+
+    if oldWindow != nil {
+      AXObserverRemoveNotification(observer, oldWindow!, kAXFocusedWindowChangedNotification as CFString)
+    }
+
+    let selfPtr = UnsafeMutableRawPointer(Unmanaged.passUnretained(self).toOpaque())
+    AXObserverAddNotification(observer, window, kAXTitleChangedNotification as CFString, selfPtr)
+
+    windowTitleChanged(
+      observer, axElement: window, notification: kAXTitleChangedNotification as CFString)
+
+    oldWindow = window
+  }
+
+  @objc func focusedAppChanged() {
+    debug("Focused app changed")
+
+    if observer != nil {
+      CFRunLoopRemoveSource(
+        RunLoop.current.getCFRunLoop(),
+        AXObserverGetRunLoopSource(observer!),
+        CFRunLoopMode.defaultMode
+      )
+    }
+
+    guard let frontmost = NSWorkspace.shared.frontmostApplication else {
+      log("Failed to get frontmost application from app change notification")
+      return
+    }
+
+    let pid = frontmost.processIdentifier
+    let focusedApp = AXUIElementCreateApplication(pid)
+
+    AXObserverCreate(
+      pid,
+      {
+        (
+          _ axObserver: AXObserver,
+          axElement: AXUIElement,
+          notification: CFString,
+          userData: UnsafeMutableRawPointer?
+        ) -> Void in
+        guard let userData = userData else {
+          log("Missing userData")
+          return
+        }
+        let application = Unmanaged<MainThing>.fromOpaque(userData).takeUnretainedValue()
+        if notification == kAXFocusedWindowChangedNotification as CFString {
+          application.focusedWindowChanged(axObserver, window: axElement)
+        } else {
+          application.windowTitleChanged(
+            axObserver,
+            axElement: axElement,
+            notification: notification
+          )
+        }
+      }, &observer)
+
+    let selfPtr = UnsafeMutableRawPointer(Unmanaged.passUnretained(self).toOpaque())
+    AXObserverAddNotification(observer!, focusedApp, kAXFocusedWindowChangedNotification as CFString, selfPtr)
+
+    CFRunLoopAddSource(
+      RunLoop.current.getCFRunLoop(),
+      AXObserverGetRunLoopSource(observer!),
+      CFRunLoopMode.defaultMode
+    )
+
+    var focusedWindow: AnyObject?
+    AXUIElementCopyAttributeValue(focusedApp, kAXFocusedWindowAttribute as CFString, &focusedWindow)
+
+    if focusedWindow != nil {
+      focusedWindowChanged(observer!, window: focusedWindow as! AXUIElement)
+    }
+  }
+}
+
+// TODO I believe this is handled by the python wrapper so it isn't needed here
+func checkAccess() -> Bool {
+  let checkOptPrompt = kAXTrustedCheckOptionPrompt.takeUnretainedValue() as NSString
+  let options = [checkOptPrompt: true]
+  let accessEnabled = AXIsProcessTrustedWithOptions(options as CFDictionary?)
+  return accessEnabled
+}

aw_watcher_window/macos_applescript.py

@@ -0,0 +1,62 @@
+from typing import Dict
+from Foundation import NSAppleScript
+
+# the applescript version of the macos strategy is kept here until the jxa
+# approach is proven out in production environments
+# https://github.com/ActivityWatch/aw-watcher-window/pull/52
+
+source = """
+global frontApp, frontAppName, windowTitle
+
+set windowTitle to ""
+tell application "System Events"
+    set frontApp to first application process whose frontmost is true
+    set frontAppName to name of frontApp
+    tell process frontAppName
+        try
+            tell (1st window whose value of attribute "AXMain" is true)
+                set windowTitle to value of attribute "AXTitle"
+            end tell
+        end try
+    end tell
+end tell
+
+return frontAppName & "
+" & windowTitle
+"""
+
+script = None
+
+
+def getInfo() -> Dict[str, str]:
+    # Cache compiled script
+    global script
+    if script is None:
+        script = NSAppleScript.alloc().initWithSource_(source)
+
+    # Call script
+    result, errorinfo = script.executeAndReturnError_(None)
+    if errorinfo:
+        raise Exception(errorinfo)
+    output = result.stringValue()
+
+    # Ensure there's no extra newlines in the output
+    assert len(output.split("\n")) == 2
+
+    app = getApp(output)
+    title = getTitle(output)
+
+    return {"app": app, "title": title}
+
+
+def getApp(info: str) -> str:
+    return info.split('\n')[0]
+
+
+def getTitle(info: str) -> str:
+    return info.split('\n')[1]
+
+
+if __name__ == "__main__":
+    info = getInfo()
+    print(info)

aw_watcher_window/macos_cli.py

@@ -0,0 +1,23 @@
+def build_swift_command(
+    binpath,
+    server_address,
+    bucket_id,
+    client_hostname,
+    client_name,
+    exclude_title=False,
+    exclude_titles=None,
+    research_category_map=None,
+    research_app_category_map=None,
+):
+    command = [binpath, server_address, bucket_id, client_hostname, client_name]
+    if exclude_title:
+        command.append("--exclude-title")
+    for title in exclude_titles or []:
+        command.extend(["--exclude-titles", title])
+    if research_category_map is not None:
+        command.append("--research")
+        for pattern, category in research_category_map.items():
+            command.extend(["--research-category", pattern, category])
+        for app_name, category in (research_app_category_map or {}).items():
+            command.extend(["--research-app-category", app_name, category])
+    return command

aw_watcher_window/macos_jxa.py

@@ -0,0 +1,75 @@
+import os
+import json
+import logging
+from typing import Dict
+
+logger = logging.getLogger(__name__)
+script = None
+
+
+def compileScript():
+    """
+    Compiles the JXA script and caches the result.
+
+    Resources:
+     - https://stackoverflow.com/questions/44209057/how-can-i-run-jxa-from-swift
+     - https://stackoverflow.com/questions/16065162/calling-applescript-from-python-without-using-osascript-or-appscript
+    """
+
+    # use a global variable to cache the compiled script for performance
+    global script
+    if script:
+        return script
+
+    from OSAKit import OSAScript, OSALanguage
+
+    scriptPath = os.path.join(
+        os.path.dirname(os.path.realpath(__file__)), "printAppStatus.jxa"
+    )
+    with open(scriptPath) as f:
+        scriptContents = f.read()
+
+        # remove shebang line
+        if scriptContents.split("\n")[0].startswith("#"):
+            scriptContents = "\n".join(scriptContents.split("\n")[1:])
+
+    script = OSAScript.alloc().initWithSource_language_(
+        scriptContents, OSALanguage.languageForName_("JavaScript")
+    )
+    success, err = script.compileAndReturnError_(None)
+
+    # should only occur if jxa was modified incorrectly
+    if not success:
+        raise Exception(f"error compiling jxa script: {err['NSLocalizedDescription']}")
+
+    return script
+
+
+def getInfo() -> Dict[str, str]:
+    script = compileScript()
+
+    result, err = script.executeAndReturnError_(None)
+
+    if err:
+        # error structure:
+        # {
+        #     NSLocalizedDescription = "Error: Error: Can't get object.";
+        #     NSLocalizedFailureReason = "Error: Error: Can't get object.";
+        #     OSAScriptErrorBriefMessageKey = "Error: Error: Can't get object.";
+        #     OSAScriptErrorMessageKey = "Error: Error: Can't get object.";
+        #     OSAScriptErrorNumberKey = "-1728";
+        #     OSAScriptErrorRangeKey = "NSRange: {0, 0}";
+        # }
+
+        raise Exception(f"jxa error: {err['NSLocalizedDescription']}")
+
+    return json.loads(result.stringValue())
+
+
+if __name__ == "__main__":
+    print(getInfo())
+    print("Waiting 5 seconds...")
+    import time
+
+    time.sleep(5)
+    print(getInfo())

aw_watcher_window/macos_permissions.py

@@ -0,0 +1,42 @@
+import logging
+from multiprocessing import Process
+
+logger = logging.getLogger(__name__)
+
+
+def background_ensure_permissions() -> None:
+    permission_process = Process(target=ensure_permissions, args=(()))
+    permission_process.start()
+    return
+
+
+def ensure_permissions() -> None:
+    # noreorder
+    from AppKit import (  # fmt: skip
+        NSURL,
+        NSAlert,
+        NSAlertFirstButtonReturn,
+        NSWorkspace,
+    )
+    from ApplicationServices import AXIsProcessTrusted  # fmt: skip
+
+    accessibility_permissions = AXIsProcessTrusted()
+    if not accessibility_permissions:
+        logger.info("No accessibility permissions, prompting user")
+        title = "Missing accessibility permissions"
+        info = "To let ActivityWatch capture window titles grant it accessibility permissions. \n If you've already given ActivityWatch accessibility permissions and are still seeing this dialog, try removing and re-adding them."
+
+        alert = NSAlert.new()
+        alert.setMessageText_(title)
+        alert.setInformativeText_(info)
+
+        alert.addButtonWithTitle_("Open accessibility settings")
+        alert.addButtonWithTitle_("Close")
+
+        choice = alert.runModal()
+        if choice == NSAlertFirstButtonReturn:
+            NSWorkspace.sharedWorkspace().openURL_(
+                NSURL.URLWithString_(
+                    "x-apple.systempreferences:com.apple.preference.security?Privacy_Accessibility"
+                )
+            )

aw_watcher_window/main.py

@@ -0,0 +1,225 @@
+import logging
+import os
+import re
+import signal
+import subprocess
+import sys
+from datetime import datetime, timezone
+from time import sleep
+
+from aw_client import ActivityWatchClient
+from aw_core.log import setup_logging
+from aw_core.models import Event
+
+from .config import parse_args
+from .exceptions import FatalError
+from .lib import get_current_window
+from .research_filter import transform as research_transform
+from .macos_cli import build_swift_command
+from .macos_permissions import background_ensure_permissions
+
+logger = logging.getLogger(__name__)
+
+# run with LOG_LEVEL=DEBUG
+log_level = os.environ.get("LOG_LEVEL")
+if log_level:
+    logger.setLevel(logging.__getattribute__(log_level.upper()))
+
+
+def compute_pulsetime(poll_time: float) -> float:
+    """Scale pulsetime with poll_time so OS scheduling jitter doesn't break heartbeat chains.
+
+    At poll_time=1s, jitter ~0.15s is well within 1s margin (poll_time+1).
+    At poll_time=5s, jitter ~0.75s exceeds the 1s margin ~10% of the time,
+    causing missing time in the timeline. max(poll_time*1.5, poll_time+1) keeps
+    backward compatibility at poll_time≤2s while fixing the problem at higher
+    polling intervals. See: https://github.com/ActivityWatch/activitywatch/issues/1177
+    """
+    return max(poll_time * 1.5, poll_time + 1.0)
+
+
+def kill_process(pid):
+    logger.info("Killing process {}".format(pid))
+    try:
+        os.kill(pid, signal.SIGTERM)
+    except ProcessLookupError:
+        logger.info("Process {} already dead".format(pid))
+
+
+def try_compile_title_regex(title):
+    try:
+        return re.compile(title, re.IGNORECASE)
+    except re.error:
+        logger.error(f"Invalid regex pattern: {title}")
+        exit(1)
+
+
+def main():
+    args = parse_args()
+
+    if sys.platform.startswith("linux") and (
+        "DISPLAY" not in os.environ or not os.environ["DISPLAY"]
+    ):
+        raise Exception("DISPLAY environment variable not set")
+
+    setup_logging(
+        name="aw-watcher-window",
+        testing=args.testing,
+        verbose=args.verbose,
+        log_stderr=True,
+        log_file=True,
+    )
+    if sys.platform == "darwin":
+        background_ensure_permissions()
+
+    client = ActivityWatchClient(
+        "aw-watcher-window", host=args.host, port=args.port, testing=args.testing
+    )
+
+    bucket_id = f"{client.client_name}_{client.client_hostname}"
+    event_type = "currentwindow"
+
+    client.create_bucket(bucket_id, event_type, queued=True)
+
+    logger.info("aw-watcher-window started")
+    client.wait_for_start()
+
+    with client:
+        research_category_map = (
+            args.research_category_map
+            if args.research_enabled
+            else None
+        )
+        research_app_category_map = (
+            args.research_app_category_map
+            if args.research_enabled
+            else None
+        )
+        if sys.platform == "darwin" and args.strategy == "swift":
+            logger.info("Using swift strategy, calling out to swift binary")
+            binpath = os.path.join(
+                os.path.dirname(os.path.realpath(__file__)), "aw-watcher-window-macos"
+            )
+
+            try:
+                p = subprocess.Popen(
+                    build_swift_command(
+                        binpath,
+                        client.server_address,
+                        bucket_id,
+                        client.client_hostname,
+                        client.client_name,
+                        exclude_title=args.exclude_title,
+                        exclude_titles=args.exclude_titles,
+                        research_category_map=research_category_map,
+                        research_app_category_map=research_app_category_map,
+                    )
+                )
+                # terminate swift process when this process dies
+                signal.signal(signal.SIGTERM, lambda *_: kill_process(p.pid))
+                p.wait()
+            except KeyboardInterrupt:
+                print("KeyboardInterrupt")
+                kill_process(p.pid)
+        else:
+            heartbeat_loop(
+                client,
+                bucket_id,
+                poll_time=args.poll_time,
+                strategy=args.strategy,
+                exclude_title=args.exclude_title,
+                exclude_titles=[
+                    try_compile_title_regex(title)
+                    for title in args.exclude_titles
+                    if title is not None
+                ],
+                research_category_map=research_category_map,
+                research_app_category_map=research_app_category_map,
+            )
+
+
+def heartbeat_loop(
+    client,
+    bucket_id,
+    poll_time,
+    strategy,
+    exclude_title=False,
+    exclude_titles=[],
+    research_category_map=None,
+    research_app_category_map=None,
+):
+    while True:
+        if os.getppid() == 1:
+            logger.info("window-watcher stopped because parent process died")
+            break
+
+        current_window = None
+        try:
+            current_window = get_current_window(strategy)
+            logger.debug(current_window)
+        except (FatalError, OSError):
+            # Fatal exceptions should quit the program
+            try:
+                logger.exception("Fatal error, stopping")
+            except OSError:
+                pass
+            break
+        except Exception:
+            # Non-fatal exceptions should be logged
+            try:
+                # If stdout has been closed, this exception-print can cause (I think)
+                #   OSError: [Errno 5] Input/output error
+                # See: https://github.com/ActivityWatch/activitywatch/issues/756#issue-1296352264
+                #
+                # However, I'm unable to reproduce the OSError in a test (where I close stdout before logging),
+                # so I'm in uncharted waters here... but this solution should work.
+                logger.exception("Exception thrown while trying to get active window")
+            except OSError:
+                break
+
+        if current_window is None:
+            logger.debug("Unable to fetch window, trying again on next poll")
+        else:
+            current_window = transform_window(
+                current_window,
+                exclude_title=exclude_title,
+                exclude_titles=exclude_titles,
+                research_category_map=research_category_map,
+                research_app_category_map=research_app_category_map,
+            )
+
+            now = datetime.now(timezone.utc)
+            current_window_event = Event(timestamp=now, data=current_window)
+
+            client.heartbeat(
+                bucket_id,
+                current_window_event,
+                pulsetime=compute_pulsetime(poll_time),
+                queued=True,
+            )
+
+        sleep(poll_time)
+
+
+def transform_window(
+    current_window,
+    exclude_title=False,
+    exclude_titles=None,
+    research_category_map=None,
+    research_app_category_map=None,
+):
+    if research_category_map is not None:
+        return research_transform(
+            current_window,
+            research_category_map,
+            app_category_map=research_app_category_map,
+        )
+
+    for pattern in exclude_titles or []:
+        if pattern.search(current_window["title"]):
+            current_window["title"] = "excluded"
+
+    if exclude_title:
+        current_window["title"] = "excluded"
+
+    return current_window

aw_watcher_window/printAppStatus.jxa

@@ -0,0 +1,65 @@
+#!/usr/bin/osascript -l JavaScript
+
+// adapted from:
+// https://gist.github.com/EvanLovely/cb01eafb0d61515c835ecd56f6ac199a
+
+// new to jxa?
+// - https://apple-dev.groups.io/g/jxa/wiki/3202
+// - interactive repl: `osascript -il JavaScript`
+// - API reference: Script Editor -> File -> Open Dictionary
+
+var seApp         = Application("System Events");
+var oProcess      = seApp.processes.whose({frontmost: true})[0];
+var appName       = oProcess.displayedName();
+
+// as of 05/01/21 incognio & url are not actively used in AW
+// variables must be set to `undefined` since this script is re-run via osascript
+// and the previously set values will be cached otherwise
+var url = undefined, incognito = undefined, title = undefined;
+
+// it's not possible to get the URL from firefox
+// https://stackoverflow.com/questions/17846948/does-firefox-offer-applescript-support-to-get-url-of-windows
+
+switch(appName) {
+  case "Safari":
+    // incognito is not available via safari applescript
+    url = Application(appName).documents[0].url();
+    title = Application(appName).documents[0].name();
+    break;
+  case "Google Chrome":
+  case "Google Chrome Canary":
+  case "Chromium":
+  case "Brave Browser":
+    const activeWindow = Application(appName).windows[0];
+    const activeTab = activeWindow.activeTab();
+
+    url = activeTab.url();
+    title = activeTab.name();
+    incognito = activeWindow.mode() === 'incognito';
+    break;
+  case "Firefox":
+  case "Firefox Developer Edition":
+    title = Application(appName).windows[0].name();
+    break;
+  default:
+    mainWindow = oProcess.
+      windows().
+      find(w => w.attributes.byName("AXMain").value() === true)
+
+    // in some cases, the primary window of an application may not be found
+    // this occurs rarely and seems to be triggered by switching to a different application
+    if(mainWindow) {
+      title = mainWindow.
+        attributes.
+        byName("AXTitle").
+        value()
+    }
+}
+
+// key names must match expected names in lib.py
+JSON.stringify({
+  app: appName,
+  url,
+  title,
+  incognito
+});

aw_watcher_window/research_filter.py

@@ -0,0 +1,186 @@
+"""
+Research Edition: privacy-sensitive window-title → category transform.
+
+Replicates the approach used by AW research forks:
+- Browser apps: window titles get classified into study categories.
+  Unmatched titles → 'excluded'.
+- Non-browser apps: app name is mapped to a broad study category where
+  possible, otherwise replaced with 'Excluded'.  Supply the mapping in
+  ``[aw-watcher-window.research_app_category_map]``.  Without a map the
+  previous behaviour is preserved (app name kept, title dropped).
+
+Enable via config (``research_enabled = true``) and optionally supply a
+category map in ``[aw-watcher-window.research_category_map]``.  If the map
+is empty, browser titles are classified as ``excluded`` and non-browser
+behaviour falls back to title-drop.
+
+macOS note: on macOS the default ``swift`` strategy captures browser URLs via
+accessibility APIs and includes them in the window dict.  This filter uses those
+URLs for more reliable category classification (``classify_title`` prefers URL
+over title) and strips them from the output before the event is recorded.
+"""
+
+from typing import Optional
+
+# Known browser applications (lowercase, for case-insensitive matching)
+BROWSER_APPS = frozenset(
+    {
+        "chrome",
+        "google chrome",
+        "google chrome canary",
+        "google-chrome",
+        "google-chrome-beta",
+        "google-chrome-unstable",
+        "chromium",
+        "chromium-browser",
+        "chromium.exe",
+        "brave browser",
+        "brave",
+        "brave-browser",
+        "arc",
+        "arc browser",
+        "firefox",
+        "firefox developer edition",
+        "firefox-esr",
+        "safari",
+        "edge",
+        "microsoft edge",
+        "microsoft-edge",
+        "microsoft-edge-beta",
+        "microsoft-edge-dev",
+        "opera",
+        "vivaldi",
+        "vivaldi.exe",
+        "chrome.exe",
+        "brave.exe",
+        "firefox.exe",
+        "msedge.exe",
+        "opera.exe",
+    }
+)
+
+
+def is_browser(app: str) -> bool:
+    """Return True if *app* is a known browser (case-insensitive)."""
+    return app.strip().lower() in BROWSER_APPS
+
+
+def _longest_match(haystack: str, category_map: dict) -> Optional[str]:
+    """
+    Return the category of the *longest* pattern contained in *haystack*.
+
+    Longest-match beats insertion order because study maps routinely contain
+    both a generic and a specific form of the same host: ``google.com`` →
+    *Search & Navigation* and ``docs.google.com`` → *Work & Productivity*.
+    First-match-wins would file every Google Docs/Drive/Calendar/Meet URL as
+    Search, since ``google.com`` is a substring of all of them.
+
+    Ties (equal-length patterns) resolve to the first one in map order.
+    """
+    best_category: Optional[str] = None
+    best_len = -1
+    for pattern, category in category_map.items():
+        pattern_lower = pattern.lower()
+        if pattern_lower and pattern_lower in haystack and len(pattern_lower) > best_len:
+            best_category = category
+            best_len = len(pattern_lower)
+    return best_category
+
+
+def classify_title(title: str, category_map: dict, url: str = "") -> str:
+    """
+    Classify a browser window into a study category via substring matching.
+
+    Tries *url* first when provided (more reliable than page titles, which can
+    change mid-load), then falls back to *title*.
+    *category_map*: ``{substring: category_name}`` — the **longest** matching
+    substring wins, so a specific host (``docs.google.com``) is not shadowed by
+    a generic one (``google.com``) that happens to be listed earlier.
+    Returns the category name, or ``"excluded"`` if nothing matched.
+    """
+    if url:
+        category = _longest_match(url.lower(), category_map)
+        if category is not None:
+            return category
+    category = _longest_match(title.lower(), category_map)
+    if category is not None:
+        return category
+    return "excluded"
+
+
+def classify_app(app: str, app_category_map: dict) -> str:
+    """
+    Classify a non-browser application into a study category.
+
+    Performs a case-insensitive exact lookup of *app* in *app_category_map*
+    (``{app_name: category_name}``).  Both the lookup key and the configured
+    keys are normalized to lowercase, so mixed-case map keys (e.g. ``"Microsoft
+    Outlook"``) match their lowercase application names.  Returns the mapped
+    category, or ``"Excluded"`` when the app is not in the map.
+
+    Note: uses capital-E ``"Excluded"`` to match Matthias's original classifier
+    convention (``EXCLUDED = "Excluded"``), distinct from the lowercase
+    ``"excluded"`` used for browser title mismatches.
+    """
+    app_lower = app.strip().lower()
+    for map_app, category in app_category_map.items():
+        if map_app.strip().lower() == app_lower:
+            return category
+    return "Excluded"
+
+
+def transform(
+    window: dict,
+    category_map: Optional[dict],
+    app_category_map: Optional[dict] = None,
+) -> dict:
+    """
+    Apply Research Edition transforms to a window-data dict.
+
+    *window*: ``{"app": str, "title": str}`` (may also carry *url*,
+    *incognito* on macOS JXA — URLs are stripped and incognito is preserved).
+    *category_map*: study-specific URL/title mapping dict, or ``None`` to no-op.
+    *app_category_map*: optional app-name → category mapping for non-browser
+    apps.  When supplied, non-browser apps are replaced with their mapped
+    category (or ``"Excluded"`` if unmapped) instead of keeping the raw app name.
+
+    Returns the transformed window dict (never mutates the input).
+    """
+    if category_map is None:
+        return window
+
+    app = window.get("app", "")
+
+    if is_browser(app):
+        # Browser: replace title with a study category
+        title = window.get("title", "")
+        url = window.get("url", "")
+        category = classify_title(title, category_map, url=url)
+        # Don't spread window dict — URLs must not be exposed for privacy
+        result = {"app": app, "title": category}
+        # Preserve incognito flag if present (metadata, not a privacy concern)
+        if "incognito" in window:
+            result["incognito"] = window["incognito"]
+        return result
+    else:
+        # Non-browser: map app to a study category when a map is provided,
+        # otherwise keep the raw app name and drop the title.
+        # An empty map is treated as "not provided" — so research_enabled with
+        # no configured app map falls back to legacy behaviour instead of
+        # classifying every non-browser app as "Excluded".
+        if app_category_map:
+            app_category = classify_app(app, app_category_map)
+            # Replace the raw app identity with its category: the app name is
+            # the sensitive identifier for non-browser apps (e.g. "Microsoft
+            # Outlook", a niche tool, a company-internal app), so it must not
+            # be retained. Title/URL are dropped entirely.
+            result = {"app": app_category}
+        else:
+            # Legacy behaviour: title dropped, app name kept as-is.
+            result = {"app": app}
+        # Preserve incognito flag if present (metadata, not a privacy concern)
+        # Note: URL is NOT preserved for non-browser apps — on macOS JXA,
+        # apps like Mail or file managers can expose sensitive URLs (mailbox://, file://)
+        if "incognito" in window:
+            result["incognito"] = window["incognito"]
+        return result

aw_watcher_window/windows.py

@@ -0,0 +1,89 @@
+import os
+import time
+from typing import Optional
+
+import win32api
+import win32gui
+import win32process
+import wmi
+
+
+def get_app_path(hwnd) -> Optional[str]:
+    """Get application path given hwnd."""
+    path = None
+
+    _, pid = win32process.GetWindowThreadProcessId(hwnd)
+    process = win32api.OpenProcess(
+        0x0400, False, pid
+    )  # PROCESS_QUERY_INFORMATION = 0x0400
+
+    try:
+        path = win32process.GetModuleFileNameEx(process, 0)
+    finally:
+        win32api.CloseHandle(process)
+
+    return path
+
+
+def get_app_name(hwnd) -> Optional[str]:
+    """Get application filename given hwnd."""
+    path = get_app_path(hwnd)
+
+    if path is None:
+        return None
+
+    return os.path.basename(path)
+
+
+def get_window_title(hwnd):
+    return win32gui.GetWindowText(hwnd)
+
+
+def get_active_window_handle():
+    hwnd = win32gui.GetForegroundWindow()
+    return hwnd
+
+
+# WMI-version, used as fallback if win32gui/win32process/win32api fails (such as for "run as admin" processes)
+
+c = wmi.WMI()
+
+"""
+Much of this derived from: http://stackoverflow.com/a/14973422/965332
+"""
+
+
+def get_app_name_wmi(hwnd) -> Optional[str]:
+    """Get application filename given hwnd."""
+    name = None
+    _, pid = win32process.GetWindowThreadProcessId(hwnd)
+    for p in c.query("SELECT Name FROM Win32_Process WHERE ProcessId = %s" % str(pid)):
+        name = p.Name
+        break
+    return name
+
+
+def get_app_path_wmi(hwnd) -> Optional[str]:
+    """Get application path given hwnd."""
+    path = None
+
+    _, pid = win32process.GetWindowThreadProcessId(hwnd)
+    for p in c.query(
+        "SELECT ExecutablePath FROM Win32_Process WHERE ProcessId = %s" % str(pid)
+    ):
+        path = p.ExecutablePath
+        break
+
+    return path
+
+
+if __name__ == "__main__":
+    while True:
+        hwnd = get_active_window_handle()
+        print("Title:", get_window_title(hwnd))
+        print("App:        ", get_app_name(hwnd))
+        print("App (wmi):  ", get_app_name_wmi(hwnd))
+        print("Path:       ", get_app_path(hwnd))
+        print("Path (wmi): ", get_app_path_wmi(hwnd))
+
+        time.sleep(1.0)

aw_watcher_window/xlib.py

@@ -0,0 +1,174 @@
+import logging
+from typing import Optional
+
+import Xlib
+import Xlib.display
+from Xlib import X
+from Xlib.xobject.drawable import Window
+
+from .exceptions import FatalError
+
+logger = logging.getLogger(__name__)
+
+display = Xlib.display.Display()
+screen = display.screen()
+
+NET_WM_NAME = display.intern_atom("_NET_WM_NAME")
+UTF8_STRING = display.intern_atom("UTF8_STRING")
+
+
+def _get_current_window_id() -> Optional[int]:
+    atom = display.get_atom("_NET_ACTIVE_WINDOW")
+    window_prop = screen.root.get_full_property(atom, X.AnyPropertyType)
+
+    if window_prop is None:
+        logger.warning("window_prop was None")
+        return None
+
+    # window_prop may contain more than one value, but it seems that it's always the first we want.
+    # The second has in my attempts always been 0 or rubbish.
+    window_id = window_prop.value[0]
+    return window_id if window_id != 0 else None
+
+
+def _get_window(window_id: int) -> Window:
+    return display.create_resource_object("window", window_id)
+
+
+def get_current_window() -> Optional[Window]:
+    """
+    Returns the current window, or None if no window is active.
+    """
+    try:
+        window_id = _get_current_window_id()
+        if window_id is None:
+            return None
+        else:
+            return _get_window(window_id)
+    except Xlib.error.ConnectionClosedError:
+        # when the X server closes the connection, we should exit
+        # note that stdio is probably closed at this point, so we can't print anything (causes OSError)
+        try:
+            logger.warning("X server closed connection, exiting")
+        except OSError:
+            pass
+        raise FatalError()
+
+
+# Things that can lead to unknown cls/name:
+#  - (cls+name) Empty desktop in xfce (no window focused)
+#  - (name) Chrome (fixed, didn't support when WM_NAME was UTF8_STRING)
+
+
+def get_window_name(window: Window) -> str:
+    """After some annoying debugging I resorted to pretty much copying selfspy.
+    Source: https://github.com/gurgeh/selfspy/blob/8a34597f81000b3a1be12f8cde092a40604e49cf/selfspy/sniff_x.py#L165"""
+    try:
+        d = window.get_full_property(NET_WM_NAME, UTF8_STRING)
+    except Xlib.error.XError as e:
+        logger.warning(
+            f"Unable to get window property NET_WM_NAME, got a {type(e).__name__} exception from Xlib"
+        )
+        # I strongly suspect window.get_wm_name() will also fail and we should return "unknown" right away.
+        # But I don't know, so I pass the thing on, for now.
+        d = None
+    if d is None or d.format != 8:
+        try:
+            # Fallback.
+            r = window.get_wm_name()
+            if r is None:
+                return "unknown"
+            elif isinstance(r, str):
+                return r
+            else:
+                logger.warning(
+                    "I don't think this case will ever happen, but not sure so leaving this message here just in case."
+                )
+                return r.decode("latin1")  # WM_NAME with type=STRING.
+        except Xlib.error.BadWindow as e:
+            # I comment on the log, the number of messages that pop up is very annoying
+            # Also, it does not give much information about the error
+            # But I leave it in case someone sees it useful
+            # logger.warning(
+            #     f"Unable to get window property WM_NAME, got a {type(e).__name__} exception from Xlib"
+            # )
+            return "unknown"
+    else:
+        # Fixing utf8 issue on Ubuntu (https://github.com/gurgeh/selfspy/issues/133)
+        # Thanks to https://github.com/gurgeh/selfspy/issues/133#issuecomment-142943681
+        try:
+            return d.value.decode("utf8")
+        except UnicodeError:
+            logger.warning(
+                f"Failed to decode one or more characters which will be skipped, bytes are: {d.value}"
+            )
+            if isinstance(d.value, bytes):
+                return d.value.decode("utf8", "ignore")
+            else:
+                return d.value.encode("utf8").decode("utf8", "ignore")
+
+
+def get_window_class(window: Window) -> str:
+    cls = None
+
+    try:
+        cls = window.get_wm_class()
+    except Xlib.error.BadWindow:
+        logger.warning("Unable to get window class, got a BadWindow exception.")
+
+    # TODO: Is this needed?
+    # nikanar: Indeed, it seems that it is. But it would be interesting to see how often this succeeds, and if it is low, maybe fail earlier.
+    if not cls:
+        print("")
+        logger.warning("Code made an unclear branch")
+        try:
+            window = window.query_tree().parent
+        except Xlib.error.BadWindow:
+            # I comment on the log, the number of messages that pop up is very annoying
+            # Also, it does not give much information about the error
+            # But I leave it in case someone sees it useful
+            # logger.warning(
+            #     "Unable to get window query_tree().parent, got a BadWindow exception."
+            # )
+            return "unknown"
+        except Xlib.error.XError as e:
+            logger.warning(
+                f"Unable to get window query_tree().parent, got a {type(e).__name__} exception from Xlib"
+            )
+            return "unknown"
+        if window:
+            return get_window_class(window)
+        else:
+            return "unknown"
+
+    cls = cls[1]
+    return cls
+
+
+def get_window_pid(window: Window) -> str:
+    atom = display.get_atom("_NET_WM_PID")
+    pid_property = window.get_full_property(atom, X.AnyPropertyType)
+    if pid_property:
+        pid = pid_property.value[-1]
+        return pid
+    else:
+        # TODO: Needed?
+        raise Exception("pid_property was None")
+
+
+if __name__ == "__main__":
+    from time import sleep
+
+    while True:
+        print("-" * 20)
+        window = get_current_window()
+        if window is None:
+            print("unable to get active window")
+            name, cls = "unknown", "unknown"
+        else:
+            cls = get_window_class(window)
+            name = get_window_name(window)
+        print("name:", name)
+        print("class:", cls)
+
+        sleep(1)

aw_watcher_window/xprop.py

@@ -0,0 +1,128 @@
+import sys
+import subprocess
+import re
+import logging
+from subprocess import PIPE
+from typing import List
+
+logger = logging.getLogger(__name__)
+
+# req_version is 3.5 due to usage of subprocess.run
+# It would be nice to be able to use 3.4 as well since it's still common as of May 2016
+req_version = (3, 5)
+cur_version = sys.version_info
+
+if not cur_version >= req_version:
+    logger.error("Your Python version is too old, 3.5 or higher is required.")
+    exit(1)
+
+
+def xprop_id(window_id) -> str:
+    cmd = ["xprop"]
+    cmd.append("-id")
+    cmd.append(window_id)
+    p = subprocess.run(cmd, stdout=PIPE)
+    return str(p.stdout, "utf8")
+
+
+def xprop_root() -> str:
+    cmd = ["xprop"]
+    cmd.append("-root")
+    p = subprocess.run(cmd, stdout=PIPE)
+    return str(p.stdout, "utf8")
+
+
+def get_active_window_id():
+    lines = xprop_root().split("\n")
+    match="_NET_ACTIVE_WINDOW(WINDOW)"
+    result = None
+    for line in lines:
+        if match in line:
+            result = line
+            break
+    wid = "0x0"
+    if result:
+        wids = re.findall("0x[0-9a-f]*", result)
+        if len(wids) > 0:
+            wid = wids[0]
+    return wid
+
+
+def get_window_ids():
+    lines = xprop_root().split("\n")
+    client_list = next(filter(lambda x: "_NET_CLIENT_LIST(" in x, lines))
+    window_ids = re.findall("0x[0-9a-f]*", client_list)
+    return window_ids
+
+
+def _extract_xprop_field(line):
+    return "".join(line.split("=")[1:]).strip(" \n")
+
+
+def get_xprop_field(fieldname, xprop_output):
+    return list(map(_extract_xprop_field, re.findall(fieldname + ".*\n", xprop_output)))
+
+
+def get_xprop_field_str(fieldname, xprop_output) -> str:
+    field = None
+    try:
+        field = get_xprop_field(fieldname, xprop_output)[0].strip('"')
+    except IndexError:
+        pass
+    if not field:
+        field = "unknown"
+    return field
+
+
+def get_xprop_field_strlist(fieldname, xprop_output) -> List[str]:
+    return [s.strip('"') for s in get_xprop_field(fieldname, xprop_output)]
+
+
+def get_xprop_field_int(fieldname, xprop_output) -> int:
+    field = None
+    try:
+        field = int(get_xprop_field(fieldname, xprop_output)[0])
+    except IndexError:
+        pass
+    if not field:
+        field = -1
+    return field
+
+
+def get_xprop_field_class(xprop_output) -> List[str]:
+    classname: List[str] = []
+    try:
+        classname = [c.strip('", ') for c in get_xprop_field("WM_CLASS", xprop_output)[0].split(',')]
+    except IndexError:
+        pass
+    if not classname:
+        classname = ["unknown"]
+    return classname
+
+
+def get_window(wid, active_window=False):
+    s = xprop_id(wid)
+    window = {
+        "id": wid,
+        "active": active_window,
+        "name": get_xprop_field_str("WM_NAME", s),
+        "class": get_xprop_field_class(s),
+        "desktop": get_xprop_field_int("WM_DESKTOP", s),
+        "command": get_xprop_field("WM_COMMAND", s),
+        "role": get_xprop_field_strlist("WM_WINDOW_ROLE", s),
+        "pid": get_xprop_field_int("WM_PID", s),
+    }
+
+    return window
+
+
+def get_windows(wids, active_window_id=None):
+    return [get_window(wid, active_window=(wid == active_window_id)) for wid in wids]
+
+
+if __name__ == "__main__":
+    from time import sleep
+    logging.basicConfig(level=logging.INFO)
+    while True:
+        sleep(1)
+        print("Active window id: " + str(get_active_window_id()))