From 22a6d163c970a4d47244edc8bc0e202c93932ffc Mon Sep 17 00:00:00 2001 From: "Prompt. Click. Run." <59135406+promptclickrun@users.noreply.github.com> Date: Sat, 12 Sep 2026 00:57:14 -0500 Subject: [PATCH] feat(notifications): add scoped native host grants and approval attention --- dashboard/plugin_api.py | 2 + docs/NATIVE_NOTIFICATIONS.md | 237 ++++++ loopdy_plugin/activity_bridge.py | 46 +- loopdy_plugin/managed_notifications.py | 769 +++++++++++++++++++ loopdy_plugin/managed_notifications_api.py | 98 +++ loopdy_plugin/registration.py | 31 + loopdy_plugin/service.py | 12 +- plugin.yaml | 4 +- tests/test_activity_bridge.py | 5 +- tests/test_managed_approval_notifications.py | 207 +++++ tests/test_managed_notifications.py | 183 +++++ tests/test_native_notification_enrollment.py | 44 ++ tests/test_registration.py | 22 +- 13 files changed, 1640 insertions(+), 20 deletions(-) create mode 100644 docs/NATIVE_NOTIFICATIONS.md create mode 100644 loopdy_plugin/managed_notifications.py create mode 100644 loopdy_plugin/managed_notifications_api.py create mode 100644 tests/test_managed_approval_notifications.py create mode 100644 tests/test_managed_notifications.py create mode 100644 tests/test_native_notification_enrollment.py diff --git a/dashboard/plugin_api.py b/dashboard/plugin_api.py index bbd12a7..ba2a94c 100644 --- a/dashboard/plugin_api.py +++ b/dashboard/plugin_api.py @@ -55,6 +55,8 @@ router = APIRouter() router.include_router(workspace_files_router) +from loopdy_plugin.managed_notifications_api import router as managed_notifications_router +router.include_router(managed_notifications_router) _WORKSPACE_GIT_STATE_PATH = data_path().with_name("workspace-git.sqlite3") _service = None _workspace_git_service = None diff --git a/docs/NATIVE_NOTIFICATIONS.md b/docs/NATIVE_NOTIFICATIONS.md new file mode 100644 index 0000000..85b45a4 --- /dev/null +++ b/docs/NATIVE_NOTIFICATIONS.md @@ -0,0 +1,237 @@ +# Optional native-host notifications + +Managed notification enrollment is separate from native Hermes authentication and +from Loopdy Link chat pairing. Foreground native chat does not require this plugin, +an APNs enrollment, or access to Loopdy's chat relay. + +This source adds account-scoped sending authority, not an operator-wide relay key. +The host receives only a public grant for one account device, recipient encryption +key, profile, allowed event types and expiry. APNs keys and tenant-wide credentials +never go to the host. The existing independently configured relay APIs remain +available for legacy installations. + +## Activation and capability discovery + +Hermes mounts `dashboard/plugin_api.py` through the supported plugin dashboard +loader at `/api/plugins/loopdy`. Its stock authentication protects the new routes. +Installation, enablement, mounted API and registered lifecycle observers are +separate states. Installing files does not hot-mount this router or activate hooks. +Use the host owner's supported activation/restart process separately when needed. + +`GET /notifications/capabilities` returns version 1, `hostKeyId`, `hostPublicKey`, +`managedEnrollmentSupported`, `supportedEventTypes`, `richLiveActivitySupported`, +and a `producerCapabilities` object. Supported events allow enrollment before the +first turn; producer booleans separately report lazy in-process observer loading. +Supported event types are `session.completed`, `session.failed`, and +`approval.required`. `producerCapabilities.nativeApproval` is true only after both +named approval observers are supported by the installed SDK and registered in this +process. Older hosts can retain their two-event enrollment; no existing grant is +expanded when support grows. +The public key is +uncompressed P-256 X9.63, unpadded base64url. Its ID is base64url SHA256 of those +65 bytes. The independently generated private key lives under the current Hermes +home's `plugin-data/loopdy/managed-notifications/`, outside the replaceable plugin +installation. The directory and key must be owner-only; symlinks are rejected. +There is no claimed loaded revision inferred from files on disk. + +Producer booleans reflect callbacks registered in the responding process. A mounted +API alone may truthfully report false. A capability is not a delivered push receipt. +The managed private-store worker currently requires POSIX file locking. Unsupported +platforms fail this optional feature without removing the legacy API. + +## Enrollment + +1. The app signs in to its Loopdy account, registers its mobile device and completes + existing account APNs recipient enrollment. No Hermes chat host pairing is needed. +2. The app reads the authenticated host capability/public key. +3. The app uses its **signed mobile identity**, not a host bearer, to create a grant + at `https://link.loopdy.app/v1/notifications/host-grants`. It chooses a profile, + event types and an expiry bounded by the current APNs recipient lease. Use the + intersection of the app-supported set, this host's `supportedEventTypes`, and + the user's requested events. Request `approval.required` explicitly for new + approval-enabled grants, never by rewriting a saved completion-only grant. + Freeze the selected set and request bytes with the idempotency key across retries. +4. The app saves grant-scoped sender trust before enabling delivery. It then calls + host `POST /notifications/enroll` with + `{version:1,idempotencyKey:,grantId:}`. +5. The host proves possession of its own key to the fixed first-party account + endpoint, checks the confirmed public grant, and persists the receipt. +6. The app subscribes an exact native session with + `PUT /notifications/enrollments//sessions` and + `{version:1,profile,sessionId,enabled:true}`. The host uses the supported read-only + profile SessionDB, exact stored ID and `profile_name`. Missing/unknown ownership + is denied rather than guessed. There is no all-sessions target. + +Readback: `GET /notifications/enrollments/` checks the current cloud grant. +Local DELETE at the same path cancels local work but **does not revoke cloud +permission**. The app separately deletes the signed-mobile cloud grant with +`{version:1,expectedRevision}`. Device/account removal also revokes managed grants. +There is only one non-revoked host/profile/recipient grant at a time; list and reuse +its receipt, or revoke it explicitly before replacing it. Token/key/revision changes +invalidate the old grant rather than silently changing its destination. + +## Significant events and retry ownership + +Completion scope remains `session.completed` and `session.failed`, emitted from +real stock `on_session_end` facts only for explicitly subscribed sessions. Cancellation +is not failure; child completion/failure is not another parent alert. Existing Home +completion history and unenrolled legacy behavior remain intact. + +A stable event ID is `:<64 lowercase SHA256 hex characters>`. The digest +covers canonical JSON `[profile,sessionId,turnId,eventType]` for completion/failure. +Approval attention instead covers +`[profile,sessionId,turnId,toolCallId,"approval.required"]`. The grant prefix binds +the exact recipient/host enrollment. The encrypted LP1 +payload uses the existing relay cryptography and a deterministic delivery ID: +`ng-` plus UUIDv5(URL namespace, event ID). Random encryption inputs and the exact +serialized request are persisted once. Retries reuse those bytes, not fresh +ciphertext. Host proof uses a fresh nonce on each HTTP attempt. + +Alert copy is fixed (`Your agent finished` / `Your agent could not finish` / +`Your agent requested approval`), not +prompts, tool arguments, questions or error details. Existing explicit device +preferences are consulted when that recipient also exists in the legacy store. +Quiet-hours suppression is recorded durably. For subscribed recipients the managed +lane owns authorized completion/failure alerts, so the legacy sender suppresses +those same events for that recipient. It does not take ownership of legacy approvals. +The app must likewise treat foreground session updates as data, not a second banner. + +After successful LP1 authentication/decryption the app resolves the exact grant +prefix, validates that grant's recipient, expiry and allowed event type, and fetches +`GET /notifications/enrollments//events/` on its originating host. +That endpoint rechecks enrollment and exact current session ownership. It returns +only event ID/type, profile, session ID, turn ID and occurrence time. A tap never +executes an approval or command. Outer APNs routing fields are not authority. + +## Bounded native approval attention + +This additive observer uses supported `ctx.register_hook("pre_approval_request", callback)` +and `ctx.register_hook("post_approval_response", callback)`. The SDK callback payload +provides `surface`, `turn_id`, `tool_call_id`, and (when its observability context is +bound) `session_id`. Only exact `surface="gateway"`, non-coalesced native session +observations qualify. Smart assessments, CLI, unknown/custom transport surfaces, +child-session hooks, missing coordinates and contradictory profile metadata do not +queue attention. The stored session must match the exact subscribed ID and profile +through the supported read-only SessionDB. `session_key` and runtime UI IDs are never +used as substitutes; no recipient comes from command text or a home target. + +A grant must explicitly include `approval.required`. The pre-hook records one +**tool-scoped attention fact**, not one notification per native pending request. +It runs just before native presentation, not after a presentation receipt. The +native hook does not supply its request ID, so no request ID is fabricated and no +command, description, decision reason or pattern key is retained. Multiple prompts +inside one tool call coalesce under one event ID. The unchanged event-detail object +retains the actual canonical `turnId`, with no composite identity relabeled as a turn. + +A private `approval_attention` journal row and its frozen event request are written +atomically. A 3-second coalescing grace delays admission; expiry is at most 60 seconds +from observation (also bounded by grant expiry). These are notification limits, not +a claim about the native prompt's configured timeout. The existing worker has a +5-second idle wake interval; scheduling/network delays can cause +conservative drops. Preference suppression is a durable tombstone, not delayed replay. + +All gateway post-hook dispositions (including immediate response, `notify_failed`, +interruption and timeout), exact tool completion, and exact turn end retire **unsent** +attention. Cancellation follows the same retirement while retaining the existing +Stopped/child-cohort behavior. Response-before-pre and turn-end tombstones prevent +late observations from resurrecting attention. Local removal/unsubscription also +retires it. Tombstones are bounded to 4096 per grant and retained until grant cleanup; +capacity exhaustion refuses new attention rather than dropping deduplication evidence. +No work cohort is created or marked waiting from an approval hook in this version. + +The drain conservatively retires attention belonging to any previous producer +lifetime; it never restores a claim of still-pending approval from SQLite. Opening +the store from an API-only process does not itself retire a live producer's rows. +Only the process-owned worker drains. Shared-store multi-process observations that +do not belong to that worker are deliberately dropped, not replayed as live prompts. +Completion/failure retry and activity recovery behavior are unchanged. + +After claiming an intent and signing, the worker rechecks exact session ownership, +subscription, event authority, local retirement, producer lifetime and expiry +immediately before transport. Retries preserve the original ciphertext, event ID, +delivery ID and expiry. Post-hook retirement racing a failed send cannot requeue it. +A response after the final local fence can still race relay admission; there is no +notification-recall endpoint. **Accepted APNs pushes cannot be recalled.** The copy +therefore says only `Your agent requested approval`, never that approval is still +pending. On tap, authenticate/decrypt, revalidate the originating grant and fixed +host event detail, then reconcile the real native pending prompt. Approval responses +remain on native `approval.respond` with the actual native request ID. APNs provides +no authority for approval buttons, commands, or automatic responses. Existing voice +observers and the explicitly selected legacy approval transport are unchanged. + +## Rich Live Activities + +The native app registers the ActivityKit token directly with the signed-mobile +account route `/v1/notifications/host-grants//live-activities/`. +The host never receives that token. It subscribes only after cloud readback with: + +`PUT /notifications/enrollments//live-activities/` + +Body: `{version:1,profile,sessionId,sessionReference,leaseExpires,turnId?:string}`. +`sessionReference` is unpadded base64url SHA256 of `profile + NUL + sessionId`. +Supply the canonical turn ID from the authenticated subscribed-session work +readback, especially for queued/overlapping work. Native timeline IDs are not +canonical host turn IDs. +Without it, subscription binds the currently observed generating parent, or the +next genuine parent start. A late terminal or child stop cannot bind a new activity. +Once bound, an activity cannot switch turns. DELETE removes the local subscription; +the app owns cloud revocation when retiring a token. + +ContentState retains the exact rich-v1 keys: `phase`, `currentAction`, `progress`, +`completedSteps`, `activeSubagentCount`, `latestTool`, `timestamp`. It uses fixed +copy, null latestTool, zero unknown completedSteps, and legacy progress 0 while +active / 100 at terminal. Clients should not display this as a percentage. +The cloud envelope adds only existing update identity/reference/expiry fields. +There is no mandatory v2 or coarse `running` phase sent to the rich decoder. + +Child membership is deduplicated and tied to its original parent turn. A failed +child does not fail the parent. Parent completion retains outstanding children; +terminal is emitted only for the bound work cohort. Current rosters are in-process +facts, not restored from old durable starts. Restart can therefore produce a stale +activity, never fabricated running progress. + +Ordinary updates are latest-state coalesced to the existing 30-second relay budget. +Terminal updates bypass that delay and are durably retained until accepted or their +120-second rich-v1 expiry. Completion/failure alerts expire within 900 seconds; +approval attention expires within 60 seconds. An accepted/duplicate +receipt means relay admission, not visible delivery; expiry is not success. The +relay's queue rechecks grant and recipient validity on each attempt and just before +APNs. A token has one managed owner, with legacy registration/update fences. + +## Deliberately unadvertised producer boundaries + +Native clarification is supported by Hermes, but its background producer is not +integrated here (`nativeClarification=false`). Bounded approval attention is +separate from native pending-request transport and is advertised only when its +observers load. No request ID or question is fabricated. Existing legacy +approval/clarification behavior remains. +Scheduled-job/task subscriptions are not added to managed grants in this first +native-session scope; their existing legacy delivery is unchanged. + +Rich v1 retains its existing terminal transport phase for cancellation, with exact +fixed `Stopped` copy and no completion/failure alert. The canonical work readback +keeps `outcome:cancelled`; outstanding children delay the terminal until settled. +The native presentation distinguishes Stopped from Finished. There is no +push-to-start implementation, synthetic liveness heartbeat, or claim of physical +APNs verification. + +## Wire proof + +Headers: `x-loopdy-host-key-id`, `x-loopdy-timestamp`, `x-loopdy-nonce`, +`x-loopdy-signature`. P-256 ECDSA/SHA256 signature uses P1363 (64 bytes), unpadded +base64url. The signed UTF-8 transcript joins these fields with newline and no final +newline: + +1. `loopdy-notification-host-v1` +2. uppercase HTTP method +3. exact query-free path +4. grant UUID +5. decimal Unix timestamp +6. random 32-byte base64url nonce +7. lowercase SHA256 hex of the exact HTTP body bytes + +The clock window is 120 seconds, with durable nonce replay protection for the whole +accepted timestamp window. No account bearer, arbitrary relay URL, shared HMAC, or +APNs secret is accepted by the host enrollment API. All redirect responses are +rejected. New clients must negotiate these routes through capabilities rather than +assuming that equal package version strings imply loaded support. diff --git a/loopdy_plugin/activity_bridge.py b/loopdy_plugin/activity_bridge.py index 3645267..41066c8 100644 --- a/loopdy_plugin/activity_bridge.py +++ b/loopdy_plugin/activity_bridge.py @@ -847,9 +847,10 @@ async def complete( if state.get("ended"): return True timestamp = _next_live_timestamp(state, occurred_at) - state["ended"] = True + state["parent_outcome"] = "completed" if succeeded else "failed" + remaining = len(state["subagents"]) + phase = "delegating" if remaining else state["parent_outcome"] name = _safe_text(agent_name, 60) or "Your agent" - phase = "completed" if succeeded else "failed" action = ( f"{name} finished the response" if succeeded @@ -862,11 +863,12 @@ async def complete( current_action=action, progress=100, completed_steps=int(state.get("completed_steps") or 0), - active_subagent_count=0, - latest_tool=state.get("latest_tool"), + active_subagent_count=remaining, + latest_tool=None, timestamp=timestamp, ) await sender(update) + state["ended"] = remaining == 0 return True def _enqueue( @@ -900,6 +902,10 @@ async def _drain(self) -> None: live_update = self._project_live_activity(payload) if live_update is not None: await live_activity_sender(live_update) + if live_update["phase"] in {"completed", "failed"}: + state = self._live_state.get(str(payload.get("sessionId"))) + if state is not None: + state["ended"] = True except asyncio.CancelledError: raise except Exception as exc: @@ -950,12 +956,14 @@ def _project_live_activity(self, payload: dict[str, Any]) -> dict[str, Any] | No settled.add(event_id) state["completed_steps"] = min(999, int(state["completed_steps"]) + 1) subagents = state["subagents"] - if kind == "subagent": + if kind in {"subagent", "bot_handoff"}: subagent_id = _coordinate(payload.get("subagentId"), 180) or event_id - if lifecycle == "running": - subagents.add(subagent_id) - else: - subagents.discard(subagent_id) + roster_id = f"{kind}:{subagent_id}" + if lifecycle == "running" and roster_id not in state["settled_children"]: + subagents.add(roster_id) + elif lifecycle != "running" and roster_id in subagents: + subagents.discard(roster_id) + state["settled_children"].add(roster_id) if kind == "reasoning": if lifecycle == "running": @@ -963,7 +971,8 @@ def _project_live_activity(self, payload: dict[str, Any]) -> dict[str, Any] | No elif lifecycle == "succeeded": phase, action, progress = "responding", "Writing the response", 85 else: - phase, action, progress = "failed", "The response could not be completed", 100 + # A failed reasoning segment/child is not a parent terminal. + phase, action, progress = "thinking", "Reviewing the result", 0 elif kind == "tool": state["latest_tool"] = title[:64] if lifecycle == "running": @@ -983,6 +992,8 @@ def _project_live_activity(self, payload: dict[str, Any]) -> dict[str, Any] | No phase, action, progress = "delegating", title, 58 else: phase, action, progress = "thinking", "Continuing the conversation", 70 + if state.get("parent_outcome"): + phase = "delegating" if subagents else state["parent_outcome"] timestamp = _next_live_timestamp(state, occurred_at) return _live_activity_wire( session_id=session_id, @@ -1764,6 +1775,8 @@ def _new_live_state() -> dict[str, Any]: return { "completed_steps": 0, "subagents": set(), + "settled_children": set(), + "parent_outcome": None, "settled": set(), "latest_tool": None, "last_timestamp": 0, @@ -1811,11 +1824,16 @@ def _live_activity_wire( "updateId": update_id, "sessionReference": session_reference, "phase": phase, - "currentAction": _safe_text(current_action, 96) or "Working on your request", - "progress": min(max(int(progress), 0), 100), - "completedSteps": min(max(int(completed_steps), 0), 999), + "currentAction": { + "thinking": "Your agent is working", "waiting": "Your agent needs attention", + "using_tool": "Your agent is working", "delegating": "Agents are working", + "responding": "Your agent is responding", "completed": "Your agent finished", + "failed": "Your agent could not finish", + }.get(phase, "Your agent is working"), + "progress": 100 if phase in {"completed", "failed"} else 0, + "completedSteps": 0, "activeSubagentCount": min(max(int(active_subagent_count), 0), 99), - "latestTool": tool, + "latestTool": None, "timestamp": timestamp, "expires": timestamp + 120, } diff --git a/loopdy_plugin/managed_notifications.py b/loopdy_plugin/managed_notifications.py new file mode 100644 index 0000000..f9d2077 --- /dev/null +++ b/loopdy_plugin/managed_notifications.py @@ -0,0 +1,769 @@ +"""Notification-only native enrollment and durable, recipient-scoped delivery. + +This module owns no Hermes runtime internals. Stock registered observers provide +lifecycle facts; a plugin-owned worker drains frozen requests over HTTPS. Chat, +Link pairing, the legacy provider and LOOPDY_HOME_TARGET are not prerequisites. +""" +from __future__ import annotations + +try: + import fcntl +except ImportError: # Unsupported private-store locking must not break legacy APIs. + fcntl = None +import hashlib +import json +import logging +import os +import re +import sqlite3 +import stat +import threading +import time +import uuid +from collections import OrderedDict +from contextlib import contextmanager +from pathlib import Path +from typing import Any, Callable +from urllib.error import HTTPError +from urllib.request import HTTPRedirectHandler, Request, build_opener + +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import ec + +from .relay_crypto import ( + b64url_decode, b64url_encode, canonical_json_bytes, encrypt_alert, key_id, + public_key_bytes, public_key_from_x963, sign_p1363, +) +from .session_state import open_profile_store + +ORIGIN = "https://link.loopdy.app" +ROOT = "/v1/notifications/host-grants" +_UUID = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$") +_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") +_PROFILE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$") +_EVENT_TYPES = {"session.completed", "session.failed", "approval.required"} +_APPROVAL_EVENT = "approval.required" +_APPROVAL_GRACE_SECONDS = 3 +_APPROVAL_TTL_SECONDS = 60 +_APPROVAL_LIMIT = 4096 +_APPROVAL_HOOKS = ("pre_approval_request", "post_approval_response") +_ACTIONS = { + "thinking": "Your agent is working", "waiting": "Your agent needs attention", + "using_tool": "Your agent is working", "delegating": "Agents are working", + "responding": "Your agent is responding", "completed": "Your agent finished", + "failed": "Your agent could not finish", +} +logger = logging.getLogger("hermes.plugins.loopdy.notifications") + + +class ManagedNotificationError(ValueError): + def __init__(self, code: str, status: int = 409): + super().__init__(code) + self.code, self.status = code, status + + +def _identifier(value: Any, pattern: re.Pattern = _ID) -> str: + if not isinstance(value, str) or pattern.fullmatch(value) is None: + raise ManagedNotificationError("notification_coordinate_invalid", 400) + return value + + +def session_reference(profile: str, session_id: str) -> str: + return b64url_encode(hashlib.sha256(f"{profile}\0{session_id}".encode()).digest()) + + +def host_request_transcript(method: str, path: str, grant_id: str, timestamp: int, nonce: str, raw: bytes) -> bytes: + return "\n".join(("loopdy-notification-host-v1", method, path, grant_id, str(timestamp), nonce, hashlib.sha256(raw).hexdigest())).encode() + + +class _NoRedirect(HTTPRedirectHandler): + def redirect_request(self, req, fp, code, msg, headers, newurl): + raise ManagedNotificationError("notification_redirect_rejected", 502) + + +def _https_request(method: str, path: str, raw: bytes, headers: dict[str, str]) -> dict[str, Any]: + if not path.startswith(ROOT + "/") or "?" in path or "#" in path: + raise ManagedNotificationError("notification_path_invalid", 400) + request = Request(ORIGIN + path, data=raw if raw else None, headers=headers, method=method) + try: + with build_opener(_NoRedirect()).open(request, timeout=12) as response: + content = response.read(65537) + if len(content) > 65536: + raise ManagedNotificationError("notification_response_too_large", 502) + value = json.loads(content) + if not isinstance(value, dict) or value.get("version") != 1: + raise ManagedNotificationError("notification_response_invalid", 502) + return value + except HTTPError as error: + # Do not copy server bodies, tokens, grant contents or headers into logs. + raise ManagedNotificationError("notification_remote_rejected", error.code) from None + except (OSError, json.JSONDecodeError) as error: + raise ManagedNotificationError("notification_service_unavailable", 503) from error + + +def _private_directory(path: Path) -> None: + path.mkdir(parents=True, exist_ok=True, mode=0o700) + info = path.lstat() + if not stat.S_ISDIR(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077: + raise ManagedNotificationError("notification_private_storage_required", 503) + + +def _private_file(path: Path) -> None: + info = path.lstat() + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.getuid() or info.st_mode & 0o077: + raise ManagedNotificationError("notification_private_storage_required", 503) + + +class ManagedNotifications: + """One process-owned observer/worker; SQLite serializes other API processes.""" + def __init__(self, directory: Path, *, transport: Callable = _https_request, + clock: Callable = time.time, session_opener: Callable = open_profile_store): + if fcntl is None: + raise ManagedNotificationError("notification_platform_unavailable", 503) + self._fcntl = fcntl + _private_directory(directory) + self.directory, self.transport, self.clock = directory, transport, clock + self.session_opener = session_opener + self.preference_policy: Callable | None = None + self._lock = threading.RLock() + self._wake, self._stop = threading.Event(), threading.Event() + self._worker: threading.Thread | None = None + self._worker_lock = None + self._loaded_profiles: set[str] = set() + self._approval_profiles: set[str] = set() + # An observation belongs to this producer lifetime, never a recovered prompt. + self._approval_owner = str(uuid.uuid4()) + self._work: OrderedDict[tuple[str, str, str], dict[str, Any]] = OrderedDict() + self._child_owners: dict[tuple[str, str, str], str] = {} + self._key = self._identity() + self.public_key = b64url_encode(public_key_bytes(self._key.public_key())) + self.key_id = key_id(public_key_bytes(self._key.public_key())) + self.db_path = directory / "journal.sqlite3" + descriptor = os.open(self.db_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + os.close(descriptor) + _private_file(self.db_path) + with self._db() as db: + db.executescript(""" + CREATE TABLE IF NOT EXISTS grants(grant_id TEXT PRIMARY KEY, public_json TEXT NOT NULL, state TEXT NOT NULL, expires INTEGER NOT NULL); + CREATE TABLE IF NOT EXISTS subscriptions(grant_id TEXT NOT NULL REFERENCES grants(grant_id) ON DELETE CASCADE, profile TEXT NOT NULL, session_id TEXT NOT NULL, session_ref TEXT NOT NULL, PRIMARY KEY(grant_id,profile,session_id)); + CREATE TABLE IF NOT EXISTS events(event_id TEXT PRIMARY KEY, grant_id TEXT NOT NULL REFERENCES grants(grant_id) ON DELETE CASCADE, detail_json TEXT NOT NULL, occurred_at INTEGER NOT NULL); + CREATE TABLE IF NOT EXISTS pending(intent_id TEXT PRIMARY KEY, grant_id TEXT NOT NULL REFERENCES grants(grant_id) ON DELETE CASCADE, path TEXT NOT NULL, raw BLOB NOT NULL, expires INTEGER NOT NULL, state TEXT NOT NULL, attempts INTEGER NOT NULL DEFAULT 0, next_attempt INTEGER NOT NULL, session_ref TEXT NOT NULL, activity_id TEXT); + CREATE INDEX IF NOT EXISTS pending_due ON pending(state,next_attempt); + CREATE TABLE IF NOT EXISTS approval_attention(event_id TEXT PRIMARY KEY, grant_id TEXT NOT NULL REFERENCES grants(grant_id) ON DELETE CASCADE, profile TEXT NOT NULL, session_id TEXT NOT NULL, turn_id TEXT NOT NULL, tool_call_id TEXT NOT NULL, owner TEXT NOT NULL, state TEXT NOT NULL, expires INTEGER NOT NULL, reason TEXT NOT NULL); + CREATE INDEX IF NOT EXISTS approval_scope ON approval_attention(grant_id,profile,session_id,turn_id); + CREATE TABLE IF NOT EXISTS activities(activity_id TEXT PRIMARY KEY, grant_id TEXT NOT NULL REFERENCES grants(grant_id) ON DELETE CASCADE, profile TEXT NOT NULL, session_id TEXT NOT NULL, session_ref TEXT NOT NULL, lease_expires INTEGER NOT NULL, work_turn TEXT, state TEXT NOT NULL, last_timestamp INTEGER NOT NULL DEFAULT 0, last_signature TEXT, last_queued_at INTEGER NOT NULL DEFAULT 0); + """) + + @contextmanager + def _db(self): + connection = sqlite3.connect(self.db_path, timeout=5) + connection.row_factory = sqlite3.Row + connection.execute("PRAGMA foreign_keys=ON") + try: + with connection: + yield connection + finally: + connection.close() + + def _identity(self): + # A separate sidecar lock avoids locking SQLite's own inode. Never place + # private key material inside the replaceable plugin install directory. + lock_path = self.directory / "identity.lock" + fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + with os.fdopen(fd, "a+b") as lock: + _private_file(lock_path) + self._fcntl.flock(lock, self._fcntl.LOCK_EX) + path = self.directory / "host-key.pem" + if not path.exists(): + key = ec.generate_private_key(ec.SECP256R1()) + encoded = key.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()) + temp = self.directory / f".host-key-{uuid.uuid4()}.tmp" + try: + fd = os.open(temp, os.O_CREAT | os.O_EXCL | os.O_WRONLY | os.O_NOFOLLOW, 0o600) + with os.fdopen(fd, "wb") as output: + output.write(encoded); output.flush(); os.fsync(output.fileno()) + os.replace(temp, path) + finally: + temp.unlink(missing_ok=True) + _private_file(path) + if path.stat().st_size > 4096: + raise ManagedNotificationError("notification_identity_invalid", 503) + key = serialization.load_pem_private_key(path.read_bytes(), password=None) + if not isinstance(key, ec.EllipticCurvePrivateKey) or not isinstance(key.curve, ec.SECP256R1): + raise ManagedNotificationError("notification_identity_invalid", 503) + return key + + def capabilities(self) -> dict[str, Any]: + with self._lock: + loaded = bool(self._loaded_profiles) + approval_loaded = bool(self._approval_profiles) + return {"version": 1, "hostKeyId": self.key_id, "hostPublicKey": self.public_key, + "managedEnrollmentSupported": True, "supportedEventTypes": sorted(_EVENT_TYPES), + "richLiveActivitySupported": True, "producerCapabilities": { + "sessionCompletion": loaded, "sessionFailure": loaded, "richLiveActivity": loaded, + "nativeApproval": approval_loaded, "nativeClarification": False}} + + def _request(self, method: str, grant_id: str, suffix: str = "", raw: bytes = b"", + *, before_transport: Callable | None = None): + _identifier(grant_id, _UUID) + path = ROOT + "/" + grant_id + suffix + timestamp, nonce = int(self.clock()), b64url_encode(os.urandom(32)) + headers = {"Content-Type": "application/json", "Accept": "application/json", + "User-Agent": "Loopdy-Managed-Notifications/1", "x-loopdy-host-key-id": self.key_id, + "x-loopdy-timestamp": str(timestamp), "x-loopdy-nonce": nonce, + "x-loopdy-signature": b64url_encode(sign_p1363(self._key, host_request_transcript(method, path, grant_id, timestamp, nonce, raw)))} + if before_transport is not None and not before_transport(): + raise ManagedNotificationError("notification_attention_retired", 410) + return self.transport(method, path, raw, headers) + + def _validate_grant(self, value: Any, grant_id: str) -> dict[str, Any]: + if not isinstance(value, dict) or value.get("grantId") != grant_id or value.get("hostKeyId") != self.key_id or value.get("hostPublicKey") != self.public_key or value.get("state") != "active": + raise ManagedNotificationError("notification_grant_identity_mismatch", 403) + for field in ("revision", "recipientRevision", "authorizationEpoch", "createdAt", "expiresAt"): + if type(value.get(field)) is not int or value[field] < 1: + raise ManagedNotificationError("notification_grant_invalid", 502) + if not value["createdAt"] < value["expiresAt"] <= value["createdAt"] + 2592000 or value["expiresAt"] <= int(self.clock()): + raise ManagedNotificationError("notification_grant_expired", 403) + _identifier(value.get("profile"), _PROFILE) + _identifier(value.get("deviceId")); _identifier(value.get("tenantId")) + recipient = b64url_decode(value.get("recipientPublicKey"), expected_length=65) + public_key_from_x963(recipient) + event_types = value.get("eventTypes") + if (key_id(recipient) != value.get("recipientKeyId") or not isinstance(event_types, list) + or not event_types or any(not isinstance(event_type, str) for event_type in event_types) + or len(set(event_types)) != len(event_types) or not set(event_types) <= _EVENT_TYPES): + raise ManagedNotificationError("notification_grant_invalid", 502) + # Public metadata only; whitelist prevents accidental future secrets at rest. + fields = ("grantId", "hostKeyId", "hostPublicKey", "deviceId", "recipientPublicKey", "recipientKeyId", "recipientRevision", "authorizationEpoch", "profile", "eventTypes", "createdAt", "expiresAt", "revision", "tenantId", "state") + return {field: value[field] for field in fields} + + def enroll(self, grant_id: str, idempotency_key: str): + _identifier(grant_id, _UUID); _identifier(idempotency_key, _UUID) + value = self._request("POST", grant_id, "/claim", canonical_json_bytes({"version": 1, "idempotencyKey": idempotency_key})) + grant = self._validate_grant(value.get("grant"), grant_id) + with self._db() as db: + if db.execute("SELECT COUNT(*) FROM grants").fetchone()[0] >= 256 and not db.execute("SELECT 1 FROM grants WHERE grant_id=?", (grant_id,)).fetchone(): + raise ManagedNotificationError("notification_enrollment_limit") + previous = db.execute("SELECT * FROM grants WHERE grant_id=?", (grant_id,)).fetchone() + encoded = canonical_json_bytes(grant).decode() + if previous and (previous["public_json"] != encoded or previous["state"] != "active"): + raise ManagedNotificationError("notification_enrollment_conflict") + db.execute("INSERT OR IGNORE INTO grants VALUES(?,?,'active',?)", (grant_id, encoded, grant["expiresAt"])) + return {"version": 1, "grant": grant} + + def _grant(self, grant_id: str) -> dict[str, Any]: + _identifier(grant_id, _UUID) + with self._db() as db: + row = db.execute("SELECT * FROM grants WHERE grant_id=? AND state='active' AND expires>?", (grant_id, int(self.clock()))).fetchone() + if not row: + raise ManagedNotificationError("notification_enrollment_inactive", 404) + return json.loads(row["public_json"]) + + def enrollment(self, grant_id: str): + local = self._grant(grant_id) + try: + remote = self._validate_grant(self._request("GET", grant_id).get("grant"), grant_id) + except ManagedNotificationError as error: + if error.status in (403, 404): self.remove(grant_id) + raise + if remote != local: + self.remove(grant_id) + raise ManagedNotificationError("notification_enrollment_changed", 403) + return {"version": 1, "grant": remote} + + def remove(self, grant_id: str): + _identifier(grant_id, _UUID) + with self._db() as db: + db.execute("UPDATE grants SET state='removed' WHERE grant_id=?", (grant_id,)) + db.execute("UPDATE approval_attention SET state='retired',reason='removed' WHERE grant_id=?", (grant_id,)) + db.execute("DELETE FROM subscriptions WHERE grant_id=?", (grant_id,)) + db.execute("DELETE FROM pending WHERE grant_id=?", (grant_id,)) + db.execute("DELETE FROM activities WHERE grant_id=?", (grant_id,)) + return {"version": 1, "state": "removed", "grantId": grant_id} + + def _session(self, profile: str, session_id: str): + _identifier(profile, _PROFILE); _identifier(session_id) + def read(db): + row = db.get_session(session_id) + if not isinstance(row, dict) or row.get("id") != session_id: + raise ManagedNotificationError("notification_session_unknown", 404) + # Public store profile metadata, never current selected UI state. + owner = row.get("profile_name") + if owner != profile or row.get("deleted_at") or row.get("archived_at"): + raise ManagedNotificationError("notification_session_forbidden", 403) + return row + try: + return self.session_opener(profile, read, read_only=True) + except (LookupError, OSError, sqlite3.Error) as error: + raise ManagedNotificationError("notification_session_unavailable", 503) from error + + def subscribe(self, grant_id: str, profile: str, session_id: str, enabled: bool): + grant = self.enrollment(grant_id)["grant"] + if profile != grant["profile"] or type(enabled) is not bool: + raise ManagedNotificationError("notification_scope_forbidden", 403) + self._session(profile, session_id) + reference = session_reference(profile, session_id) + with self._db() as db: + if enabled: + count = db.execute("SELECT COUNT(*) FROM subscriptions WHERE grant_id=?", (grant_id,)).fetchone()[0] + if count >= 128 and not db.execute("SELECT 1 FROM subscriptions WHERE grant_id=? AND profile=? AND session_id=?", (grant_id, profile, session_id)).fetchone(): + raise ManagedNotificationError("notification_session_limit") + db.execute("INSERT OR IGNORE INTO subscriptions VALUES(?,?,?,?)", (grant_id, profile, session_id, reference)) + else: + db.execute("UPDATE approval_attention SET state='retired',reason='unsubscribed' WHERE grant_id=? AND profile=? AND session_id=?", (grant_id, profile, session_id)) + db.execute("DELETE FROM subscriptions WHERE grant_id=? AND profile=? AND session_id=?", (grant_id, profile, session_id)) + db.execute("DELETE FROM pending WHERE grant_id=? AND session_ref=?", (grant_id, reference)) + db.execute("DELETE FROM activities WHERE grant_id=? AND session_ref=?", (grant_id, reference)) + return {"version": 1, "grantId": grant_id, "profile": profile, "sessionId": session_id, "sessionReference": reference, "enabled": enabled} + + def work_snapshot(self, grant_id: str, profile: str, session_id: str): + """Read only this process's public-hook observations, never DB starts. + + A current session observation is not correlation with a mobile frame. + Retained older turns remain addressable by explicit activity registration. + """ + grant = self.enrollment(grant_id)["grant"] + if profile != grant["profile"]: + raise ManagedNotificationError("notification_scope_forbidden", 403) + self._session(profile, session_id) + with self._lock, self._db() as db: + self._require_subscription(db, grant_id, profile, session_id) + work = next((value for key, value in reversed(self._work.items()) + if key[:2] == (profile, session_id)), None) + snapshot = None + if work is not None: + phase, count, terminal = self._work_projection(work) + snapshot = {"profile": profile, "sessionId": session_id, + "turnId": work["turn"], "phase": phase, + "activeSubagentCount": count, "terminal": terminal, + "outcome": work["outcome"], "observedAt": work["observed_at"]} + return {"version": 1, "grantId": grant_id, "work": snapshot} + + def _require_subscription(self, db, grant_id: str, profile: str, session_id: str): + # Recheck local authority after cloud/SessionDB awaits, in the transaction + # used by the read/write. A concurrent removal cannot resurrect authority. + if not db.execute("SELECT 1 FROM subscriptions s JOIN grants g USING(grant_id) " + "WHERE s.grant_id=? AND s.profile=? AND s.session_id=? " + "AND g.state='active' AND g.expires>?", + (grant_id, profile, session_id, int(self.clock()))).fetchone(): + raise ManagedNotificationError("notification_session_not_subscribed") + + @staticmethod + def _work_projection(work): + count = sum(state == "active" for state in work["children"].values()) + outcome = work["outcome"] + phase = "delegating" if count else ("completed" if outcome == "cancelled" else (outcome if outcome in ("completed", "failed") else work["phase"])) + return phase, min(count, 99), outcome is not None and count == 0 + + def event(self, grant_id: str, event_id: str): + self.enrollment(grant_id) + if not re.fullmatch(re.escape(grant_id) + r":[0-9a-f]{64}", event_id): + raise ManagedNotificationError("notification_event_unknown", 404) + with self._db() as db: + row = db.execute("SELECT detail_json FROM events WHERE event_id=? AND grant_id=?", (event_id, grant_id)).fetchone() + if not row: + raise ManagedNotificationError("notification_event_unknown", 404) + event = json.loads(row["detail_json"]) + self._session(event["profile"], event["sessionId"]) + return {"version": 1, "event": event} + + def subscribe_activity(self, grant_id: str, activity_id: str, profile: str, session_id: str, reference: str, lease_expires: int, turn_id: str | None = None): + _identifier(activity_id) + if turn_id is not None: _identifier(turn_id) + grant = self.enrollment(grant_id)["grant"] + self._session(profile, session_id) + if profile != grant["profile"] or reference != session_reference(profile, session_id): + raise ManagedNotificationError("notification_scope_forbidden", 403) + receipt = self._request("GET", grant_id, "/live-activities/" + activity_id).get("activity") + if not isinstance(receipt, dict) or receipt.get("grantId") != grant_id or receipt.get("activityId") != activity_id or receipt.get("sessionReference") != reference or receipt.get("status") != "active" or receipt.get("leaseExpires") != lease_expires or type(lease_expires) is not int or lease_expires <= int(self.clock()): + raise ManagedNotificationError("notification_activity_unconfirmed") + with self._lock: + with self._db() as db: + self._require_subscription(db, grant_id, profile, session_id) + prior = db.execute("SELECT * FROM activities WHERE activity_id=?", (activity_id,)).fetchone() + if prior and (prior["grant_id"] != grant_id or prior["session_ref"] != reference + or prior["profile"] != profile or prior["session_id"] != session_id + or prior["state"] not in ("active", "terminal_pending", "terminal_accepted") + or (turn_id is not None and prior["work_turn"] not in (None, turn_id))): + raise ManagedNotificationError("notification_activity_conflict") + if not prior and db.execute("SELECT COUNT(*) FROM activities WHERE lease_expires>?", (int(self.clock()),)).fetchone()[0] >= 128: + raise ManagedNotificationError("notification_activity_limit") + # Explicit delayed registration must select that retained observed + # turn, never today's current turn. A retry preserves its owner. + turn = prior["work_turn"] if prior and prior["work_turn"] is not None else turn_id + if turn is not None: + work = self._work.get((profile, session_id, turn)) + if work is None: + raise ManagedNotificationError("notification_work_unobserved") + else: + work = next((value for key, value in reversed(self._work.items()) + if key[:2] == (profile, session_id) and not value["terminal"] + and value["outcome"] is None), None) + turn = work["turn"] if work else None + db.execute("INSERT INTO activities(activity_id,grant_id,profile,session_id,session_ref,lease_expires,work_turn,state) VALUES(?,?,?,?,?,?,?,'active') ON CONFLICT(activity_id) DO UPDATE SET lease_expires=excluded.lease_expires,work_turn=COALESCE(activities.work_turn,excluded.work_turn)", (activity_id, grant_id, profile, session_id, reference, lease_expires, turn)) + # Commit the owner first, then enqueue under the same observation lock. + # No later hook is needed (including a turn ending before token arrival). + # Terminal pending/accepted retries keep their original frozen request. + if work is not None: + phase, count, terminal = self._work_projection(work) + self._queue_activity(profile, session_id, work["turn"], phase, count, terminal, stopped=work["outcome"] == "cancelled" and terminal) + return {"version": 1, "activityId": activity_id, "grantId": grant_id, "sessionReference": reference, "state": "subscribed"} + + def remove_activity(self, grant_id: str, activity_id: str): + self._grant(grant_id); _identifier(activity_id) + with self._db() as db: + db.execute("DELETE FROM activities WHERE activity_id=? AND grant_id=?", (activity_id, grant_id)) + db.execute("DELETE FROM pending WHERE activity_id=? AND grant_id=?", (activity_id, grant_id)) + return {"version": 1, "activityId": activity_id, "grantId": grant_id, "state": "removed"} + + def owns_alert(self, event: Any, device_id: str) -> bool: + # Native observer attention never steals the legacy approval transport. + if getattr(event, "type", None) not in {"session.completed", "session.failed"}: return False + with self._db() as db: + rows = db.execute("SELECT g.public_json FROM grants g JOIN subscriptions s USING(grant_id) WHERE g.state='active' AND g.expires>? AND s.profile=? AND s.session_id=?", (int(self.clock()), event.profile, event.session_id)).fetchall() + return any((grant := json.loads(row["public_json"]))["deviceId"] == device_id + and event.type in grant["eventTypes"] for row in rows) + + @staticmethod + def _approval_event_id(grant_id: str, profile: str, session_id: str, turn_id: str, tool_call_id: str): + # Tool-scoped attention, NOT the identity of a native approval request. + digest = hashlib.sha256(canonical_json_bytes( + [profile, session_id, turn_id, tool_call_id, _APPROVAL_EVENT])).hexdigest() + return grant_id + ":" + digest + + def _retire_approval_scope(self, profile: str, session_id: str, turn_id: str, + tool_call_id: str, reason: str): + """Empty tool is an exact turn-end tombstone; never an invented turn.""" + now = int(self.clock()) + with self._db() as db: + db.execute("BEGIN IMMEDIATE") + rows = db.execute("SELECT g.* FROM grants g JOIN subscriptions s USING(grant_id) WHERE g.state='active' AND g.expires>? AND s.profile=? AND s.session_id=?", (now, profile, session_id)).fetchall() + for row in rows: + grant = json.loads(row["public_json"]) + if _APPROVAL_EVENT not in grant["eventTypes"]: continue + scope = (grant["grantId"], profile, session_id, turn_id) + where = "grant_id=? AND profile=? AND session_id=? AND turn_id=?" + if tool_call_id: + where += " AND tool_call_id=?" + scope += (tool_call_id,) + db.execute(f"UPDATE approval_attention SET state='retired',reason=? WHERE {where}", (reason, *scope)) + db.execute(f"UPDATE pending SET state='retired' WHERE state IN ('pending','sending') AND intent_id IN (SELECT event_id FROM approval_attention WHERE {where})", scope) + # Also fence response-before-pre and delayed pre after turn end. + event_id = self._approval_event_id(grant["grantId"], profile, session_id, turn_id, tool_call_id) + db.execute("INSERT OR IGNORE INTO approval_attention SELECT ?,?,?,?,?,?,?,'retired',?,? WHERE (SELECT COUNT(*) FROM approval_attention WHERE grant_id=?)? AND s.profile=? AND s.session_id=?", (now, profile, session_id)).fetchall() + for row in rows: + grant = json.loads(row["public_json"]) + if event_type not in grant["eventTypes"]: continue + digest = hashlib.sha256(canonical_json_bytes([profile, session_id, turn_id, event_type])).hexdigest() + event_id = self._approval_event_id(grant["grantId"], profile, session_id, turn_id, tool_call_id) if approval else grant["grantId"] + ":" + digest + if db.execute("SELECT 1 FROM events WHERE event_id=?", (event_id,)).fetchone(): continue + if approval: + turn_end = self._approval_event_id(grant["grantId"], profile, session_id, turn_id, "") + if db.execute("SELECT 1 FROM approval_attention WHERE event_id IN (?,?)", (event_id, turn_end)).fetchone(): continue + if db.execute("SELECT COUNT(*) FROM approval_attention WHERE grant_id=?", (grant["grantId"],)).fetchone()[0] >= _APPROVAL_LIMIT: continue + if db.execute("SELECT COUNT(*) FROM events WHERE grant_id=?", (grant["grantId"],)).fetchone()[0] >= 4096: continue + if db.execute("SELECT COUNT(*) FROM pending WHERE grant_id=? AND state IN ('pending','sending')", (grant["grantId"],)).fetchone()[0] >= 256: continue + from .events import LoopdyEvent + policy_event = LoopdyEvent(event_id=event_id, type=event_type, profile=profile, session_id=session_id) + policy = self.preference_policy(policy_event, grant["deviceId"]) if self.preference_policy else {} + detail = {"eventId": event_id, "eventType": event_type, "profile": profile, "sessionId": session_id, "turnId": turn_id, "occurredAt": now} + expires = min(now + (_APPROVAL_TTL_SECONDS if approval else 900), grant["expiresAt"]) + if approval: + db.execute("INSERT INTO approval_attention VALUES(?,?,?,?,?,?,?,?,?,?)", + (event_id, grant["grantId"], profile, session_id, turn_id, tool_call_id, + self._approval_owner, "retired" if policy.get("suppression") else "pending", + expires, "suppressed" if policy.get("suppression") else "observed")) + if policy.get("suppression"): + # Suppression is durable; a retry or quiet-hours boundary cannot replay it. + db.execute("INSERT INTO events VALUES(?,?,?,?)", (event_id, grant["grantId"], canonical_json_bytes(detail).decode(), now)) + continue + body = "Your agent requested approval" if approval else ("Your agent finished" if event_type == "session.completed" else "Your agent could not finish") + envelope = encrypt_alert(tenant_id=grant["tenantId"], device_id=grant["deviceId"], + delivery_id="ng-" + str(uuid.uuid5(uuid.NAMESPACE_URL, event_id)), event_id=event_id, + event_type=event_type, title="Loopdy", body=body, + recipient_public_key=b64url_decode(grant["recipientPublicKey"], expected_length=65), sender_private_key=self._key, + issued=now, expires=expires, ephemeral_private_key=ec.generate_private_key(ec.SECP256R1()), salt=os.urandom(32), nonce=os.urandom(12)) + reference = session_reference(profile, session_id) + raw = canonical_json_bytes({"version": 1, "eventId": event_id, "eventType": event_type, "sessionReference": reference, "envelope": envelope, "sound": policy.get("sound") is not False}) + db.execute("INSERT INTO events VALUES(?,?,?,?)", (event_id, grant["grantId"], canonical_json_bytes(detail).decode(), now)) + due = now + _APPROVAL_GRACE_SECONDS if approval else now + db.execute("INSERT INTO pending(intent_id,grant_id,path,raw,expires,state,next_attempt,session_ref) VALUES(?,?,?,?,?,'pending',?,?)", (event_id, grant["grantId"], "/events", raw, expires, due, reference)) + self._wake.set() + + def observe(self, hook: str, *, profile: str, **payload: Any): + """Synchronous stock hook: persist only; never perform network I/O here.""" + try: + self._observe(hook, profile=profile, **payload) + except (ValueError, OSError, sqlite3.Error): + logger.warning("Managed notification lifecycle observation unavailable") + + def _observe(self, hook: str, *, profile: str, **payload: Any): + if self._stop.is_set(): return + if hook in _APPROVAL_HOOKS: + self._observe_approval(hook, profile=profile, **payload) + return + if hook not in ("pre_llm_call", "post_llm_call", "pre_tool_call", "post_tool_call", + "on_session_end", "subagent_start", "subagent_stop"): + return + profile = _identifier(payload.get("profile_name") or profile, _PROFILE) + child_hook = hook in ("subagent_start", "subagent_stop") + session_id = payload.get("parent_session_id") if child_hook else payload.get("session_id") + if not isinstance(session_id, str) or not _ID.fullmatch(session_id): return + if not child_hook and (payload.get("parent_session_id") or payload.get("platform") == "subagent"): return + with self._db() as db: + if not db.execute("SELECT 1 FROM subscriptions s JOIN grants g USING(grant_id) WHERE s.profile=? AND s.session_id=? AND g.state='active' AND g.expires>?", (profile, session_id, int(self.clock()))).fetchone(): return + turn = payload.get("turn_id") + if not child_hook and isinstance(turn, str) and _ID.fullmatch(turn): + if hook == "on_session_end": + self._retire_approval_scope(profile, session_id, turn, "", "turn_end") + elif hook == "post_tool_call": + tool = payload.get("tool_call_id") + if isinstance(tool, str) and _ID.fullmatch(tool): + self._retire_approval_scope(profile, session_id, turn, tool, "tool_end") + if (hook == "on_session_end" and isinstance(turn, str) and _ID.fullmatch(turn) + and payload.get("interrupted") is not True): + if payload.get("failed") is True: + self._queue_event(profile, session_id, turn, "session.failed") + elif payload.get("completed") is True: + self._queue_event(profile, session_id, turn, "session.completed") + with self._lock: + if child_hook: + child = payload.get("child_session_id") or payload.get("child_subagent_id") + turn = self._child_owners.get((profile, session_id, child)) if isinstance(child, str) else None + if turn is None and hook == "subagent_start": turn = payload.get("parent_turn_id") + if not isinstance(turn, str) or not _ID.fullmatch(turn): return + coordinate = (profile, session_id, turn) + work = self._work.get(coordinate) + if hook == "pre_llm_call" and isinstance(turn, str) and _ID.fullmatch(turn): + if work is None or work["turn"] != turn: + if len(self._work) >= 128: + # Never discard a live cohort or an activity's retained + # canonical owner merely because newer work appeared. + with self._db() as db: + bound = {(row["profile"], row["session_id"], row["work_turn"]) + for row in db.execute("SELECT profile,session_id,work_turn FROM activities WHERE lease_expires>?", (int(self.clock()),))} + evicted = next((key for key, value in self._work.items() + if value["terminal"] and key not in bound), None) + if evicted is None: return + self._work.pop(evicted) + for child_key in tuple(self._child_owners): + if child_key[:2] == evicted[:2] and self._child_owners[child_key] == evicted[2]: self._child_owners.pop(child_key, None) + work = {"turn": turn, "phase": "thinking", "outcome": None, "children": {}, "terminal": False, + "observed_at": int(self.clock())} + self._work[coordinate] = work + elif work["terminal"]: return + if work is None or work["terminal"]: return + if child_hook: + child = payload.get("child_session_id") or payload.get("child_subagent_id") + parent_turn = payload.get("parent_turn_id") + if not isinstance(child, str) or not _ID.fullmatch(child): return + owner_key = (profile, session_id, child) + if hook == "subagent_start": + if not isinstance(parent_turn, str) or parent_turn != work["turn"]: return + if len(work["children"]) >= 128 and child not in work["children"]: return + self._child_owners.setdefault(owner_key, parent_turn) + if self._child_owners[owner_key] != work["turn"]: return + # A repeated start after stop cannot resurrect this child. + work["children"].setdefault(child, "active") + else: + if self._child_owners.get(owner_key) != work["turn"] or work["children"].get(child) != "active": return + work["children"][child] = "ended" + elif turn != work["turn"] or work["terminal"] or work["outcome"] is not None: return + elif hook == "on_session_end": + if payload.get("interrupted") is True: + work["outcome"] = "cancelled" + elif payload.get("failed") is True: + work["outcome"] = "failed" + # Significant alert already persisted independently of activity state. + elif payload.get("completed") is True: + work["outcome"] = "completed" + # Significant alert already persisted independently of activity state. + else: return + elif hook == "post_llm_call": + work["phase"] = "responding" + elif hook in ("pre_tool_call", "post_tool_call"): + work["phase"] = "using_tool" + work["observed_at"] = int(self.clock()) + phase, active_count, terminal = self._work_projection(work) + work["terminal"] = terminal + # Rich-v1 uses completed as the terminal transport state. Stopped + # remains explicit fixed copy; cancellation never queues an alert. + self._queue_activity(profile, session_id, work["turn"], phase, active_count, terminal, + allow_bind=hook == "pre_llm_call", stopped=work["outcome"] == "cancelled" and terminal) + + def _queue_activity(self, profile: str, session_id: str, turn: str, phase: str, count: int, terminal: bool, *, allow_bind: bool = False, stopped: bool = False): + now = int(self.clock()) + with self._db() as db: + rows = db.execute("SELECT a.*,g.expires AS grant_expires FROM activities a JOIN grants g USING(grant_id) WHERE a.profile=? AND a.session_id=? AND a.state='active' AND a.lease_expires>? AND g.state='active' AND g.expires>?", (profile, session_id, now, now)).fetchall() + for row in rows: + if row["work_turn"] not in (None, turn) or (row["work_turn"] is None and not allow_bind): continue + action = "Stopped" if stopped and terminal and phase == "completed" else _ACTIONS[phase] + signature = json.dumps([turn, phase, count, action]) + if row["last_signature"] == signature: continue + # Queue latest significant state, with relay's existing 30s budget. + pending = db.execute("SELECT raw,next_attempt FROM pending WHERE activity_id=? AND state='pending' ORDER BY next_attempt LIMIT 1", (row["activity_id"],)).fetchone() + timestamp = max(now, json.loads(bytes(pending["raw"]))["timestamp"]) if pending else max(now, row["last_timestamp"] + 1) + expires = min(timestamp + 120, row["grant_expires"], row["lease_expires"]) + if expires <= timestamp: continue + update_id = b64url_encode(hashlib.sha256(canonical_json_bytes([row["grant_id"], row["activity_id"], turn, signature, timestamp])).digest()) + update = {"version": 1, "updateId": update_id, "sessionReference": row["session_ref"], "phase": phase, + "currentAction": action, "progress": 100 if terminal else 0, "completedSteps": 0, + "activeSubagentCount": count, "latestTool": None, "timestamp": timestamp, "expires": expires} + due = now if terminal or phase == "waiting" else max(now, pending["next_attempt"] if pending else row["last_queued_at"] + 30) + # Only nonterminal updates may be superseded; terminal is durable. + db.execute("DELETE FROM pending WHERE activity_id=? AND state='pending'", (row["activity_id"],)) + db.execute("INSERT INTO pending(intent_id,grant_id,path,raw,expires,state,next_attempt,session_ref,activity_id) VALUES(?,?,?,?,?,'pending',?,?,?)", (update_id, row["grant_id"], "/live-activities/" + row["activity_id"] + "/updates", canonical_json_bytes(update), expires, due, row["session_ref"], row["activity_id"])) + db.execute("UPDATE activities SET work_turn=?,state=?,last_timestamp=?,last_signature=?,last_queued_at=? WHERE activity_id=?", (turn, "terminal_pending" if terminal else "active", timestamp, signature, due, row["activity_id"])) + self._wake.set() + + def producer_loaded(self, profile: str, *, start_worker: bool = True, approval_hooks_loaded: bool = False): + with self._lock: + self._loaded_profiles.add(profile) + if approval_hooks_loaded: + self._approval_profiles.add(profile) + if start_worker and self._worker is None: + lock_path = self.directory / "worker.lock" + fd = os.open(lock_path, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600) + lock = os.fdopen(fd, "a+b") + try: self._fcntl.flock(lock, self._fcntl.LOCK_EX | self._fcntl.LOCK_NB) + except BlockingIOError: + lock.close(); return + self._worker_lock = lock + self._worker = threading.Thread(target=self._run, name="loopdy-managed-notifications", daemon=True) + self._worker.start() + + def close(self): + self._stop.set(); self._wake.set() + if self._worker is not None: self._worker.join(timeout=15) + with self._lock: + self._loaded_profiles.clear() + self._approval_profiles.clear() + if self._worker_lock is not None and (self._worker is None or not self._worker.is_alive()): + self._worker_lock.close(); self._worker_lock = None + + def _run(self): + while not self._stop.is_set(): + try: self.drain_pending() + except (ValueError, OSError, sqlite3.Error): logger.warning("Managed notification journal unavailable") + self._wake.wait(5); self._wake.clear() + + def _approval_transport_ready(self, row) -> bool: + """Last local fence after signing, immediately before the HTTPS call. + + Do not hold a lock across network I/O: observers must not delay the + native decision. A decision after this fence can still race admission; + an accepted push is not recallable and only states a past-tense fact. + """ + if self._stop.is_set(): return False + with self._db() as db: + attention = db.execute("SELECT * FROM approval_attention WHERE event_id=? AND grant_id=?", + (row["intent_id"], row["grant_id"])).fetchone() + if not attention or attention["owner"] != self._approval_owner or attention["state"] != "pending": return False + try: + self._session(attention["profile"], attention["session_id"]) + except (ValueError, OSError, sqlite3.Error): + return False + now = int(self.clock()) + with self._db() as db: + current = db.execute("SELECT g.public_json,p.raw,p.session_ref FROM pending p " + "JOIN approval_attention a ON a.event_id=p.intent_id AND a.grant_id=p.grant_id " + "JOIN grants g ON g.grant_id=a.grant_id " + "JOIN subscriptions s ON s.grant_id=a.grant_id AND s.profile=a.profile AND s.session_id=a.session_id " + "WHERE p.intent_id=? AND p.state='sending' AND p.expires>? AND a.state='pending' AND a.owner=? AND a.expires>? " + "AND g.state='active' AND g.expires>?", + (row["intent_id"], now, self._approval_owner, now, now)).fetchone() + if not current or self._stop.is_set(): return False + grant = json.loads(current["public_json"]) + return (grant["profile"] == attention["profile"] and _APPROVAL_EVENT in grant["eventTypes"] + and bytes(current["raw"]) == bytes(row["raw"]) + and current["session_ref"] == session_reference(attention["profile"], attention["session_id"])) + + def drain_pending(self): + """Bounded durable retry, exposed for main-owned no-send composition tests.""" + now = int(self.clock()) + with self._db() as db: + # Never replay a previous producer's assertion of human waiting. + # API-only construction is read-only here: retirement happens only in + # the process-owned drain (not when another API reader opens the DB). + db.execute("UPDATE approval_attention SET state='retired',reason=CASE WHEN owner!=? THEN 'recovery' ELSE 'timeout' END WHERE state='pending' AND (owner!=? OR expires<=?)", (self._approval_owner, self._approval_owner, now)) + db.execute("UPDATE pending SET state='retired' WHERE state IN ('pending','sending') AND intent_id IN (SELECT event_id FROM approval_attention WHERE state='retired')") + db.execute("UPDATE pending SET state='pending' WHERE state='sending' AND next_attempt<=?", (now,)) + db.execute("UPDATE pending SET state='expired' WHERE state='pending' AND expires<=?", (now,)) + db.execute("DELETE FROM pending WHERE expires? AND g.state='active' AND g.expires>? ORDER BY p.next_attempt,p.intent_id LIMIT 32", (now, now, now)).fetchall() + for row in rows: + if self._stop.is_set(): return + now = int(self.clock()) + with self._db() as db: + claimed = db.execute("UPDATE pending SET state='sending',next_attempt=? WHERE intent_id=? AND state='pending' AND expires>? AND EXISTS(SELECT 1 FROM grants WHERE grants.grant_id=pending.grant_id AND grants.state='active' AND grants.expires>?)", (now + 30, row["intent_id"], now, now)).rowcount + if claimed != 1: continue + try: + approval = row["path"] == "/events" and json.loads(bytes(row["raw"])).get("eventType") == _APPROVAL_EVENT + result = self._request("POST", row["grant_id"], row["path"], bytes(row["raw"]), + before_transport=(lambda: self._approval_transport_ready(row)) if approval else None) + if result.get("status") not in ("accepted", "duplicate") or not isinstance(result.get("deliveryId"), str): + raise ManagedNotificationError("notification_delivery_unconfirmed", 503) + except ManagedNotificationError as error: + if error.status in (403, 404): + self.remove(row["grant_id"]) + else: + with self._db() as db: + state = "failed" if error.status in (400, 401, 410, 422) else "pending" + # A post-hook/removal racing an in-flight failed request + # must not resurrect the retired intent. + db.execute("UPDATE pending SET state=?,attempts=attempts+1,next_attempt=? WHERE intent_id=? AND state='sending'", (state, now + min(60, 2 ** min(row["attempts"] + 1, 6)), row["intent_id"])) + if state == "failed": + db.execute("UPDATE approval_attention SET state='retired',reason='send_failed' WHERE event_id=? AND state='pending'", (row["intent_id"],)) + continue + with self._db() as db: + db.execute("UPDATE pending SET state='accepted' WHERE intent_id=? AND state='sending'", (row["intent_id"],)) + if row["activity_id"] and json.loads(bytes(row["raw"])).get("phase") in ("completed", "failed"): + db.execute("UPDATE activities SET state='terminal_accepted' WHERE activity_id=? AND state='terminal_pending'", (row["activity_id"],)) + + +_instances: dict[str, ManagedNotifications] = {} +_instances_lock = threading.Lock() + + +def get_managed_notifications() -> ManagedNotifications: + from hermes_constants import get_hermes_home + directory = get_hermes_home() / "plugin-data" / "loopdy" / "managed-notifications" + with _instances_lock: + key = str(directory) + if key not in _instances: + _instances[key] = ManagedNotifications(directory) + return _instances[key] diff --git a/loopdy_plugin/managed_notifications_api.py b/loopdy_plugin/managed_notifications_api.py new file mode 100644 index 0000000..e215a17 --- /dev/null +++ b/loopdy_plugin/managed_notifications_api.py @@ -0,0 +1,98 @@ +"""Mounted only by Hermes' supported authenticated plugin router loader.""" +from __future__ import annotations + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, ConfigDict, Field, StrictBool, StrictInt + +from .managed_notifications import ManagedNotificationError, get_managed_notifications + +router = APIRouter(prefix="/notifications") +_UUID = r"^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$" +_ID = r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$" +_PROFILE = r"^[A-Za-z0-9][A-Za-z0-9_-]{0,63}$" + + +class VersionedBody(BaseModel): + model_config = ConfigDict(extra="forbid") + version: StrictInt = Field(ge=1, le=1) + + +class EnrollmentBody(VersionedBody): + idempotencyKey: str = Field(pattern=_UUID) + grantId: str = Field(pattern=_UUID) + + +class SessionBody(VersionedBody): + profile: str = Field(pattern=_PROFILE) + sessionId: str = Field(pattern=_ID) + enabled: StrictBool + + +class WorkBody(VersionedBody): + profile: str = Field(pattern=_PROFILE) + sessionId: str = Field(pattern=_ID) + + +class ActivityBody(VersionedBody): + profile: str = Field(pattern=_PROFILE) + sessionId: str = Field(pattern=_ID) + sessionReference: str = Field(pattern=r"^[A-Za-z0-9_-]{43}$") + leaseExpires: StrictInt = Field(gt=0, le=9_999_999_999) + turnId: str | None = Field(default=None, pattern=_ID) + + +def _call(operation, *args): + try: + return operation(*args) + except ManagedNotificationError as error: + raise HTTPException(status_code=error.status, detail={"code": error.code}) from None + except (OSError, ValueError): + raise HTTPException(status_code=503, detail={"code": "notification_service_unavailable"}) from None + + +# Synchronous handlers run in FastAPI's threadpool: no blocking HTTPS on the +# Hermes event loop. Authentication is the stock host middleware, not an account +# bearer forwarded from the app. No secret authority is accepted in these bodies. +@router.get("/capabilities") +def capabilities(): + return _call(lambda: get_managed_notifications().capabilities()) + + +@router.post("/enroll") +def enroll(body: EnrollmentBody): + return _call(lambda: get_managed_notifications().enroll(body.grantId, body.idempotencyKey)) + + +@router.get("/enrollments/{grant_id}") +def enrollment(grant_id: str): + return _call(lambda: get_managed_notifications().enrollment(grant_id)) + + +@router.delete("/enrollments/{grant_id}") +def remove(grant_id: str): + return _call(lambda: get_managed_notifications().remove(grant_id)) + + +@router.put("/enrollments/{grant_id}/sessions") +def subscribe(grant_id: str, body: SessionBody): + return _call(lambda: get_managed_notifications().subscribe(grant_id, body.profile, body.sessionId, body.enabled)) + + +@router.post("/enrollments/{grant_id}/work") +def work_snapshot(grant_id: str, body: WorkBody): + return _call(lambda: get_managed_notifications().work_snapshot(grant_id, body.profile, body.sessionId)) + + +@router.get("/enrollments/{grant_id}/events/{event_id}") +def event(grant_id: str, event_id: str): + return _call(lambda: get_managed_notifications().event(grant_id, event_id)) + + +@router.put("/enrollments/{grant_id}/live-activities/{activity_id}") +def activity(grant_id: str, activity_id: str, body: ActivityBody): + return _call(lambda: get_managed_notifications().subscribe_activity(grant_id, activity_id, body.profile, body.sessionId, body.sessionReference, body.leaseExpires, body.turnId)) + + +@router.delete("/enrollments/{grant_id}/live-activities/{activity_id}") +def remove_activity(grant_id: str, activity_id: str): + return _call(lambda: get_managed_notifications().remove_activity(grant_id, activity_id)) diff --git a/loopdy_plugin/registration.py b/loopdy_plugin/registration.py index 8c3f79a..22878e1 100644 --- a/loopdy_plugin/registration.py +++ b/loopdy_plugin/registration.py @@ -7,6 +7,7 @@ import logging import os import subprocess +import sqlite3 import sys import time import weakref @@ -73,6 +74,7 @@ "post_llm_call", "post_tool_call", "pre_approval_request", + "post_approval_response", *DIRECT_OBSERVER_HOOKS, *NOTIFICATION_HOOKS, ) @@ -304,6 +306,35 @@ def present_approval(request): plugin_context=ctx, ), ) + # Independent, public native observers. The optional Link adapter is not + # constructed to enable these notifications, and no core hook is patched. + try: + from .managed_notifications import get_managed_notifications + managed = get_managed_notifications() + policy = getattr(active_service, "managed_notification_policy", None) + if callable(policy): managed.preference_policy = policy + active_service.managed_alert_owner = managed.owns_alert + def observe_managed(hook, **payload): + managed.observe(hook, profile=str(ctx.profile_name or profile), **payload) + for hook in ("pre_llm_call", "post_llm_call", "pre_tool_call", "post_tool_call", + "on_session_end", "subagent_start", "subagent_stop"): + ctx.register_hook(hook, partial(observe_managed, hook)) + # Registration is additive: retain the existing voice observer and the + # explicitly selected legacy transport. Older SDKs can warn-and-register + # unknown names, so registration alone is not proof of an emitter. + try: + from hermes_cli.plugins import VALID_HOOKS + approval_hooks_supported = {"pre_approval_request", "post_approval_response"} <= VALID_HOOKS + except ImportError: + approval_hooks_supported = False + if approval_hooks_supported: + for hook in ("pre_approval_request", "post_approval_response"): + ctx.register_hook(hook, partial(observe_managed, hook)) + managed.producer_loaded(profile, approval_hooks_loaded=approval_hooks_supported) + ctx.on_unload(managed.close) + except (OSError, ValueError, sqlite3.Error): + # Notification storage/identity failure must never break foreground chat. + logger.warning("Managed notification producer unavailable") ctx.on_unload(partial(release_service, active_service)) diff --git a/loopdy_plugin/service.py b/loopdy_plugin/service.py index 4f74c07..266fc17 100644 --- a/loopdy_plugin/service.py +++ b/loopdy_plugin/service.py @@ -86,6 +86,7 @@ def __init__( queue_size: int = 256, ): self.store = store + self.managed_alert_owner: Callable[[LoopdyEvent, str], bool] | None = None self._providers = dict(providers or {}) self._sleep = sleep_fn self._jitter = jitter_fn @@ -1180,6 +1181,13 @@ def test_notification(self, target: str = "all", *, profile: str = "default") -> target=target, ) + def managed_notification_policy(self, event: LoopdyEvent, device_id: str) -> dict[str, Any]: + """Reuse existing explicit device preferences without provisioning a legacy sender.""" + device = self.store.get_device(device_id) + preferences = (device or {}).get("preferences") or {} + return {"suppression": _suppression_reason(event, preferences, self._now()), + "sound": preferences.get("priority_sound") is not False} + def enqueue(self, event: LoopdyEvent, *, target: str) -> bool: if self._closed or self._closing: return False @@ -1256,7 +1264,9 @@ def deliver(self, event: LoopdyEvent, *, target: str) -> dict[str, Any]: previous = claimed claim_token = str(claimed["claim_token"]) preferences = device.get("preferences") or {} - suppression = _suppression_reason(event, preferences, self._now()) + owner = getattr(self, "managed_alert_owner", None) + managed_owned = callable(owner) and owner(event, str(device["device_id"])) + suppression = "managed_notification_owner" if managed_owned else _suppression_reason(event, preferences, self._now()) if suppression: self.store.record_device_delivery( event_id=event.event_id, diff --git a/plugin.yaml b/plugin.yaml index c93d816..ebaf2b8 100644 --- a/plugin.yaml +++ b/plugin.yaml @@ -8,7 +8,9 @@ description: > Native Hermes platform for encrypted Loopdy Link chat and workspace controls, proactive notifications, approvals, verified device identity, and native Generative UI renderers. Delivers proactive notifications through Expo, Apple, or an encrypted relay while keeping - authoritative event details on the authenticated Hermes host. + authoritative event details on the authenticated Hermes host. Supports optional + account-scoped native notification grants independently of Link chat pairing; + mounted notification capabilities report actual in-process producer readiness. author: Loopdy contributors config_schema: direct: diff --git a/tests/test_activity_bridge.py b/tests/test_activity_bridge.py index b9ec8c6..af80991 100644 --- a/tests/test_activity_bridge.py +++ b/tests/test_activity_bridge.py @@ -243,8 +243,9 @@ async def live_sender(payload): self.assertEqual(encrypted[0]["result"], "private forecast output") self.assertEqual(live[0]["type"], "live_activity.update") self.assertEqual(live[0]["phase"], "using_tool") - self.assertEqual(live[0]["currentAction"], "Checking weather") - self.assertEqual(live[0]["latestTool"], "Checking weather") + self.assertEqual(live[0]["currentAction"], "Your agent is working") + self.assertIsNone(live[0]["latestTool"]) + self.assertEqual(live[0]["progress"], 0) self.assertEqual(len(live[0]["sessionReference"]), 43) self.assertNotIn("private location", repr(live[0])) self.assertNotIn("API token", repr(live[0])) diff --git a/tests/test_managed_approval_notifications.py b/tests/test_managed_approval_notifications.py new file mode 100644 index 0000000..8aab4da --- /dev/null +++ b/tests/test_managed_approval_notifications.py @@ -0,0 +1,207 @@ +from __future__ import annotations + +import json +import sqlite3 +import unittest +import uuid + +from typing import Any +from loopdy_plugin.managed_notifications import ManagedNotifications +from tests import test_managed_notifications as fixtures + + +class ManagedApprovalNotificationTests(unittest.TestCase): + service: ManagedNotifications + grant: dict[str, Any] + calls: list[tuple[str, str, bytes, dict[str, str]]] + now: int + setUp = fixtures.ManagedNotificationTests.setUp + tearDown = fixtures.ManagedNotificationTests.tearDown + get_session = fixtures.ManagedNotificationTests.get_session + transport = fixtures.ManagedNotificationTests.transport + + def enroll_approval(self): + self.grant_id = str(uuid.uuid4()) + self.grant = dict(self.grant, grantId=self.grant_id, + eventTypes=["session.completed", "session.failed", "approval.required"]) + self.service.enroll(self.grant_id, str(uuid.uuid4())) + self.service.subscribe(self.grant_id, "default", "native-session", True) + self.calls.clear() + + def approval(self, hook: str = "pre_approval_request", **changes: Any): + payload = dict(profile="default", session_id="native-session", session_key="runtime-session", + turn_id="turn-a", tool_call_id="tool-a", surface="gateway", + command="PRIVATE command arguments", description="PRIVATE description") + payload.update(changes) + self.service.observe(hook, **payload) + + def pending(self): + with sqlite3.connect(self.service.db_path) as db: + return [json.loads(row[0]) for row in db.execute( + "SELECT raw FROM pending WHERE state='pending' AND grant_id=? AND path='/events'", + (self.grant_id,))] + + def test_human_approval_queues_one_exact_grant_owned_generic_alert(self): + self.enroll_approval() + self.service.observe("pre_llm_call", profile="default", session_id="native-session", + turn_id="turn-a", platform="desktop") + self.approval() + self.approval() + rows = self.pending() + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["eventType"], "approval.required") + self.assertTrue(rows[0]["eventId"].startswith(self.grant_id + ":")) + detail = self.service.event(self.grant_id, rows[0]["eventId"])["event"] + self.assertEqual(detail["turnId"], "turn-a") + self.assertEqual(detail["sessionId"], "native-session") + self.assertNotIn("PRIVATE", json.dumps(detail)) + self.assertNotIn("requestId", detail, "The observer did not supply a native request ID") + + def test_old_completion_only_grant_does_not_authorize_approval(self): + self.approval() + self.assertEqual(self.pending(), []) + + def test_smart_coalesced_or_unowned_approval_never_queues(self): + self.enroll_approval() + for change in ({"surface": "smart"}, {"coalesced": True}, {"session_id": "other"}, + {"profile": "other"}, {"session_id": ""}, {"tool_call_id": ""}, + {"turn_id": ""}, {"surface": "cli"}): + with self.subTest(change=change): + self.approval(**dict[str, Any](change)) + self.assertEqual(self.pending(), []) + + def test_response_retires_unsent_tool_scoped_attention(self): + self.enroll_approval() + self.approval() + self.assertEqual(len(self.pending()), 1) + self.approval("post_approval_response", choice="notify_failed") + self.assertEqual(self.pending(), []) + self.now += 10 + self.service.drain_pending() + self.assertFalse(any(path.endswith("/events") for _, path, _, _ in self.calls)) + + def test_revocation_refuses_queued_approval(self): + self.enroll_approval() + self.approval() + self.service.remove(self.grant_id) + self.now += 10 + self.service.drain_pending() + self.assertFalse(any(path.endswith("/events") for _, path, _, _ in self.calls)) + + def event_calls(self): + return [call for call in self.calls if call[1].endswith("/events")] + + def test_grace_then_frozen_retry_preserves_exact_ciphertext(self): + self.enroll_approval() + self.approval() + self.service.drain_pending() + self.assertEqual(self.event_calls(), []) + self.now += 3 + self.fail_send = True + self.service.drain_pending() + self.assertEqual(len(self.event_calls()), 1) + first = self.event_calls()[0] + self.now += 3 + self.fail_send = False + self.service.drain_pending() + second = self.event_calls()[1] + self.assertEqual(first[2], second[2]) + self.assertNotEqual(first[3]["x-loopdy-nonce"], second[3]["x-loopdy-nonce"]) + self.assertEqual(self.pending(), []) + + def test_response_before_pre_and_late_pre_cannot_resurrect(self): + self.enroll_approval() + self.approval("post_approval_response", choice="deny") + self.approval() + self.assertEqual(self.pending(), []) + self.approval(tool_call_id="tool-b") + self.assertEqual(len(self.pending()), 1) + self.service.observe("on_session_end", profile="default", session_id="native-session", + turn_id="turn-a", interrupted=True) + self.approval(tool_call_id="tool-c") + self.assertEqual(self.pending(), []) + + def test_tool_end_retires_only_its_exact_attention(self): + self.enroll_approval() + self.approval() + self.approval(tool_call_id="tool-b") + self.service.observe("post_tool_call", profile="default", session_id="native-session", + turn_id="turn-a", tool_call_id="tool-a") + self.approval() + self.assertEqual(len(self.pending()), 1) + self.now += 4 + self.service.drain_pending() + self.assertEqual(len(self.event_calls()), 1) + + def test_new_process_drops_stale_attention_without_sending(self): + self.enroll_approval() + self.approval() + self.service.close() + self.service = ManagedNotifications(self.service.directory, transport=self.transport, + clock=lambda: self.now, session_opener=lambda profile, read, read_only: read(self)) + self.now += 4 + self.service.drain_pending() + self.assertEqual(self.event_calls(), []) + self.approval() + self.assertEqual(self.pending(), []) + + def test_api_reader_does_not_retire_live_producer_attention(self): + self.enroll_approval() + self.approval() + reader = ManagedNotifications(self.service.directory, transport=self.transport, + clock=lambda: self.now, session_opener=lambda profile, read, read_only: read(self)) + try: + self.assertEqual(len(self.pending()), 1) + self.now += 4 + self.service.drain_pending() + self.assertEqual(len(self.event_calls()), 1) + finally: + reader.close() + + def test_expired_attention_is_never_sent(self): + self.enroll_approval() + self.approval() + self.now += 61 + self.service.drain_pending() + self.assertEqual(self.event_calls(), []) + self.assertEqual(self.pending(), []) + + def test_response_during_final_session_check_prevents_transport(self): + self.enroll_approval() + self.approval() + def opener(profile, read, read_only): + self.approval("post_approval_response", choice="once") + return read(self) + self.service.session_opener = opener + self.now += 4 + self.service.drain_pending() + self.assertEqual(self.event_calls(), []) + self.assertEqual(self.pending(), []) + + def test_inflight_error_cannot_resurrect_answered_approval(self): + from loopdy_plugin.managed_notifications import ManagedNotificationError + self.enroll_approval() + self.approval() + def transport(method, path, raw, headers): + self.calls.append((method, path, raw, headers)) + self.approval("post_approval_response", choice="deny") + raise ManagedNotificationError("synthetic_timeout", 503) + self.service.transport = transport + self.now += 4 + self.service.drain_pending() + self.assertEqual(self.pending(), []) + self.now += 20 + self.service.drain_pending() + self.assertEqual(len(self.event_calls()), 1) + + def test_unsubscribe_fences_delayed_attention(self): + self.enroll_approval() + self.approval() + self.service.subscribe(self.grant_id, "default", "native-session", False) + self.now += 4 + self.service.drain_pending() + self.assertEqual(self.event_calls(), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_managed_notifications.py b/tests/test_managed_notifications.py new file mode 100644 index 0000000..b4310c0 --- /dev/null +++ b/tests/test_managed_notifications.py @@ -0,0 +1,183 @@ +from __future__ import annotations +import json +from pathlib import Path +import sqlite3 +import tempfile +import unittest +import uuid +from cryptography.hazmat.primitives.asymmetric import ec +from loopdy_plugin.managed_notifications import ManagedNotifications, ManagedNotificationError, host_request_transcript +from loopdy_plugin.relay_crypto import b64url_encode, b64url_decode, key_id, public_key_bytes, public_key_from_x963, verify_p1363 + + +class ManagedNotificationTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.now = 1_800_000_000 + self.calls = [] + self.fail_send = False + self.grant_id = str(uuid.uuid4()) + self.service = ManagedNotifications(Path(self.temp.name)/"managed", transport=self.transport, + clock=lambda: self.now, session_opener=lambda profile, read, read_only: read(self)) + recipient = public_key_bytes(ec.generate_private_key(ec.SECP256R1()).public_key()) + self.grant = dict(grantId=self.grant_id,hostKeyId=self.service.key_id,hostPublicKey=self.service.public_key, + deviceId="mobile-a",recipientPublicKey=b64url_encode(recipient),recipientKeyId=key_id(recipient), + recipientRevision=1,authorizationEpoch=1,profile="default",eventTypes=["session.completed","session.failed"], + createdAt=self.now-10,expiresAt=self.now+3600,revision=1,tenantId="fixture-tenant",state="active") + self.service.enroll(self.grant_id,str(uuid.uuid4())) + self.service.subscribe(self.grant_id,"default","native-session",True) + self.calls.clear() + + def tearDown(self): + self.service.close() + self.temp.cleanup() + + def get_session(self, sid): + return {"id":sid,"profile_name":"default"} + + def transport(self, method, path, raw, headers): + self.calls.append((method,path,raw,headers)) + public = public_key_from_x963(b64url_decode(self.service.public_key)) + signed = host_request_transcript(method,path,self.grant_id,int(headers["x-loopdy-timestamp"]),headers["x-loopdy-nonce"],raw) + verify_p1363(public,b64url_decode(headers["x-loopdy-signature"]),signed) + if path.endswith("/events"): + if self.fail_send: raise ManagedNotificationError("synthetic_unavailable",503) + return {"version":1,"status":"accepted","deliveryId":json.loads(raw)["envelope"]["delivery_id"]} + return {"version":1,"grant":self.grant} + + def test_capability_distinguishes_supported_events_from_lazy_producer_load(self): + caps=self.service.capabilities() + self.assertEqual(set(caps["supportedEventTypes"]),{"session.completed","session.failed","approval.required"}) + self.assertFalse(caps["producerCapabilities"]["nativeApproval"]) + self.assertFalse(caps["producerCapabilities"]["sessionCompletion"]) + self.service.producer_loaded("default",start_worker=False) + self.assertTrue(self.service.capabilities()["producerCapabilities"]["sessionCompletion"]) + self.assertFalse(self.service.capabilities()["producerCapabilities"]["nativeApproval"]) + self.service.producer_loaded("default",start_worker=False,approval_hooks_loaded=True) + self.assertTrue(self.service.capabilities()["producerCapabilities"]["nativeApproval"]) + self.service.close() + self.assertFalse(self.service.capabilities()["producerCapabilities"]["nativeApproval"]) + + def test_cancelled_work_queues_neutral_terminal_without_alert(self): + self.service.observe("pre_llm_call",profile="default",session_id="native-session",turn_id="turn-a",platform="desktop") + with sqlite3.connect(self.service.db_path) as db: + db.execute("INSERT INTO activities(activity_id,grant_id,profile,session_id,session_ref,lease_expires,work_turn,state) VALUES(?,?,?,?,?,?,?,'active')", + ("activity-a",self.grant_id,"default","native-session","x"*43,self.now+600,"turn-a")) + self.service.observe("on_session_end",profile="default",session_id="native-session",turn_id="turn-a",interrupted=True,platform="desktop") + with sqlite3.connect(self.service.db_path) as db: + row=db.execute("SELECT raw FROM pending WHERE activity_id='activity-a'").fetchone() + self.assertIsNotNone(row) + value=json.loads(row[0]);self.assertEqual(value["phase"],"completed");self.assertEqual(value["currentAction"],"Stopped") + self.assertEqual(db.execute("SELECT COUNT(*) FROM events").fetchone()[0],0) + + def test_late_activity_registration_keeps_original_turn_and_frozen_terminal(self): + from loopdy_plugin.managed_notifications import session_reference + reference = session_reference("default", "native-session") + original_transport = self.service.transport + def transport(method, path, raw, headers): + if method == "GET" and "/live-activities/" in path: + return {"version": 1, "activity": {"grantId": self.grant_id, + "activityId": "late-activity", "sessionReference": reference, + "status": "active", "leaseExpires": self.now + 600}} + return original_transport(method, path, raw, headers) + self.service.transport = transport + self.service.observe("pre_llm_call", profile="default", session_id="native-session", turn_id="original") + self.service.observe("on_session_end", profile="default", session_id="native-session", turn_id="original", completed=True) + self.service.observe("pre_llm_call", profile="default", session_id="native-session", turn_id="newer") + args=(self.grant_id,"late-activity","default","native-session",reference,self.now+600,"original") + self.service.subscribe_activity(*args) + with sqlite3.connect(self.service.db_path) as db: + first=db.execute("SELECT raw FROM pending WHERE activity_id='late-activity'").fetchone()[0] + self.assertEqual(json.loads(first)["phase"],"completed") + self.assertEqual(db.execute("SELECT work_turn FROM activities").fetchone()[0],"original") + self.service.subscribe_activity(*args) + with sqlite3.connect(self.service.db_path) as db: + self.assertEqual(db.execute("SELECT raw FROM pending WHERE activity_id='late-activity'").fetchone()[0],first) + self.assertEqual(self.service.work_snapshot(self.grant_id,"default","native-session")["work"]["turnId"],"newer") + with self.assertRaises(ManagedNotificationError): + self.service.subscribe_activity(*args[:-1],"invented") + + def test_cancelled_parent_waits_for_owned_child_before_neutral_terminal(self): + self.service.observe("pre_llm_call",profile="default",session_id="native-session",turn_id="parent") + self.service.observe("subagent_start",profile="default",parent_session_id="native-session",parent_turn_id="parent",child_session_id="child") + self.service.observe("on_session_end",profile="default",session_id="native-session",turn_id="parent",interrupted=True) + snapshot=self.service.work_snapshot(self.grant_id,"default","native-session")["work"] + self.assertEqual(snapshot["phase"],"delegating") + self.assertFalse(snapshot["terminal"]) + self.service.observe("subagent_stop",profile="default",parent_session_id="native-session",parent_turn_id="later-parent",child_session_id="child") + snapshot=self.service.work_snapshot(self.grant_id,"default","native-session")["work"] + self.assertEqual(snapshot["phase"],"completed") + self.assertEqual(snapshot["outcome"],"cancelled") + self.assertTrue(snapshot["terminal"]) + + def test_fresh_process_does_not_restore_live_work_from_subscription(self): + self.service.observe("pre_llm_call",profile="default",session_id="native-session",turn_id="old") + reopened=ManagedNotifications(Path(self.temp.name)/"managed",transport=self.transport,clock=lambda:self.now, + session_opener=lambda profile,read,read_only:read(self)) + try: + self.assertIsNone(reopened.work_snapshot(self.grant_id,"default","native-session")["work"]) + finally: + reopened.close() + + def test_current_work_snapshot_is_scoped_and_canonical(self): + self.service.observe("pre_llm_call",profile="default",session_id="native-session",turn_id="canonical-turn",platform="desktop") + snapshot = self.service.work_snapshot(self.grant_id,"default","native-session") + self.assertEqual(snapshot["work"]["turnId"],"canonical-turn") + self.assertEqual(snapshot["work"]["sessionId"],"native-session") + self.assertEqual(snapshot["work"]["phase"],"thinking") + self.assertNotIn("private",repr(snapshot).lower()) + with self.assertRaises(ManagedNotificationError): + self.service.work_snapshot(self.grant_id,"default","unsubscribed") + + def test_one_native_terminal_event_persists_and_retries_identical_ciphertext(self): + payload=dict(profile="default",session_id="native-session",turn_id="turn-a",completed=True,platform="desktop") + self.service.observe("on_session_end",**payload) + self.service.observe("on_session_end",**payload) + with sqlite3.connect(self.service.db_path) as db: + self.assertEqual(db.execute("SELECT COUNT(*) FROM events").fetchone()[0],1) + self.assertEqual(db.execute("SELECT COUNT(*) FROM pending").fetchone()[0],1) + self.fail_send=True + self.service.drain_pending() + first=self.calls[-1] + self.now+=5 + self.fail_send=False + self.service.drain_pending() + second=self.calls[-1] + self.assertEqual(first[2],second[2]) + self.assertNotEqual(first[3]["x-loopdy-nonce"],second[3]["x-loopdy-nonce"]) + self.assertTrue(json.loads(second[2])["eventId"].startswith(self.grant_id+":")) + with sqlite3.connect(self.service.db_path) as db: + self.assertEqual(db.execute("SELECT state FROM pending").fetchone()[0],"accepted") + + def test_unsubscribed_other_profile_children_and_cancellation_do_not_alert(self): + for change in [dict(session_id="other-session"),dict(profile="other"),dict(platform="subagent"),dict(interrupted=True)]: + payload=dict(profile="default",session_id="native-session",turn_id="turn-a",completed=True,platform="desktop")|change + self.service.observe("on_session_end",**payload) + self.service.drain_pending() + self.assertEqual(self.calls,[]) + + def test_local_revocation_cancels_pending_and_retains_identity_after_reopen(self): + self.service.observe("on_session_end",profile="default",session_id="native-session",turn_id="turn-a",failed=True,platform="desktop") + self.service.remove(self.grant_id) + self.service.drain_pending() + self.assertEqual(self.calls,[]) + self.service.close() + reopened=ManagedNotifications(Path(self.temp.name)/"managed",transport=self.transport,clock=lambda:self.now) + try: + self.assertEqual(reopened.public_key,self.service.public_key) + with self.assertRaises(ManagedNotificationError): reopened.enrollment(self.grant_id) + finally: reopened.close() + + def test_policy_suppression_is_durable_not_delayed_until_quiet_hours_end(self): + self.service.preference_policy=lambda event,device:{"suppression":"quiet_hours","sound":False} + payload=dict(profile="default",session_id="native-session",turn_id="turn-a",completed=True,platform="desktop") + self.service.observe("on_session_end",**payload) + self.service.preference_policy=None + self.service.observe("on_session_end",**payload) + self.service.drain_pending() + self.assertEqual(self.calls,[]) + with sqlite3.connect(self.service.db_path) as db: + self.assertEqual(db.execute("SELECT COUNT(*) FROM events").fetchone()[0],1) + self.assertEqual(db.execute("SELECT COUNT(*) FROM pending").fetchone()[0],0) + +if __name__ == "__main__": unittest.main() diff --git a/tests/test_native_notification_enrollment.py b/tests/test_native_notification_enrollment.py new file mode 100644 index 0000000..936f032 --- /dev/null +++ b/tests/test_native_notification_enrollment.py @@ -0,0 +1,44 @@ +from __future__ import annotations +import importlib.util +import os +from pathlib import Path +import sys +import tempfile +import unittest +from fastapi import FastAPI +from fastapi.testclient import TestClient + + +class NativeNotificationEnrollmentTests(unittest.TestCase): + def test_mounted_router_exposes_notification_identity_without_link_pairing(self): + with tempfile.TemporaryDirectory() as directory: + previous = os.environ.get("HERMES_HOME") + os.environ["HERMES_HOME"] = directory + try: + path = Path(__file__).resolve().parents[1] / "dashboard" / "plugin_api.py" + spec = importlib.util.spec_from_file_location("notification_bootstrap_test", path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + app = FastAPI() + app.include_router(module.router) + with TestClient(app) as client: + response = client.get("/notifications/capabilities") + self.assertEqual(response.status_code, 200) + body = response.json() + self.assertEqual(body["version"], 1) + self.assertTrue(body["managedEnrollmentSupported"]) + self.assertTrue(body["hostKeyId"]) + self.assertTrue(body["hostPublicKey"]) + self.assertNotIn("private", response.text.lower()) + finally: + if previous is None: + os.environ.pop("HERMES_HOME", None) + else: + os.environ["HERMES_HOME"] = previous + sys.modules.pop("notification_bootstrap_test", None) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_registration.py b/tests/test_registration.py index 3d2fa72..c014345 100644 --- a/tests/test_registration.py +++ b/tests/test_registration.py @@ -54,13 +54,30 @@ def register_approval_transport(self, name, present_fn): self.approval = (name, present_fn) def register_hook(self, name, callback): - self.hooks[name] = callback + previous = self.hooks.get(name) + if previous is None: + self.hooks[name] = callback + return + def dispatch(*args, **kwargs): + first = previous(*args, **kwargs) + second = callback(*args, **kwargs) + return second if second is not None else first + self.hooks[name] = dispatch def register_cli_command(self, **kwargs): self.cli = kwargs def on_unload(self, callback): - self.unload = callback + previous = self.unload + if previous is None: + self.unload = callback + else: + def unload(): + try: + callback() + finally: + previous() + self.unload = unload class _Service: @@ -918,6 +935,7 @@ def test_native_surfaces_and_nonblocking_hooks_are_registered(self) -> None: set(context.hooks), { "pre_approval_request", + "post_approval_response", "pre_tool_call", "post_tool_call", "pre_llm_call",