"""routes_forms.py — ⭐ wave-23 item 8 (contract W23-C9): THE PUBLIC FORM DOOR.

⛔ BOTH ROUTES HERE ARE UNAUTHENTICATED, AND THAT IS THE POINT. A form link goes to somebody who
has no account — a supplier, a lead, a warehouse hand with a phone — so `Depends(require_session)`
is exactly what must NOT be on them. Everything that would normally be decided by a session is
therefore decided by the TOKEN, and this file is the whole of that decision:

  * the token is minted server-side (`secrets.token_urlsafe(24)` — 192 bits) and stored on ONE
    view. There is no enumeration to brute-force and no id in the URL to increment.
  * a wrong token answers ONE thing for every tenant, every table and every typo: 403
    `bad_form_token`. It never leaks whether a form exists, whether a tenant exists, or which of
    the two you got wrong — the automation-hook pattern (`routes_automation.py:359-381`) one
    level up, because a public door that answers 404-vs-403 is an oracle.
  * the payload a valid token buys is the SMALLEST thing that can render a form: title,
    description, submit label, and the fields with their labels, types and options. No tenant
    slug, no table key, no row count, no field key that is not on the form, no username.

WHAT AN ATTACKER WITH A VALID TOKEN CAN DO, stated plainly because it is the honest boundary:
append rows to ONE user table, up to the per-token daily cap, with values that pass this file's
validation. They cannot read a row, cannot see any other field, cannot reach another table, and
cannot cause an automation to run as anybody (`form_fired` is a HUMAN door by A's design — see
its contract note in the split doc).

⚠ THE `_DEV_FIXTURES` AUDIT (main.py:257-295) applied to this router, because that scar is
exactly "a thing became publicly reachable and nobody looked": the only bytes these routes can
return are the ones assembled in `_public_form` below, field by field. There is no passthrough of
a stored dict anywhere in this file — every response is built from named keys.

⚠ PROTECTIONS ARE IN-PROCESS AND THEREFORE PER WORKER. One Space process today, so the sliding
window is real; the day there are two, both halves of a limit are halved-per-worker rather than
bypassed. Booked as DEBT rather than described as a guarantee — a rate limit that quietly stops
being one is worse than none.
"""
import hmac
import os
import secrets
import time

from fastapi import APIRouter, Depends, Request

from deps import Session, err, require_session

router = APIRouter(prefix="/api/v1")

#: ⭐⭐ WAVE-29 R7 — THE TOKEN INDEX, one bucket per tenant: `{token: {table, view}}`.
#:
#: ⛔ WHY THE TOKEN IS NOT IN THE FORM SPEC, which is where wave 23 assumed it would live. The spec
#: rides `view.config.display.form`, which a BROWSER writes — so a token stored beside it is a
#: token a browser can choose. `_resolve` walks tenants and answers with the FIRST match, so a
#: tenant that set its token to a value it had seen elsewhere would quietly receive another
#: tenant's submissions. Nothing about that is exotic: it needs one `PATCH` and a token somebody
#: pasted into a chat. Minting server-side into a bucket the client cannot write closes it by
#: construction rather than by validation, and `aios_grid._clean_form` drops a wire `token`.
#:
#: ⭐ It is also FASTER than what it replaces. The old resolution walked every table's workspace,
#: every user's stratum and every view of every tenant looking for a matching token — booked in
#: this file's own header as "a FULL SCAN, deliberately, and its cost is BOOKED". Now it is one
#: small dict read per tenant, and the view is loaded ONCE, after the token has already matched.
TOKENS_KEY = "form_tokens"

#: The mint. 24 bytes url-safe = 32 characters, 192 bits — the same shape the automation hook's
#: per-automation token uses, so there is ONE token strength in the product rather than two.
TOKEN_BYTES = 24

#: The honeypot's field name. It is rendered by `FormPublic.tsx` as a real input, visually
#: hidden and `autocomplete="off"` + `tabIndex={-1}` — a human never sees it and a keyboard user
#: never lands on it, so anything that fills it is a script walking the DOM. ⚠ Named for
#: something a bot WANTS to fill; a field called `honeypot` is one `if` away from being skipped.
HONEYPOT_FIELD = "website_url"

