Gabriel Vidal
deb01427d5
newproject/: the policy (guestDefaults.newProjects — quota, templates, the ones the owner approves first, visibility, level, domain — and a per-guest createProjects + address bucket), the New project form built from it, and the flow: POST /api/new-project/form (the run's guesses as defaults) → submit → shape/policy and host checks (a problem sends a corrected form back) → an owner ask for a backend template → sidecar /scaffold → grant → /relocate + resume the same session inside the project with an app-authored turn. SpawnBody.newProject starts a guest's run in ~/projects/.new-project/<uid> (staging_scope: /tmp only); /api/me reports newProjects; the thread follows ConversationDetail.continuedIn. Fix: a guest's turn into an existing conversation resumed with no cwd, so the sidecar ran it in the lab checkout (its CLAUDE.md loaded) — it now resumes where the session lives. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
454 lines
23 KiB
Python
454 lines
23 KiB
Python
"""
|
|
The *New project* flow, from the form to the conversation going on inside the
|
|
project it made. The same steps for the owner and for guests:
|
|
|
|
1. :func:`open_form` — the run asks for the form with its guesses (``notify
|
|
new-project``, as the new-frontend-project skill says); it is built from
|
|
the requester's policy (``policy.options``) and goes to the conversation's
|
|
audience like any form. The run then ends its turn.
|
|
2. :func:`on_form_submitted` — the forms route calls it on every submit; only
|
|
the forms in the store are ours. The answers are checked here (shape,
|
|
policy) and on the host (the generator's ``--check``: name, address and
|
|
repository free); a problem sends a fresh form back, pre-filled, with the
|
|
problems on top.
|
|
3. A guest's template in ``approve`` waits for the owner: an Approve / Decline
|
|
ask on their phone (:func:`on_ask_answered`).
|
|
4. :func:`_create`, on a thread: the host sidecar scaffolds the project
|
|
(``/scaffold``); a guest is granted it; the session is copied into the
|
|
project's directory (``/relocate``) and resumed there with what was made —
|
|
the same conversation, now inside the project, builds the first version.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import os
|
|
import re
|
|
import threading
|
|
|
|
from fastapi import HTTPException
|
|
|
|
from newproject import form as form_mod
|
|
from newproject import policy
|
|
from projects.catalog import PROJECTS_DIR, PROJECTS_PROMPT_DIR
|
|
from users.caller import OWNER, OWNER_ID, OWNER_NAME, guest_from_record
|
|
|
|
log = logging.getLogger("ai-agent.newproject")
|
|
|
|
# A guest's New project run has no project yet: it starts in its own empty
|
|
# directory (a dot-dir, so the projects catalog skips it).
|
|
STAGING_DIR = ".new-project"
|
|
# npm install + the first push; the sidecar's own cap is 900 s.
|
|
SCAFFOLD_TIMEOUT_S = float(os.environ.get("NEW_PROJECT_SCAFFOLD_TIMEOUT_S", "960"))
|
|
APPROVE, DECLINE = "Create it", "Decline"
|
|
# The app's own turns in a thread (a conversation's author tag, users/runs.py).
|
|
APP_TAG = '<ai-agent-from user="new-project" name="New project" role="app"/>'
|
|
|
|
_NAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,38}[a-z0-9])?$")
|
|
_LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
|
|
|
|
|
|
# ── who ──────────────────────────────────────────────────────────────────────
|
|
def requester_of(deps, sid: str) -> str:
|
|
"""Whose project a conversation's form makes: its guest owner, else the
|
|
owner."""
|
|
return deps.meta_store.peek(sid).get("owner") or OWNER_ID
|
|
|
|
|
|
def _caller(deps, uid: str):
|
|
if uid == OWNER_ID:
|
|
return OWNER
|
|
u = deps.users.active(uid)
|
|
return guest_from_record(u) if u else None
|
|
|
|
|
|
def _name(deps, uid: str) -> str:
|
|
if uid == OWNER_ID:
|
|
return OWNER_NAME
|
|
return (deps.users.get(uid) or {}).get("name") or uid
|
|
|
|
|
|
def staging_cwd(uid: str) -> str:
|
|
"""The host directory a guest's New project run starts in — created here
|
|
(the backend mounts the projects dir) so the sidecar finds it."""
|
|
d = PROJECTS_DIR / STAGING_DIR / uid
|
|
if not d.is_dir():
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
st = PROJECTS_DIR.stat()
|
|
for p in (d.parent, d):
|
|
try: # the container may run as root; runs are the host user
|
|
os.chown(p, st.st_uid, st.st_gid)
|
|
except OSError:
|
|
pass
|
|
return f"{PROJECTS_PROMPT_DIR}/{STAGING_DIR}/{uid}"
|
|
|
|
|
|
def staging_note(caller) -> str:
|
|
"""What a New project spawn's prompt gets appended: the run's one job."""
|
|
who = "they" if not caller.is_owner else "you"
|
|
return ("\n\n---\n**New project.** This conversation starts a new project from one of "
|
|
f"{OWNER_NAME}'s templates. Follow the `new-frontend-project` skill"
|
|
+ ("" if caller.is_owner else " (`shared:new-frontend-project`)")
|
|
+ ": read the request, then open the *New project* form with your best "
|
|
"guesses (`notify new-project …`) and end your turn — the app creates the "
|
|
f"project when {who} submit it and this conversation carries on inside it.")
|
|
|
|
|
|
# ── 1. the form ──────────────────────────────────────────────────────────────
|
|
def open_form(deps, sid: str, guesses: dict, problems: list[dict] | None = None) -> dict:
|
|
"""Send the conversation's requester a New project form, pre-filled with
|
|
``guesses`` (``name``, ``title``, ``description``, ``template``,
|
|
``label``, ``bucket``, ``visibility``, ``summary``)."""
|
|
from forms.service import _create_form
|
|
sid = (sid or "").strip()
|
|
if not sid:
|
|
raise HTTPException(400, "sessionId is required (the conversation the form goes to)")
|
|
meta = deps.meta_store.peek(sid)
|
|
if meta.get("worker") and meta.get("worker") != "local":
|
|
raise HTTPException(409, {"message": "new projects are created on the lab: this "
|
|
"session runs on a worker", "kind": "new_project"})
|
|
requester = requester_of(deps, sid)
|
|
caller = _caller(deps, requester)
|
|
if caller is None:
|
|
raise HTTPException(403, {"message": f"{requester} is not an enabled user",
|
|
"kind": "new_project"})
|
|
acc = policy.access(deps, caller)
|
|
if not acc["allowed"]:
|
|
raise HTTPException(403, {"message": f"{_name(deps, requester)} can't start a "
|
|
f"project: {acc['reason']}",
|
|
"kind": "new_project"})
|
|
opts = policy.options(deps, requester)
|
|
if not opts["templates"]:
|
|
raise HTTPException(409, {"message": "no template to start from", "kind": "new_project"})
|
|
clean = {k: str(v).strip() for k, v in (guesses or {}).items()
|
|
if k in ("name", "title", "description", "template", "label", "bucket",
|
|
"visibility", "summary") and v is not None}
|
|
f = _create_form(deps, {
|
|
"title": form_mod.TITLE, "description": form_mod.description(opts, None),
|
|
"fields": form_mod.fields(opts, clean, problems), "type": "code",
|
|
"sessionId": sid, "notify": True})
|
|
deps.new_projects.create(f["id"], sid, requester, clean)
|
|
deps.hub.publish({"type": "new-project", "id": f["id"]})
|
|
return {"formId": f["id"], "focusUrl": f.get("focusUrl") or "", "requester": requester,
|
|
"options": opts}
|
|
|
|
|
|
# ── 2. the answers ───────────────────────────────────────────────────────────
|
|
def on_form_submitted(deps, form: dict) -> None:
|
|
fid = form.get("id") or ""
|
|
if deps.new_projects.claim(fid, ("pending",), "checking") is None:
|
|
return
|
|
threading.Thread(target=_guarded, args=(deps, fid, _review, form.get("answers") or {}),
|
|
daemon=True, name=f"new-project-{fid}").start()
|
|
|
|
|
|
def on_form_cancelled(deps, form: dict) -> None:
|
|
deps.new_projects.claim(form.get("id") or "", ("pending",), "cancelled")
|
|
|
|
|
|
def _guarded(deps, fid: str, fn, *args) -> None:
|
|
try:
|
|
fn(deps, fid, *args)
|
|
except HTTPException as e: # the sidecar's, or a refused re-opened form
|
|
log.warning("new project %s failed: %s", fid, e.detail)
|
|
d = e.detail
|
|
_fail(deps, fid, d.get("message") if isinstance(d, dict) else str(d))
|
|
except Exception as e: # a thread: nobody else would see it
|
|
log.exception("new project %s failed", fid)
|
|
_fail(deps, fid, f"{type(e).__name__}: {e}")
|
|
|
|
|
|
def _spec(opts: dict, answers: dict) -> tuple[dict, list[dict]]:
|
|
"""The answers as a scaffold spec, plus whatever's wrong with them."""
|
|
name = str(answers.get("name") or "").strip().lower()
|
|
label = str(answers.get("label") or "").strip().lower() or name
|
|
bucket = opts["bucket"] or str(answers.get("bucket") or "").strip().lower() or "dev"
|
|
template = str(answers.get("template") or "")
|
|
visibility = str(answers.get("visibility") or opts["visibilityDefault"])
|
|
problems = []
|
|
if not _NAME_RE.match(name) or name in ("templates", "node_modules"):
|
|
problems.append({"field": "name", "message": f"“{name}” can't be a project name — "
|
|
"lowercase letters, digits and dashes, 2 to 40 characters"})
|
|
if not _LABEL_RE.match(label):
|
|
problems.append({"field": "label", "message": f"“{label}” can't be a web address — "
|
|
"lowercase letters, digits and dashes"})
|
|
if not _LABEL_RE.match(bucket):
|
|
problems.append({"field": "subdomain", "message": f"“{bucket}” can't be a subdomain"})
|
|
if template not in opts["templates"]:
|
|
problems.append({"field": "template", "message": f"“{template}” isn't on offer"})
|
|
if visibility not in opts["visibility"]:
|
|
problems.append({"field": "visibility", "message": f"“{visibility}” isn't on offer"})
|
|
if opts["domain"] != "gabvdl.xyz":
|
|
# The generator and the templates' deploy scripts only know gabvdl.xyz.
|
|
problems.append({"field": "label", "message": f"sites on {opts['domain']} aren't "
|
|
"supported yet"})
|
|
host = f"{label}.{bucket}.{opts['domain']}"
|
|
spec = {"name": name, "title": str(answers.get("title") or name).strip()[:80] or name,
|
|
"description": str(answers.get("description") or "").strip()[:400],
|
|
"template": template, "label": label, "bucket": bucket,
|
|
"domain": opts["domain"], "host": host, "url": f"https://{host}",
|
|
"visibility": visibility}
|
|
return spec, problems
|
|
|
|
|
|
def _vars(deps, spec: dict, requester: str) -> dict:
|
|
v = {"NAME": spec["name"], "TITLE": spec["title"], "DESCRIPTION": spec["description"],
|
|
"LABEL": spec["label"], "SUBDOMAIN": spec["bucket"], "URL": spec["url"]}
|
|
if requester != OWNER_ID:
|
|
# Each guest has a Gitea account of their own (made on their first
|
|
# project, never logged into): their repositories live under it.
|
|
u = deps.users.get(requester) or {}
|
|
v.update({"GITEA_OWNER": requester, "GITEA_EMAIL": u.get("email") or "",
|
|
"GITEA_FULL_NAME": u.get("name") or requester})
|
|
return v
|
|
|
|
|
|
def _sidecar(deps, path: str, payload: dict, timeout: float) -> dict:
|
|
from runs.sidecar_client import _sidecar_call
|
|
return _sidecar_call(deps, path, payload, timeout=timeout)
|
|
|
|
|
|
def _host_problems(deps, spec: dict, requester: str) -> list[dict]:
|
|
out = _sidecar(deps, "/scaffold/check", {"template": spec["template"],
|
|
"vars": _vars(deps, spec, requester)}, 90)
|
|
res = out.get("result") or {}
|
|
if out.get("ok"):
|
|
return []
|
|
if res.get("problems"):
|
|
return list(res["problems"])
|
|
raise RuntimeError("the host couldn't check the project: "
|
|
+ _tail(out.get("log") or out.get("error") or "", 4))
|
|
|
|
|
|
def _review(deps, fid: str, answers: dict) -> None:
|
|
req = deps.new_projects.get(fid)
|
|
requester, sid = req["requester"], req["sessionId"]
|
|
opts = policy.options(deps, requester)
|
|
if opts["quota"] is not None and deps.new_projects.used(requester) >= opts["quota"]:
|
|
# Two forms opened side by side both passed the check at opening.
|
|
raise RuntimeError(f"the quota of {opts['quota']} projects is reached")
|
|
spec, problems = _spec(opts, answers)
|
|
if not problems:
|
|
problems = _host_problems(deps, spec, requester)
|
|
if problems:
|
|
guesses = {**{k: answers.get(k) for k in ("name", "title", "description", "template",
|
|
"label", "bucket", "visibility")},
|
|
"summary": (req.get("guesses") or {}).get("summary")}
|
|
nxt = open_form(deps, sid, guesses, problems)
|
|
deps.new_projects.update(fid, {"status": "retried", "answers": answers,
|
|
"next": nxt["formId"],
|
|
"error": "; ".join(p["message"] for p in problems)})
|
|
return
|
|
deps.new_projects.update(fid, {"answers": answers, "spec": spec,
|
|
"visibility": spec["visibility"],
|
|
"template": spec["template"], "url": spec["url"]})
|
|
if requester != OWNER_ID and spec["template"] in opts["approve"]:
|
|
_ask_approval(deps, fid, spec)
|
|
return
|
|
if deps.new_projects.claim(fid, ("checking",), "creating"):
|
|
_create(deps, fid)
|
|
|
|
|
|
# ── 3. the owner's OK ────────────────────────────────────────────────────────
|
|
def _ask_approval(deps, fid: str, spec: dict) -> None:
|
|
from notifications.service import _conversation_url, _create_ask, _create_notify
|
|
req = deps.new_projects.get(fid)
|
|
who = _name(deps, req["requester"])
|
|
ask = _create_ask(deps, {
|
|
"question": f"{who} wants a new project: {spec['title']} ({spec['template']} — a "
|
|
f"backend on raspy2) at {spec['url']}. Create it?",
|
|
"options": [APPROVE, DECLINE], "sessionId": "", "type": "code",
|
|
"url": _conversation_url(deps, req["sessionId"]), "timeoutSecs": 14 * 86400})
|
|
if deps.new_projects.claim(fid, ("checking",), "approval", {"askId": ask["id"]}) is None:
|
|
return
|
|
_create_notify(deps, {
|
|
"title": f"⏳ {spec['title']} waits for {OWNER_NAME}'s OK",
|
|
"message": f"{spec['template']} runs a backend on the server, so {OWNER_NAME} "
|
|
"approves it first — you'll hear back here.",
|
|
"type": "code", "sessionId": req["sessionId"]})
|
|
|
|
|
|
def on_ask_answered(deps, ask: dict) -> None:
|
|
req = deps.new_projects.by_ask(ask.get("id") or "")
|
|
if req is None:
|
|
return
|
|
if ask.get("answer") == APPROVE:
|
|
if deps.new_projects.claim(req["id"], ("approval",), "creating"):
|
|
threading.Thread(target=_guarded, args=(deps, req["id"], _create),
|
|
daemon=True, name=f"new-project-{req['id']}").start()
|
|
return
|
|
if deps.new_projects.claim(req["id"], ("approval",), "declined") is None:
|
|
return
|
|
spec = req.get("spec") or {}
|
|
_say(deps, req["sessionId"],
|
|
f"{OWNER_NAME} declined **{spec.get('title') or 'the project'}** "
|
|
f"(`{spec.get('template')}`). Tell them, kindly; if a template without a "
|
|
"backend would do, offer to open the *New project* form again with it "
|
|
"(`notify new-project`).")
|
|
|
|
|
|
# ── 4. making it ─────────────────────────────────────────────────────────────
|
|
def _create(deps, fid: str) -> None:
|
|
from users.runs import project_cwd
|
|
req = deps.new_projects.get(fid)
|
|
spec, requester, sid = req["spec"], req["requester"], req["sessionId"]
|
|
deps.meta_store.update(sid, {"state": "running"})
|
|
deps.hub.publish({"type": "meta", "id": sid})
|
|
deps.hub.publish({"type": "new-project", "id": fid})
|
|
out = _sidecar(deps, "/scaffold", {"template": spec["template"],
|
|
"vars": _vars(deps, spec, requester),
|
|
"password": spec["visibility"] == "password"},
|
|
SCAFFOLD_TIMEOUT_S)
|
|
res = out.get("result") or {}
|
|
if not out.get("ok"):
|
|
_fail(deps, fid, out.get("error") or "the scaffold failed", out.get("log") or "")
|
|
return
|
|
slug = spec["name"]
|
|
level = policy.policy(deps)["level"]
|
|
if requester != OWNER_ID:
|
|
u = deps.users.get(requester) or {}
|
|
deps.users.update(requester, {"projects": {**(u.get("projects") or {}), slug: level}})
|
|
deps.hub.publish({"type": "users"})
|
|
deps.new_projects.update(fid, {"slug": slug, "url": res.get("url") or spec["url"],
|
|
"repo": res.get("repo"), "error": res.get("gitError")})
|
|
error = None
|
|
try:
|
|
_move_and_resume(deps, sid, requester, slug, project_cwd(slug),
|
|
_made_prompt(deps, spec, res, out.get("log") or "", requester, level))
|
|
except Exception as e: # the project exists either way: say so, don't redo it
|
|
log.exception("new project %s: moving the conversation failed", fid)
|
|
error = f"made, but the conversation didn't move into it: {e}"
|
|
deps.new_projects.update(fid, {"status": "created",
|
|
**({"error": error} if error else {})})
|
|
deps.hub.publish({"type": "new-project", "id": fid})
|
|
if requester != OWNER_ID:
|
|
_ping_owner(deps, sid, f"👤 {_name(deps, requester)} created {spec['title']}",
|
|
f"{spec['template']} · {spec['url']}")
|
|
|
|
|
|
def _move_and_resume(deps, sid: str, requester: str, slug: str, cwd: str, prompt: str) -> None:
|
|
"""The session goes on in the project's directory: copy it there (the
|
|
sidecar stops a live run first), then resume it with ``prompt``."""
|
|
from runs.service import _send_message, _session_account
|
|
from users import runs as guest_runs
|
|
_sidecar(deps, "/relocate", {"sessionId": sid, "cwd": cwd,
|
|
"account": _session_account(deps, sid)}, 30)
|
|
deps.meta_store.update(sid, {"projects": [slug]})
|
|
scope = None
|
|
if requester != OWNER_ID:
|
|
caller = _caller(deps, requester)
|
|
scope = guest_runs.scope_for(caller, slug, caller.level(slug),
|
|
guest_runs.shared_skills(deps),
|
|
new_projects=policy.scope_line(deps, caller))
|
|
_send_message(deps, prompt, session_id=sid, cwd=cwd, harness="claude", scope=scope)
|
|
|
|
|
|
def _made_prompt(deps, spec: dict, res: dict, log_text: str, requester: str,
|
|
level: str) -> str:
|
|
slug, url = spec["name"], res.get("url") or spec["url"]
|
|
who = _name(deps, requester)
|
|
lines = [APP_TAG,
|
|
f"**{spec['title']}** is created from the `{spec['template']}` template, in "
|
|
f"`{PROJECTS_PROMPT_DIR}/{slug}` — this session now runs there.", ""]
|
|
lines.append(f"- Address: {url} — live after the first deploy.")
|
|
if res.get("repo"):
|
|
lines.append(f"- Repository: {res['repo']}")
|
|
elif res.get("gitError"):
|
|
lines.append(f"- Repository: not pushed ({res['gitError']}) — retry with "
|
|
"`commit-project` once you have commits.")
|
|
if requester != OWNER_ID:
|
|
lines.append(f"- {who} has **{level}** access to it.")
|
|
if spec["visibility"] == "password":
|
|
if res.get("password"):
|
|
lines.append(f"- It's behind a login page. The password is `{res['password']}` "
|
|
"— give it to them in your final message (deploys keep the lock).")
|
|
else:
|
|
lines.append("- It was meant to be behind a password but the lock failed: "
|
|
f"run `zipgo password set {spec['host']} --generate` before deploying.")
|
|
if spec["visibility"] == "geo":
|
|
lines.append("- It's for France + Germany only: right after the first successful "
|
|
f"deploy, run `geo-lock add {spec['host']}` (the site must be served "
|
|
"first).")
|
|
deploy = ("commit, deploy with `npm run deploy`, check the site answers, and finish "
|
|
f"with `notify-done --final --url {url}`"
|
|
if spec["template"] != "datakit" else
|
|
"commit, and finish with `notify-done --final`")
|
|
lines += ["", "Now build the first version of what was asked for in this conversation: "
|
|
"replace the starter page, keep the template's conventions (its notes "
|
|
f"below), {deploy}."]
|
|
notes = _summary(log_text)
|
|
if notes:
|
|
lines += ["", "Scaffold notes:", "```", notes, "```"]
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _summary(log_text: str) -> str:
|
|
"""The generator's SCAFFOLD SUMMARY (+ TODO) section, if the log has it."""
|
|
i = log_text.find("SCAFFOLD SUMMARY")
|
|
text = log_text[i:] if i >= 0 else ""
|
|
text = "\n".join(ln for ln in text.splitlines()
|
|
if not ln.startswith(("SCAFFOLD_RESULT=", "Site password")))
|
|
return text.strip()[:4000]
|
|
|
|
|
|
def _tail(text: str, n: int) -> str:
|
|
return " / ".join(ln.strip() for ln in text.strip().splitlines()[-n:] if ln.strip())
|
|
|
|
|
|
def _fail(deps, fid: str, error: str, log_text: str = "") -> None:
|
|
req = deps.new_projects.update(fid, {"status": "failed",
|
|
"error": (error + (" — " + _tail(log_text, 6)
|
|
if log_text else ""))[:1000]})
|
|
deps.hub.publish({"type": "new-project", "id": fid})
|
|
if not req:
|
|
return
|
|
title = (req.get("spec") or {}).get("title") or "the project"
|
|
_ping_owner(deps, req["sessionId"], f"⚠️ New project failed: {title}",
|
|
f"{_name(deps, req['requester'])} · {error}"[:300])
|
|
try:
|
|
_say(deps, req["sessionId"],
|
|
f"Creating **{title}** failed: {error}. Tell them it didn't work and that "
|
|
f"{OWNER_NAME} was told; don't try to scaffold it by hand.")
|
|
except Exception:
|
|
log.exception("could not report the failure of %s", fid)
|
|
|
|
|
|
def _say(deps, sid: str, text: str) -> None:
|
|
"""An app turn in the conversation, run where it already is."""
|
|
from runs.service import _send_message
|
|
from users import runs as guest_runs
|
|
requester = requester_of(deps, sid)
|
|
scope = None
|
|
if requester != OWNER_ID:
|
|
caller = _caller(deps, requester)
|
|
slug = (deps.meta_store.peek(sid).get("projects") or [None])[0]
|
|
if caller and slug and caller.can(slug, "chat"):
|
|
scope = guest_runs.scope_for(caller, slug, caller.level(slug),
|
|
guest_runs.shared_skills(deps))
|
|
elif caller:
|
|
scope = {**guest_runs.scope_for(caller, slug or "-", None,
|
|
guest_runs.shared_skills(deps)),
|
|
"writeDirs": ["/tmp"]}
|
|
_send_message(deps, f"{APP_TAG}\n{text}", session_id=sid, harness="claude",
|
|
cwd=session_cwd(deps, sid), scope=scope)
|
|
|
|
|
|
def session_cwd(deps, sid: str) -> str | None:
|
|
"""Where a session last ran: the cwd of its newest transcript."""
|
|
from conversations.catalog import _conv_ids_by_session
|
|
from core.config import TRANSCRIPTS_DIR
|
|
conv = _conv_ids_by_session(deps).get(sid)
|
|
if not conv:
|
|
return None
|
|
summ = deps.store.get_summary(str(TRANSCRIPTS_DIR / conv)) or {}
|
|
return summ.get("cwd") or None
|
|
|
|
|
|
def _ping_owner(deps, sid: str, title: str, message: str) -> None:
|
|
from notifications.service import _conversation_url, _create_notify
|
|
try:
|
|
_create_notify(deps, {"title": title, "message": message, "type": "code",
|
|
"url": _conversation_url(deps, sid), "sessionId": ""})
|
|
except Exception as exc:
|
|
log.warning("owner ping failed: %s", exc)
|