Gabriel Vidal
6c793d2dca
Rich forms the agent asks the user to fill, replacing Claude Code's interactive AskUserQuestion for spawned sessions: - backend/forms.py + /api/forms REST (create / list / get with waitSecs long-poll / submit / cancel), JSON sidecar store at /data/forms.json, 'form' SSE events; field types: text textarea number select multiselect radio checkbox slider file date, answers validated server-side - conversation viewer renders the ask-form Bash call as a live inline form card (submit with confirm recap, cancel, file uploads via /api/upload); answers stay in the thread read-only after submit; wait/cancel subcommands render as compact status strips; new 'Form cards' visibility switch - ?form=<id> query param (the notification deep link) opens the form focused in a full-screen modal - sidecar disallows AskUserQuestion on spawned claude runs (SIDECAR_DISALLOWED_TOOLS to override) - mock backend: /api/forms routes + seeded pending/submitted/cancelled forms and kitchen-sink thread cards (SEED_VERSION 7) Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
281 lines
11 KiB
Python
281 lines
11 KiB
Python
"""
|
|
Forms — rich structured questions the agent asks the user, inline in the
|
|
conversation.
|
|
|
|
The ``ask-form`` skill POSTs ``/api/forms`` with a field spec (text inputs,
|
|
select, multi-select, radio, checkboxes, sliders, file uploads, …), then sends
|
|
a normal push notification linking to the conversation with ``?form=<id>`` so
|
|
the PWA opens the form focused in a full-screen modal. The conversation thread
|
|
also renders the form inline on the Bash tool card that created it; submitting
|
|
POSTs the answers back and the skill's ``wait`` command (which the agent runs
|
|
under a Monitor, not a blocking foreground call) exits with them.
|
|
|
|
This replaces Claude Code's interactive AskUserQuestion tool for spawned
|
|
sessions (the sidecar disallows it) — unlike a 2-3-button phone ask, a form
|
|
can carry many typed fields and both harnesses (claude and pi) can drive it,
|
|
since it's just HTTP against the backend.
|
|
|
|
Store shape (single JSON file, atomic rewrite, same pattern as ``notify.py``):
|
|
|
|
{ "forms": [ {"id", "title", "description", "type", "url", "sessionId",
|
|
"at", "fields": [...], "status": "pending"|"submitted"|
|
|
"cancelled", "answers", "submittedAt", "cancelledAt"}, … ] }
|
|
|
|
The log is capped (newest kept). Waiting for a submit is in-memory (one
|
|
``threading.Event`` per pending form) — a backend restart drops the wait; the
|
|
skill's poll loop just re-polls.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import threading
|
|
import uuid
|
|
from datetime import datetime, timezone
|
|
|
|
_MAX_LOG = 500
|
|
_ID_RE = re.compile(r"[^a-z0-9_-]+")
|
|
|
|
# Field types the UI knows how to render. An unknown type is rejected at
|
|
# create time — better a loud 400 for the skill than a card the user can't
|
|
# answer.
|
|
FIELD_TYPES = (
|
|
"text", "textarea", "number", "select", "multiselect", "radio",
|
|
"checkbox", "slider", "file", "date",
|
|
)
|
|
_OPTION_TYPES = ("select", "multiselect", "radio")
|
|
|
|
|
|
def _now_iso() -> str:
|
|
return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
|
|
|
|
|
def _slug(s: str) -> str:
|
|
return _ID_RE.sub("-", (s or "").strip().lower()).strip("-")
|
|
|
|
|
|
def _num(v, fallback=None):
|
|
try:
|
|
f = float(v)
|
|
except (TypeError, ValueError):
|
|
return fallback
|
|
return int(f) if f.is_integer() else f
|
|
|
|
|
|
def valid_fields(raw: list) -> list[dict]:
|
|
"""Validate + normalize a field spec list; raises ValueError on bad input."""
|
|
if not isinstance(raw, list) or not raw:
|
|
raise ValueError("fields must be a non-empty list")
|
|
out: list[dict] = []
|
|
seen: set[str] = set()
|
|
for i, f in enumerate(raw):
|
|
if not isinstance(f, dict):
|
|
raise ValueError(f"field #{i}: must be an object")
|
|
ftype = str(f.get("type") or "text").strip().lower()
|
|
if ftype not in FIELD_TYPES:
|
|
raise ValueError(
|
|
f"field #{i}: unknown type {ftype!r} (one of {', '.join(FIELD_TYPES)})")
|
|
label = str(f.get("label") or "").strip()
|
|
key = _slug(str(f.get("key") or "")) or _slug(label)
|
|
if not key:
|
|
raise ValueError(f"field #{i}: needs a key or a label")
|
|
base, n = key, 2
|
|
while key in seen:
|
|
key = f"{base}-{n}"
|
|
n += 1
|
|
seen.add(key)
|
|
field: dict = {
|
|
"key": key,
|
|
"label": label or key,
|
|
"type": ftype,
|
|
"required": bool(f.get("required", False)),
|
|
}
|
|
for opt_key in ("placeholder", "help", "accept"):
|
|
v = str(f.get(opt_key) or "").strip()
|
|
if v:
|
|
field[opt_key] = v
|
|
if ftype in _OPTION_TYPES:
|
|
opts = []
|
|
for o in (f.get("options") or []):
|
|
if isinstance(o, dict):
|
|
val = str(o.get("value") if o.get("value") is not None
|
|
else o.get("label") or "").strip()
|
|
lab = str(o.get("label") or val).strip()
|
|
else:
|
|
val = lab = str(o).strip()
|
|
if val:
|
|
opts.append({"value": val, "label": lab or val})
|
|
if len(opts) < 2:
|
|
raise ValueError(f"field {key!r}: {ftype} needs >= 2 options")
|
|
field["options"] = opts
|
|
if ftype in ("number", "slider"):
|
|
for k in ("min", "max", "step"):
|
|
v = _num(f.get(k))
|
|
if v is not None:
|
|
field[k] = v
|
|
if ftype == "slider":
|
|
field.setdefault("min", 0)
|
|
field.setdefault("max", 100)
|
|
field.setdefault("step", 1)
|
|
if ftype == "file" and f.get("multiple"):
|
|
field["multiple"] = True
|
|
if f.get("default") is not None:
|
|
field["default"] = f.get("default")
|
|
out.append(field)
|
|
return out
|
|
|
|
|
|
def check_answers(fields: list[dict], answers: dict) -> dict:
|
|
"""Validate submitted answers against the spec; raises ValueError.
|
|
|
|
Returns the answers reduced to known keys — lenient on shapes (the UI is
|
|
the trusted producer) but strict on required fields and option membership,
|
|
so the agent never reads back a value the form couldn't have produced."""
|
|
if not isinstance(answers, dict):
|
|
raise ValueError("answers must be an object")
|
|
out: dict = {}
|
|
for f in fields:
|
|
key, ftype = f["key"], f["type"]
|
|
v = answers.get(key)
|
|
empty = v is None or v == "" or v == []
|
|
if empty:
|
|
if f.get("required") and ftype != "checkbox":
|
|
raise ValueError(f"field {f['label']!r} is required")
|
|
if ftype == "checkbox":
|
|
out[key] = bool(v)
|
|
continue
|
|
if ftype in ("select", "radio"):
|
|
allowed = {o["value"] for o in f.get("options") or []}
|
|
if str(v) not in allowed:
|
|
raise ValueError(f"field {f['label']!r}: {v!r} not an option")
|
|
out[key] = str(v)
|
|
elif ftype == "multiselect":
|
|
allowed = {o["value"] for o in f.get("options") or []}
|
|
vals = [str(x) for x in (v if isinstance(v, list) else [v])]
|
|
bad = [x for x in vals if x not in allowed]
|
|
if bad:
|
|
raise ValueError(f"field {f['label']!r}: {bad} not options")
|
|
out[key] = vals
|
|
elif ftype == "checkbox":
|
|
out[key] = bool(v)
|
|
elif ftype in ("number", "slider"):
|
|
n = _num(v)
|
|
if n is None:
|
|
raise ValueError(f"field {f['label']!r}: not a number")
|
|
out[key] = n
|
|
elif ftype == "file":
|
|
files = v if isinstance(v, list) else [v]
|
|
keep = []
|
|
for x in files:
|
|
if isinstance(x, dict) and (x.get("repoPath") or x.get("url")):
|
|
keep.append({k: x[k] for k in
|
|
("name", "size", "contentType", "repoPath",
|
|
"url") if k in x})
|
|
out[key] = keep
|
|
else:
|
|
out[key] = str(v)
|
|
return out
|
|
|
|
|
|
class FormStore:
|
|
"""Form definitions + answers in one JSON sidecar."""
|
|
|
|
def __init__(self, path: str):
|
|
self.path = pathlib.Path(path)
|
|
self._lock = threading.Lock()
|
|
self._waiters: dict[str, threading.Event] = {}
|
|
self._data: dict = {"forms": []}
|
|
self._load()
|
|
|
|
def _load(self) -> None:
|
|
try:
|
|
data = json.loads(self.path.read_text(encoding="utf-8"))
|
|
self._data = {"forms": list(data.get("forms") or [])}
|
|
except (OSError, ValueError):
|
|
self._data = {"forms": []}
|
|
|
|
def _save(self) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = self.path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(self._data, indent=2), encoding="utf-8")
|
|
os.replace(tmp, self.path)
|
|
|
|
def create(self, title: str, fields: list[dict], *, description: str = "",
|
|
type_: str = "", url: str = "", session_id: str = "") -> dict:
|
|
entry = {
|
|
"id": "frm_" + uuid.uuid4().hex[:12],
|
|
"title": title, "description": description or "",
|
|
"type": type_ or "", "url": url or "",
|
|
"sessionId": session_id or "", "at": _now_iso(),
|
|
"fields": fields, "status": "pending",
|
|
"answers": None, "submittedAt": None, "cancelledAt": None,
|
|
}
|
|
with self._lock:
|
|
self._waiters[entry["id"]] = threading.Event()
|
|
self._data["forms"].append(entry)
|
|
if len(self._data["forms"]) > _MAX_LOG:
|
|
self._data["forms"] = self._data["forms"][-_MAX_LOG:]
|
|
self._save()
|
|
return json.loads(json.dumps(entry))
|
|
|
|
def get(self, fid: str) -> dict | None:
|
|
with self._lock:
|
|
for f in self._data["forms"]:
|
|
if f.get("id") == fid:
|
|
return json.loads(json.dumps(f))
|
|
return None
|
|
|
|
def list(self, *, session_id: str | None = None,
|
|
limit: int = 100) -> list[dict]:
|
|
with self._lock:
|
|
forms = [f for f in self._data["forms"]
|
|
if not session_id or f.get("sessionId") == session_id]
|
|
out = forms[-max(1, limit):]
|
|
out.reverse() # newest first
|
|
return json.loads(json.dumps(out))
|
|
|
|
def _finish(self, fid: str, patch: dict) -> dict | None:
|
|
"""Apply a terminal patch; idempotent — a form that already reached a
|
|
terminal state is returned untouched (a double-submit or a cancel
|
|
racing a submit can't clobber the recorded answers)."""
|
|
with self._lock:
|
|
cur = next((f for f in self._data["forms"]
|
|
if f.get("id") == fid), None)
|
|
if not cur:
|
|
return None
|
|
if cur.get("status") == "pending":
|
|
cur.update(patch)
|
|
self._save()
|
|
ev = self._waiters.pop(fid, None)
|
|
out = json.loads(json.dumps(cur))
|
|
if ev:
|
|
ev.set()
|
|
return out
|
|
|
|
def submit(self, fid: str, answers: dict) -> dict | None:
|
|
"""Record the answers; raises ValueError on invalid input."""
|
|
cur = self.get(fid)
|
|
if not cur:
|
|
return None
|
|
checked = check_answers(cur.get("fields") or [], answers)
|
|
return self._finish(fid, {"status": "submitted", "answers": checked,
|
|
"submittedAt": _now_iso()})
|
|
|
|
def cancel(self, fid: str) -> dict | None:
|
|
return self._finish(fid, {"status": "cancelled",
|
|
"cancelledAt": _now_iso()})
|
|
|
|
def wait(self, fid: str, wait_s: float) -> dict | None:
|
|
"""Block up to ``wait_s`` for a submit/cancel; returns the form.
|
|
|
|
The waiter event is (re-)armed lazily, so waits survive a backend
|
|
restart (which empties ``_waiters``) — the poll loop just re-arms."""
|
|
cur = self.get(fid)
|
|
if not cur or cur.get("status") != "pending" or wait_s <= 0:
|
|
return cur
|
|
with self._lock:
|
|
ev = self._waiters.setdefault(fid, threading.Event())
|
|
ev.wait(min(wait_s, 55.0))
|
|
return self.get(fid)
|