#: v1 request protections (contract C9). Per IP and per token respectively.
RATE_WINDOW_S = 60
RATE_PER_WINDOW = 30
DAILY_PER_TOKEN = 500
MAX_BODY_BYTES = 16 * 1024
#: One text answer's ceiling. Generous for a paragraph, far below the body cap, and it exists so
#: a single 15 KB answer cannot be smuggled past a body check that only measures the whole.
MAX_VALUE_LEN = 4000

#: `{ip: [timestamps]}` and `{token: (day, count)}`. Module-level and unbounded-by-design within
#: a day: an IP list is pruned to the window on every touch, and the token map holds one small
#: tuple per form that was submitted to today.
_HITS: dict = {}
_DAILY: dict = {}


def _rate_ok(ip: str, now: float) -> bool:
    """A sliding window, not a fixed bucket. A fixed one lets a caller spend the whole allowance
    at 11:59:59 and the whole next allowance at 12:00:00 — double the rate at the boundary,
    which is the moment a script is most likely to be hammering."""
    seen = [t for t in _HITS.get(ip, ()) if now - t < RATE_WINDOW_S]
    seen.append(now)
    _HITS[ip] = seen
    # ⚠ Prune the WHOLE map occasionally, or a process that has served a million IPs holds a
    # million lists forever. Cheap and amortised: only when the map is already large.
    if len(_HITS) > 4096:
        for k in [k for k, v in _HITS.items() if not v or now - v[-1] > RATE_WINDOW_S]:
            _HITS.pop(k, None)
    return len(seen) <= RATE_PER_WINDOW


