Gabriel Vidal
3d741b5b1e
Notifications follow the conversation's audience (users/access.audience):
whoever started it and the people it's shared with, plus the owner once they
follow a guest's conversation (writing in it, or PUT /api/conversation-follow),
minus whoever muted it (meta.follow). Pushes and asks fan out to each of them
on their own channels — the owner on the webhooks, a guest in-app + email —
and the feeds (/api/notifications, /api/notify-log, the phone's /unread)
list only what the caller hears, links rebased to the reader's host. Forms
carry forYou so "waiting for feedback" is per user.
/api/users carries spentUsd (this month), totalSpentUsd and conversations per
guest (users/runs.spend_by_user, the same sum the budget is checked against).
POST /api/users/{id}/impersonate opens a 4-hour guest session for the owner's
browser on the admin host (imp on the record: never extended, not counted in
the guest's sessions); /api/me says impersonating; logout ends it. A spawn
while impersonating skips the "guest started" ping.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
254 lines
11 KiB
Python
254 lines
11 KiB
Python
"""Creating and recording pushes and asks, and the flat notification feed."""
|
|
|
|
import datetime
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import uuid as uuidlib
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from conversations.cards import _conv_meta, _conv_title
|
|
from conversations.catalog import _conv_ids_by_session
|
|
from conversations.tree import _conv_id
|
|
from core.config import SOURCE_DIRS
|
|
from core.state import AppState
|
|
from notifications import store as notify_mod
|
|
from users import access, deliver
|
|
from users.caller import OWNER, OWNER_ID, Caller
|
|
|
|
# ── flat notification feed (server-side mirror of the frontend's buildFeed) ──
|
|
# The PWA flattens /api/notifications + /api/notify-log into one list of feed
|
|
# items (lib/notifications.ts). The phone can't run that, so we do the same
|
|
# flattening here — same stable ids — for the unread endpoint and the read
|
|
# ledger. Keep the two in sync: id scheme is `<convId>#<at>#<index>` for
|
|
# conv-linked pushes and `log#<id>` for hub-only entries.
|
|
_DEDUP_WINDOW_MS = 120_000
|
|
|
|
|
|
def _parse_iso_ms(at: str | None) -> float | None:
|
|
if not at:
|
|
return None
|
|
try:
|
|
s = at.replace("Z", "+00:00")
|
|
return datetime.datetime.fromisoformat(s).timestamp() * 1000.0
|
|
except (ValueError, AttributeError):
|
|
return None
|
|
|
|
|
|
def _flat_notifications(deps: AppState) -> list[dict]:
|
|
"""Every notification **the owner hears** as a flat, de-duplicated,
|
|
newest-first feed item — the exact set the owner's PWA shows, so "unread"
|
|
means the same on the phone. A guest's conversation the owner doesn't
|
|
follow isn't in it (``users/access.audience``)."""
|
|
feed: list[dict] = []
|
|
for path, s in deps.store.all_summaries():
|
|
if not s.get("messages") or s.get("isSidechain"):
|
|
continue
|
|
sid = s.get("sessionId") or ""
|
|
meta = deps.meta_store.peek(sid)
|
|
notes = meta.get("notifications") or []
|
|
if not notes or not access.hears(OWNER, meta):
|
|
continue
|
|
m = _conv_meta(deps, s)
|
|
cid = _conv_id(path)
|
|
title = _conv_title(deps, s)
|
|
last_seen: dict[str, float] = {}
|
|
for i, n in enumerate(notes):
|
|
key = (f"{n.get('title') or ''} {n.get('body') or ''} "
|
|
f"{n.get('type') or ''} {n.get('url') or ''}")
|
|
t = _parse_iso_ms(n.get("at"))
|
|
prev = last_seen.get(key)
|
|
if prev is not None and (t is None or abs(t - prev) <= _DEDUP_WINDOW_MS):
|
|
continue # duplicate push within the conversation
|
|
if t is not None:
|
|
last_seen[key] = t
|
|
feed.append({
|
|
"id": f"{cid}#{n.get('at') or ''}#{i}",
|
|
"convId": cid, "convTitle": title,
|
|
"title": n.get("title") or "", "body": n.get("body") or "",
|
|
"type": n.get("type") or "", "url": n.get("url") or "",
|
|
"image": n.get("image") or "",
|
|
"at": n.get("at"),
|
|
"projects": m.get("projects") or [],
|
|
"services": m.get("services") or [],
|
|
"kind": "notify",
|
|
"spoken": n.get("spoken") or "",
|
|
})
|
|
# Merge the hub log: a log entry that matches a conv push (title+body within
|
|
# the dedup window) is a duplicate and skipped; hub-only entries are added.
|
|
for e in deps.notify_store.log(limit=500):
|
|
if not log_heard(deps, OWNER, e):
|
|
continue
|
|
t = _parse_iso_ms(e.get("at"))
|
|
dup = False
|
|
for n in feed:
|
|
if n["title"] == (e.get("title") or "") and \
|
|
n["body"] == (e.get("body") or ""):
|
|
nt = _parse_iso_ms(n.get("at"))
|
|
if t is None or nt is None or abs(nt - t) <= _DEDUP_WINDOW_MS:
|
|
dup = True
|
|
break
|
|
if dup:
|
|
continue
|
|
feed.append({
|
|
"id": f"log#{e.get('id')}",
|
|
"convId": "", "convTitle": "",
|
|
"title": e.get("title") or "", "body": e.get("body") or "",
|
|
"type": e.get("type") or "", "url": e.get("url") or "",
|
|
"image": e.get("image") or "",
|
|
"at": e.get("at"),
|
|
"projects": [], "services": [],
|
|
"kind": e.get("kind") or "notify",
|
|
})
|
|
feed.sort(key=lambda n: n.get("at") or "", reverse=True)
|
|
return feed
|
|
|
|
|
|
def log_heard(deps: AppState, caller: Caller, entry: dict) -> bool:
|
|
"""Is a hub-log entry (push / ask) one ``caller`` hears? A session's
|
|
entries follow its conversation's audience as it stands now; one with no
|
|
session (a host script, a guest-start ping) is the owner's."""
|
|
sid = entry.get("sessionId") or ""
|
|
if not sid:
|
|
return caller.is_owner
|
|
return access.hears(caller, deps.meta_store.peek(sid))
|
|
|
|
|
|
# Where the viewer lives, for the links a push carries when the caller gave
|
|
# no --url: the conversation that sent it.
|
|
VIEW_URL = os.environ.get("AI_AGENT_VIEW_URL",
|
|
"https://ai-agent.lab.gabvdl.xyz").rstrip("/")
|
|
_CLIENT_ID_RE = re.compile(r"^(ntf|ask|frm)_[0-9a-f]{12}$")
|
|
|
|
|
|
def _client_id(given: str | None, prefix: str) -> str:
|
|
"""A caller-minted id (the notify CLI's), validated, else a fresh one."""
|
|
given = (given or "").strip()
|
|
if given:
|
|
if not _CLIENT_ID_RE.match(given) or not given.startswith(prefix + "_"):
|
|
raise HTTPException(400, f"bad id {given!r} (want {prefix}_<12 hex>)")
|
|
return given
|
|
return f"{prefix}_{uuidlib.uuid4().hex[:12]}"
|
|
|
|
|
|
def _conversation_url(deps: AppState, session_id: str) -> str:
|
|
"""The viewer page of the session that is notifying — the default link
|
|
of every push, ask and form. Resolved **here** rather than in the CLI
|
|
because only the hub knows where a worker's transcript was mirrored to:
|
|
the indexed summary first, else the transcript on disk under any source
|
|
dir (a session too young to be indexed), else the conversations list."""
|
|
sid = (session_id or "").strip()
|
|
if not sid:
|
|
return f"{VIEW_URL}/conversations"
|
|
cid = _conv_ids_by_session(deps, ).get(sid)
|
|
if not cid:
|
|
for root in SOURCE_DIRS:
|
|
for hit in pathlib.Path(root).glob(f"*/{sid}.jsonl"):
|
|
cid = f"{hit.parent.name}/{hit.name}"
|
|
break
|
|
if cid:
|
|
break
|
|
return f"{VIEW_URL}/conversation/{cid}" if cid else f"{VIEW_URL}/conversations"
|
|
|
|
|
|
def _stamp_notified(deps: AppState, session_id: str, note: dict, final: bool) -> None:
|
|
"""What ``conv-meta notify`` used to do from the skill: record the push on
|
|
the conversation's metadata (the viewer's notification feed + the
|
|
``notified`` lifecycle check) and, for the closing push, mark it
|
|
finished. Best-effort; never fails the push."""
|
|
sid = (session_id or "").strip()
|
|
if not sid:
|
|
return
|
|
patch: dict = {"notification": note}
|
|
if final:
|
|
patch["state"] = "finished"
|
|
try:
|
|
deps.meta_store.update(sid, patch)
|
|
deps.hub.publish({"type": "meta", "id": sid})
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _ok(to: list[str], delivered: list[dict]) -> bool:
|
|
"""Every delivery went through — and something was delivered, unless
|
|
nobody was meant to hear it (everyone muted the conversation)."""
|
|
return all(x["ok"] for x in delivered) and (bool(delivered) or not to)
|
|
|
|
|
|
def _deliver(deps: AppState, to: list[str], event: str, payload: dict,
|
|
title: str, message: str, url: str) -> list[dict]:
|
|
"""Reach everyone in ``to`` on their channels: the owner through the
|
|
webhooks, each guest through theirs (users/deliver.py)."""
|
|
out: list[dict] = []
|
|
if OWNER_ID in to:
|
|
out += [{**r, "user": OWNER_ID} for r in
|
|
notify_mod.forward(deps.notify_store.targets(event), payload)]
|
|
return out + deliver.to_guests(deps, to, title, message, url, VIEW_URL)
|
|
|
|
|
|
def _create_notify(deps: AppState, d: dict) -> dict:
|
|
"""Record a push and deliver it to the conversation's audience. ``d`` is
|
|
a NotifyBody dump — the route and the outbox mirror both come through
|
|
here. The owner hears it on the webhooks, each guest on their own
|
|
channels (users/deliver.py); a push with no session is the owner's."""
|
|
nid = _client_id(d.get("id"), "ntf")
|
|
sid = (d.get("sessionId") or "").strip()
|
|
if not (d.get("url") or "").strip():
|
|
d["url"] = _conversation_url(deps, sid)
|
|
to = deliver.recipients(deps, sid)
|
|
actions = [a for a in (d.get("actions") or [])
|
|
if isinstance(a, dict) and a.get("action") and a.get("title")]
|
|
payload = {"event": "notify", "id": nid,
|
|
**{k: v for k, v in d.items()
|
|
if k not in ("sessionId", "id", "final") and v not in (None, "", [])}}
|
|
delivered = _deliver(deps, to, "notify", payload, d.get("title") or "",
|
|
d.get("message") or "", d["url"])
|
|
entry = {
|
|
"id": nid, "kind": "notify", "title": d.get("title") or "",
|
|
"body": d.get("message") or "", "type": d.get("type") or "",
|
|
"url": d.get("url") or "", "image": d.get("image") or "",
|
|
"at": notify_mod._now_iso(), "sessionId": sid,
|
|
"to": to, "delivered": delivered,
|
|
# Kept (not just forwarded) so the UI can replay what the phone said.
|
|
"spoken": d.get("spoken") or "",
|
|
}
|
|
if actions:
|
|
entry.update({"actions": actions, "status": "sent",
|
|
"actionIndex": None, "actionTitle": None, "actedAt": None})
|
|
deps.notify_store.watch_actions(nid)
|
|
deps.notify_store.record(entry)
|
|
_stamp_notified(deps, sid, {"title": entry["title"], "body": entry["body"],
|
|
"type": entry["type"], "url": entry["url"],
|
|
"at": entry["at"]}, bool(d.get("final")))
|
|
deps.hub.publish({"type": "notification", "id": nid})
|
|
return {"ok": _ok(to, delivered), "id": nid, "delivered": delivered,
|
|
"url": entry["url"]}
|
|
|
|
|
|
def _create_ask(deps: AppState, d: dict) -> dict:
|
|
"""Create an ask, forward it, return its id (AskBody dump in)."""
|
|
options = [o.strip() for o in (d.get("options") or []) if o and o.strip()]
|
|
if not (d.get("question") or "").strip() or not 2 <= len(options) <= 3:
|
|
raise HTTPException(400, "need a question and 2-3 options")
|
|
nid = _client_id(d.get("id"), "ask")
|
|
sid = (d.get("sessionId") or "").strip()
|
|
url = (d.get("url") or "").strip() or _conversation_url(deps, sid)
|
|
to = deliver.recipients(deps, sid)
|
|
ask = deps.notify_store.create_ask(
|
|
d["question"].strip(), options, type_=d.get("type") or "",
|
|
url=url, image=d.get("image") or "", session_id=sid,
|
|
timeout_s=d.get("timeoutSecs") or notify_mod.DEFAULT_ASK_TIMEOUT_S,
|
|
nid=nid)
|
|
payload = {"event": "ask", "id": ask["id"], "question": ask["body"],
|
|
"options": options, "url": url,
|
|
**{k: v for k, v in d.items()
|
|
if k in ("type", "image", "icon", "color")
|
|
and v not in (None, "")}}
|
|
delivered = _deliver(deps, to, "ask", payload, "❓ " + ask["body"],
|
|
" / ".join(options), url)
|
|
deps.notify_store.set_delivered(ask["id"], delivered, to)
|
|
deps.hub.publish({"type": "notification", "id": ask["id"]})
|
|
return {"ok": _ok(to, delivered), "id": ask["id"], "delivered": delivered,
|
|
"url": url}
|