def _daily_ok(token: str, now: float) -> bool:
    day = int(now // 86400)
    d, n = _DAILY.get(token, (day, 0))
    if d != day:
        d, n = day, 0
    n += 1
    _DAILY[token] = (d, n)
    return n <= DAILY_PER_TOKEN


def _client_ip(request: Request) -> str:
    """⚠ `X-Forwarded-For`'s FIRST entry, and only because this app runs behind exactly one
    proxy (the HF Space router). It is caller-controlled, so it is a rate-limit key and NOTHING
    else — never a permission input. Falls back to the socket peer."""
    fwd = request.headers.get("x-forwarded-for") or ""
    first = fwd.split(",")[0].strip()
    return first or (request.client.host if request.client else "?")


def _refuse():
    """THE ONE REFUSAL. Every failure to resolve a token answers this — wrong token, disabled
    form, deleted view, deleted table, a tenant whose store is down. Callers must not branch a
    more specific message out of it: the difference between "no such form" and "that form is
    disabled" tells an enumerator which tokens are real."""
    return err(403, "bad_form_token", "that form link is not valid")


def _same(a: str, b: str) -> bool:
    """Constant-time. A `==` here leaks a token's prefix through timing, one character at a
    time, which is the whole reason the automation hook compares this way too."""
    return hmac.compare_digest(str(a or ""), str(b or ""))


# --- resolution ------------------------------------------------------------------------------
#
# ⚠ A FULL SCAN, deliberately, and its cost is BOOKED (DEBT — "the form-token index"). The
# automation hook resolves the same way (`routes_automation.py:370` walks `known_tenants()`)
# because the alternative — a global token→tenant map — is a second index that has to be written
# on every mint, regenerate, view delete and table delete, and an index that misses one of those
# is a live form that answers 403 or, far worse, a dead token that still resolves. With one
# process, a handful of tenants and a 30/min ceiling in front of it, the scan is affordable; the
# index becomes worth its risk when either number grows.


def _find_view(rt, table_key: str, view_id: str):
    """`(view, owner_username)` for one view in one table's workspace, or `(None, None)`.

    The bucket is `{username: {views: {id: view}}}` — one stratum per user plus the shared one —
    and a view id is unique across them, so the first match is the view.
    """
    try:
        bucket = rt.get(f"{table_key}_table_workspace") or {}
    except Exception:                                             # noqa: BLE001
        return None, None
    if not isinstance(bucket, dict):
        return None, None
    for owner, ws in bucket.items():
        if not isinstance(ws, dict):
            continue
        view = (ws.get("views") or {}).get(view_id)
        if isinstance(view, dict):
            return view, str(owner)
    return None, None


def _spec_of(view) -> dict:
    """The form spec on a view, or `{}`. One place knows the key path."""
    spec = (((view or {}).get("config") or {}).get("display") or {}).get("form")
    return spec if isinstance(spec, dict) else {}


def _tokens(rt) -> dict:
    """This tenant's `{token: {table, view}}` index, always a dict."""
    try:
        found = rt.get(TOKENS_KEY) or {}
    except Exception:                                             # noqa: BLE001
        return {}
    return found if isinstance(found, dict) else {}


def _resolve(token: str):
    """`(runtime, tenant_slug, table_key, form_spec)` for a token, or None.

    ⚠ THE LOOP DOES NOT SHORT-CIRCUIT ON THE TENANT, only on the match. A `break` on the first
    tenant whose store answered would make "the third tenant's form" depend on the first two
    being healthy, and a form that stops working because somebody else's store blipped is the
    kind of failure nobody can reproduce.

    ⛔ THE INDEX IS A POINTER, NEVER A PERMISSION. A token that resolves to a view which has since
    been DELETED, or whose form spec has been removed, answers the same 403 as a token that never
    existed — the index is written by the mint route and nothing deletes the tenant's views on its
    behalf, so "the row is in the index" cannot be allowed to mean "the form is live". A stale
    pointer outliving its view is exactly how a door that was closed stays open.
    """
    from harness import runtime as _rt

    if not token or len(token) < 16:
        return None
    for slug in _rt.known_tenants():
        try:
            rt = _rt.get_runtime(slug)
        except Exception:                                         # noqa: BLE001
            continue
        for stored, where in _tokens(rt).items():
            if not _same(str(stored), token) or not isinstance(where, dict):
                continue
            table_key = str(where.get("table") or "")
            view, _owner = _find_view(rt, table_key, str(where.get("view") or ""))
            spec = _spec_of(view)
            # No view, no questions, no form. Checked HERE rather than by the caller so every
            # public route inherits it from one place.
            if not table_key or not spec.get("fields"):
                return None
            return rt, slug, table_key, spec
    return None


# --- the public shape -------------------------------------------------------------------------


def _public_form(rt, table_key: str, spec: dict) -> dict:
    """The ONLY bytes a valid token buys. Built key by key — there is no `**stored` anywhere in
    this function, which is what makes "nothing else leaks" a property of the code rather than a
    promise about what happens to be in the bag today.

    ⚠ The field ORDER is the form's own (`spec['fields']`), not the table's: the builder let
    somebody arrange these questions, and re-deriving the order from the schema would silently
    rearrange a form every time a column was added.
    """
    import aios_grid
    import core.user_tables as user_tables

    defn = user_tables.get(table_key, st=rt) or {}
    by_key = {f.get("key"): f for f in (defn.get("fields") or []) if isinstance(f, dict)}
    required = {str(k) for k in (spec.get("required") or [])}
    out = []
    for key in (spec.get("fields") or []):
        f = by_key.get(str(key))
        # A field deleted since the form was built is DROPPED, never rendered as a dead input —
        # and never an error: losing one column must not take a live form offline.
        if not f or f.get("source") not in (None, "overlay"):
            continue
        # ⭐ WAVE-29 R7 — TWO WALLS, AND NEITHER IS REDUNDANT.
        #
        # (1) The TYPE allow-list: a form may only ask questions this door can answer honestly.
        #     `_clean_values` below coerces twelve types and passes everything else through as
        #     text, so an `image`, a `json` document or a `link` reached through a form would be
        #     a string typed into a column that means something else.
        # (2) `is_computed_cell`: the SERVER owns that number. And it catches what the type list
        #     structurally cannot — a METRIC bag rides an ordinary `int`, so a machine-computed
        #     column passes the allow-list by type and must still be refused.
        #
        # ⚠ Enforced HERE rather than only in the builder, because the builder is a courtesy and
        # this is the wall: the spec is a stored bag, and a field can become computed (a formula,
        # a rollup, a metric binding) long after it was added to a live form
        # ([[schema-role-is-not-a-value-wall]]).
        if str(f.get("type") or "text") not in aios_grid.FORM_FIELD_TYPES:
            continue
        if user_tables.is_computed_cell(f):
            continue
        row = {"key": str(f.get("key")), "label": str(f.get("label") or f.get("key")),
               "type": str(f.get("type") or "text"), "required": str(key) in required}
        options = f.get("options")
        if isinstance(options, list) and options:
            row["options"] = [str(o) for o in options][:200]
        if f.get("type") == "rating":
            row["max"] = int(f.get("max") or 5)
        out.append(row)
    return {
        "title": str(spec.get("title") or "")[:200],
        "desc": str(spec.get("desc") or "")[:1000],
        "submitLabel": str(spec.get("submitLabel") or "")[:60] or "Submit",
        "fields": out,
        "honeypot": HONEYPOT_FIELD,
        # ⛔ THE FLAG, NEVER THE LIST. `requireEmail` tells the page to ask for an address; the
        # ADDRESSES stay server-side. Returning them would hand an unauthenticated caller the
        # guest list — every colleague's work email, to anyone who has the link — which is a
        # worse leak than the door this restriction exists to close. It is also the only reason
        # this key can be computed here rather than shipped as `spec['emails']`.
        "requireEmail": spec.get("access") == "emails",
    }


def _invited(spec: dict, email: str) -> bool:
    """May this address submit? Case- and space-insensitive against the stored list.

    ⚠ **IDENTIFICATION, NOT AUTHENTICATION, and the honest boundary is stated rather than
    implied** (this file's header does the same for the token). Nothing here proves the submitter
    owns the address they typed: anyone holding the link who knows an invited address can pass.
    What it buys is real but bounded — a link forwarded outside the invited group stops working
    for whoever cannot name a member of it, and every submission carries the address it claimed.
    A per-recipient link, which WOULD prove possession, is booked rather than half-built.
    """
    if spec.get("access") != "emails":
        return True
    wanted = str(email or "").strip().lower()
    return bool(wanted) and wanted in {str(e).strip().lower()
                                       for e in (spec.get("emails") or [])}


# --- validation -------------------------------------------------------------------------------


def _clean_values(fields: list, raw: dict):
    """`(values, error_sentence)` — REFUSE, NEVER COERCE (C9, and the engine's own law).

    The temptation with a public form is to be forgiving: read "12 units" as 12, "yes" as
    checked, "tomorrow" as a date. Every one of those writes a number nobody typed into somebody
    else's database, and the person who submitted the form is not there to see it happen. So a
    value that does not answer its own field's question is refused with a sentence naming the
    FIELD LABEL — the only name the submitter has ever seen.
    """
    out = {}
    for f in fields:
        key, label, ftype = f["key"], f["label"], f["type"]
        v = raw.get(key)
        text = "" if v is None else str(v)
        text = text.strip() if isinstance(v, str) else text
        if len(text) > MAX_VALUE_LEN:
            return None, f"“{label}” is too long — {MAX_VALUE_LEN} characters at most."
        if text == "":
            if f.get("required"):
                return None, f"“{label}” is required."
            continue
        if ftype in ("int", "currency", "pct", "rating"):
            try:
                num = float(text.replace(",", ""))
            except ValueError:
                return None, f"“{label}” must be a number."
            if ftype == "rating":
                top = int(f.get("max") or 5)
                if not (1 <= num <= top) or num != int(num):
                    return None, f"“{label}” must be a whole number from 1 to {top}."
            text = str(int(num)) if ftype in ("int", "rating") else str(num)
        elif ftype == "checkbox":
            # The storage contract is '1' or blank (types.ts). Anything a checkbox can actually
            # send is one of these four spellings; anything else is not a checkbox answer.
            if text.lower() not in ("1", "true", "on", "yes", "0", "false", "off", "no"):
                return None, f"“{label}” must be checked or unchecked."
            text = "1" if text.lower() in ("1", "true", "on", "yes") else ""
        elif ftype in ("select", "status"):
            options = f.get("options") or []
            if options and text not in options:
                return None, f"“{label}” must be one of the listed choices."
        elif ftype == "multiselect":
            options = f.get("options") or []
            parts = [p.strip() for p in text.split(",") if p.strip()]
            if options and any(p not in options for p in parts):
                return None, f"“{label}” must be chosen from the listed choices."
            text = ", ".join(parts)
        elif ftype == "date":
            # ISO only. A public form has no timezone, no locale and no user to ask, so
            # "03/04/2026" is genuinely ambiguous and guessing it wrong is a silent data error.
            if len(text) != 10 or text[4] != "-" or text[7] != "-":
                return None, f"“{label}” must be a date."
            try:
                time.strptime(text, "%Y-%m-%d")
            except ValueError:
                return None, f"“{label}” must be a real date."
        elif ftype == "email":
            if "@" not in text or text.startswith("@") or text.endswith("@"):
                return None, f"“{label}” must be an email address."
        out[key] = text
    return out, ""


# --- the routes -------------------------------------------------------------------------------


# --- the AUTHENTICATED half: minting, reading and revoking a form's share link ----------------
#
# ⚠ PATH: `/form-link`, NOT `/forms/link`. The public routes are `/forms/{token}`, and a literal
# segment beside a path parameter is decided by declaration ORDER — a reader (or a later
# refactor that moves a decorator) would have to know that to know which handler answers. A
# different noun cannot be shadowed by a token somebody chooses.


def _public_base() -> str:
    """The origin a share link is built on. Empty = the client builds it from its own location.

    `AIOS_PUBLIC_BASE` is the same variable the OAuth callback resolves against, so a domain move
    takes the form links with it. Absent, we return a PATH rather than guessing a hostname:
    a link with the wrong origin looks right, gets pasted into an email and 404s for the reader.
    """
    return (os.environ.get("AIOS_PUBLIC_BASE") or os.environ.get("APP_BASE_URL") or "").rstrip("/")


def _link_of(token: str) -> str:
    return f"{_public_base()}/#/form/{token}"


def _may_administer_view(session: Session, table_key: str, view_id: str):
    """`(view, spec)` when this caller may publish THIS view, else raises.

    ⭐ `_may_administer`, deliberately, not `_may_edit`. Minting a public link is a PERMISSIONS
    act — it opens a write door onto the tenant's table for anybody holding the URL — and
    `table_store` already draws that line for views: a collaborator may edit a shared view, only
    its creator or an admin may change who reaches it. Publishing is the strongest form of that
    question, so it takes the narrower answer.
    ⚠ A view sitting in the caller's OWN stratum is theirs even with no `createdBy` stamp (a
    personal view predating the stamp has none), which `_may_administer` alone would refuse.
    """
    import core.table_store as table_store

    if not table_key.startswith("ut_"):
        # `add_row` is the only write a form can perform and it is a `ut_*` door; an Odoo-backed
        # grid has no such door, so a link over one could never be anything but a 500 later.
        raise err(400, "not_a_database", "forms collect into your own databases only")
    view, owner = _find_view(session.runtime, table_key, view_id)
    if not isinstance(view, dict):
        raise err(404, "no_such_view", "that view no longer exists")
    # ⛔ `uname` / `admin`, NOT `username` / `is_admin`. `Session` has no such attributes, and
    # reading one raises OUTSIDE any handler's own try: — which is exactly D-107, where four
    # finished Odoo routes were mounted, uncallable, and answered a bare plain-text 500 for a day
    # ([[mounted-is-not-callable]]). Written wrong here first; caught by this file's own gate.
    if not (owner == session.uname
            or table_store._may_administer(view, session.uname, session.admin)):
        raise err(403, "not_yours", "only this view's creator or an admin can publish it")
    return view, _spec_of(view)


@router.get("/form-link")
def read_form_link(topic: str, view: str, session: Session = Depends(require_session)):
    """This form's current share link, or `{"token": null}` — never a mint.

    ⛔ A GET MUST NOT MINT. The builder reads this every time the panel opens, and a route that
    created a token on read would publish a form the moment somebody looked at the tab.
    """
    _v, spec = _may_administer_view(session, topic, view)
    token = ""
    for stored, where in _tokens(session.runtime).items():
        if isinstance(where, dict) and where.get("table") == topic and where.get("view") == view:
            token = str(stored)
            break
    return {
        "token": token or None,
        "url": _link_of(token) if token else None,
        # ⭐ THE SERVER'S OWN VIEW OF THE SPEC, and it exists to make ONE failure visible rather
        # than silent: the builder's `spec` prop comes from the client's `cleanDisplay`, so if the
        # client ever stops carrying `display.form`, the panel would render an empty form over a
        # stored one and the next save would erase real questions. Reporting what the SERVER holds
        # lets the panel say so instead ([[read-path-cannot-witness-write-path]]).
        "stored": ({"fields": len(spec.get("fields") or []),
                    "access": spec.get("access") or "public",
                    "title": str(spec.get("title") or "")[:120]} if spec else None),
    }


@router.post("/form-link")
def mint_form_link(body: dict, session: Session = Depends(require_session)):
    """Mint (or regenerate) this view's share link. `{topic, view, regenerate?}`.

    Idempotent without `regenerate`: opening the panel twice must not invalidate the link
    somebody already emailed. With it, the previous token is DROPPED in the same write — a
    "regenerate" that leaves the old link alive has revoked nothing.
    """
    topic, view = str((body or {}).get("topic") or ""), str((body or {}).get("view") or "")
    _v, spec = _may_administer_view(session, topic, view)
    if not spec.get("fields"):
        raise err(400, "empty_form", "add at least one question before sharing this form")
    fresh = secrets.token_urlsafe(TOKEN_BYTES)
    keep = not (body or {}).get("regenerate")

    minted = {"token": fresh}

    def _set(cur):
        cur = dict(cur or {})
        for stored, where in list(cur.items()):
            if isinstance(where, dict) and where.get("table") == topic and where.get("view") == view:
                if keep:
                    minted["token"] = str(stored)
                    return cur
                cur.pop(stored, None)          # regenerate: the old link dies in this same write
        cur[fresh] = {"table": topic, "view": view}
        return cur

    # `sync`: a share link the user is about to paste into an email must be durable before the
    # response says it exists. This is not a hot path.
    session.runtime.update(TOKENS_KEY, _set, flush="sync")
    return {"token": minted["token"], "url": _link_of(minted["token"])}


@router.delete("/form-link")
def revoke_form_link(topic: str, view: str, session: Session = Depends(require_session)):
    """Turn the link off. The form spec stays; only the door closes."""
    _may_administer_view(session, topic, view)

    def _drop(cur):
        cur = dict(cur or {})
        for stored, where in list(cur.items()):
            if isinstance(where, dict) and where.get("table") == topic and where.get("view") == view:
                cur.pop(stored, None)
        return cur

    session.runtime.update(TOKENS_KEY, _drop, flush="sync")
    return {"ok": True}


@router.get("/forms/{token}")
def get_form(token: str, request: Request):
    """Render-time payload for a public form. Rate-limited like the POST: an unauthenticated GET
    that walks every tenant's workspace is the cheapest way to make this process do work."""
    if not _rate_ok(_client_ip(request), time.time()):
        raise err(429, "too_many_requests", "too many requests — wait a moment and try again")
    found = _resolve(token)
    if not found:
        raise _refuse()
    rt, _slug, table_key, spec = found
    return _public_form(rt, table_key, spec)


async def _bounded_body(request: Request) -> dict:
    """The request body, READ WITH A BOUND — never `Body(...)`, never `await request.body()`.

    ⛔ THE FIRST VERSION OF THIS ROUTE TOOK `body: dict = Body(default=None)` AND ITS 16 KB CAP
    CAPPED NOTHING. FastAPI reads and JSON-parses the WHOLE body before the handler's first line
    runs, so a `content-length` check inside the handler is inspected after the allocation it
    claims to prevent — and `content-length` is caller-supplied anyway, so omitting it or sending
    chunked left no bound at all. On the one unauthenticated write path in the product.

    ⚠ It was GREEN, and that is the part worth remembering: the gate leg set the header itself,
    so it only ever exercised the honest path. A protection whose test supplies the very value it
    is protecting against is testing its own politeness.

    Streaming with a running total is the actual bound: the read STOPS at the ceiling rather than
    discovering afterwards that it should have.
    """
    size, chunks = 0, []
    async for chunk in request.stream():
        size += len(chunk)
        if size > MAX_BODY_BYTES:
            raise err(413, "body_too_large", "that submission is too large")
        chunks.append(chunk)
    import json
    try:
        parsed = json.loads(b"".join(chunks) or b"{}")
    except ValueError:
        raise err(400, "no_values", "that submission could not be read")
    return parsed if isinstance(parsed, dict) else {}


@router.post("/forms/{token}")
async def submit_form(token: str, request: Request):
    """One submission: validate, append the row, fire `form_submitted`.

    ⚠ THE ORDER IS LOAD-BEARING. The row is written FIRST and the trigger fired second, off the
    id the write returned — so an automation can never run against a record that does not exist,
    and a trigger that raises cannot cost the submitter the answer they just typed.
    """
    now = time.time()
    if not _rate_ok(_client_ip(request), now):
        raise err(429, "too_many_requests", "too many requests — wait a moment and try again")
    body = await _bounded_body(request)
    found = _resolve(token)
    if not found:
        raise _refuse()
    rt, _slug, table_key, spec = found

    values = (body or {}).get("values")
    if not isinstance(values, dict):
        raise err(400, "no_values", "that submission was empty")

    # ⛔ THE HONEYPOT ANSWERS 200 AND WRITES NOTHING. A 403 would tell the script it was
    # detected, and the next version of it simply stops filling the field. A success it can
    # never verify is the only answer that costs the operator nothing to give.
    if str(values.get(HONEYPOT_FIELD) or "").strip():
        return {"ok": True}

    form = _public_form(rt, table_key, spec)
    if not form["fields"]:
        raise _refuse()

    # ⭐ WAVE-29 R7 — the authorized-by-email door. Checked BEFORE `_clean_values` so an
    # uninvited caller learns nothing about the form's questions from the shape of its refusal,
    # and before the daily cap so strangers cannot spend a live form's allowance.
    # ⚠ NOT `_refuse()`: this one is deliberately a DIFFERENT answer. `_refuse` exists so a wrong
    # token cannot be told apart from a missing form; here the caller already holds a valid link
    # and the only useful thing we can say is "not with that address" — telling them nothing
    # would leave a colleague staring at a form that silently does nothing.
    submitter = str((body or {}).get("email") or "").strip().lower()[:254]
    if not _invited(spec, submitter):
        raise err(403, "not_invited",
                  "this form only accepts submissions from invited email addresses")

    clean, problem = _clean_values(form["fields"], values)
    if problem:
        raise err(400, "invalid_submission", problem)

    # ⚠ THE DAILY CAP IS SPENT ON A WRITE, NOT ON A REQUEST, and the order is the whole point.
    # Counted before validation — where it was first — 500 malformed POSTs take a LIVE form
    # offline for 24 hours, and one address can send them in about 17 minutes under the per-IP
    # window. The cap exists to bound what reaches the tenant's table; pacing abuse is the
    # window's job, and a refusal costs the attacker the same either way.
    if not _daily_ok(token, time.time()):
        raise err(429, "form_daily_cap",
                  "this form has reached today's submission limit — try again tomorrow")

    import core.user_tables as user_tables

    # ⚠ THE STAMP. `user_tables.add_row` filters values to the table's own field keys, so a
    # `created_by` pair passed in `values` would simply be dropped — the row itself has no meta
    # slot today. The `username` argument is the channel that DOES survive: it rides the
    # `record_created` row event, so every trigger and audit downstream sees `form:<prefix>`
    # rather than a blank actor. Booked as a dated amendment in the split doc; a persistent
    # per-row stamp needs `add_row` to grow one, which is SESSION A's file.
    # ⭐ WAVE-29 — an invited submitter is stamped BY ADDRESS rather than by token prefix: on an
    # email-restricted form the address is the most identifying thing we honestly have, and it is
    # what makes `record_created` legible in a run log ("form:ana@…" beats "form:Kx2p9Lm4").
    actor = f"form:{submitter or token[:8]}"[:120]
    row_id = user_tables.add_row(table_key, clean, username=actor, st=rt)
    if not row_id:
        raise err(409, "not_accepted", "that submission could not be saved — try again later")

    # W23-W7 — A's FROZEN signature, called verbatim (CP1-a). Failure here must not turn a
    # SAVED row into an error for the submitter: the answer is theirs and it is already stored.
    try:
        import automation_engine as engine
        engine.form_fired(rt, table_key, row_id, values=clean, form_token=token)
    except Exception:                                             # noqa: BLE001
        pass
    return {"ok": True}