"""Asana — Ahmed's project management (projects, tasks, due dates, assignees).

THE single source of truth for Asana in this repo. Two doors use this ONE module:
  • the hosted connector  — jarvis-tools-service/mcp_server.py  (@mcp.tool wrappers)
  • voice Jarvis          — voice/asana_control.py               (thin SDK-MCP wrapper
                            that imports THIS file; no REST logic is duplicated)
Fix a bug here and both doors are fixed.

Plain HTTPS REST against https://app.asana.com/api/1.0 with a Personal Access
Token — no OAuth, no SDK, stdlib urllib only (the Docker image and the voice
engine both depend on there being no new deps).

Env: ASANA_PAT       — Personal Access Token (asana.com → Settings → Apps →
                       Developer apps → Personal access tokens).
     ASANA_WORKSPACE — optional; workspace gid or name to pin. If unset we
                       resolve (and cache) the account's first workspace.

Results are read aloud by voice Jarvis → return short plain prose / simple
dashed lists. No markdown, no raw JSON (the only exception is `raw()`, the
escape hatch, which is for the model, not for speech). Nothing here ever raises:
every failure comes back as one human sentence.

API gotchas honoured here (learned the hard way / from the OpenAPI spec):
  • {"data": …} envelope BOTH ways — wrap request bodies, unwrap responses.
  • opt_fields or you get NOTHING but gid/resource_type/name. Every call sends it.
  • projects/tags/followers are CREATE-ONLY on a task — changing them needs
    addProject / removeProject / addTag / addFollowers. A "move" is
    addProject + removeProject (add alone leaves the task in BOTH projects).
  • Project create needs a TEAM when the workspace is an organization.
  • Archive (PUT {archived:true}) ≠ delete. There is no /archive endpoint.
  • /tasks/search is PREMIUM (402 on free) → transparent typeahead fallback.
  • next_page lives OUTSIDE data → _paged() follows it.
  • 429 carries Retry-After; rejected calls still burn quota → we sleep + retry once.

NOTE: changing MCP tools needs a Railway redeploy (push to origin/jarvis-v2).
"""
from __future__ import annotations

import json
import os
import time
import urllib.error
import urllib.parse
import urllib.request

_BASE = os.environ.get("ASANA_URL", "https://app.asana.com/api/1.0").rstrip("/")
_WS_CACHE: dict[str, str] = {}      # {"gid", "name", "is_org"} — resolved once
# name(lower) → gid for tasks WE just created. Asana's typeahead text index lags
# by minutes (measured), so "create a task … now comment on it" would otherwise
# fail to resolve the name. See _task_gid().
_RECENT: dict[str, str] = {}

_UNCONFIGURED = "Asana isn't configured (no ASANA_PAT)."

# opt_fields sets — Asana returns gid/resource_type/name and NOTHING else unless asked
_F_TASK = "name,due_on,due_at,completed,assignee.name,projects.name"
_F_TASK_FULL = (
    "name,notes,completed,completed_at,due_on,due_at,start_on,assignee.name,"
    "projects.name,parent.name,tags.name,followers.name,num_subtasks,"
    "memberships.project.name,memberships.section.name,custom_fields.name,"
    "custom_fields.display_value,created_at,modified_at,permalink_url")
_F_PROJECT = "name,archived,due_on,owner.name,current_status.title,current_status.color"


def _pat() -> str:
    return os.environ.get("ASANA_PAT", "").strip()


def configured() -> bool:
    return bool(_pat())


class AsanaError(RuntimeError):
    """A short, speakable failure reason. `.code` = HTTP status (0 if not HTTP)."""

    def __init__(self, msg: str, code: int = 0):
        super().__init__(msg)
        self.code = code


# ---------------------------------------------------------------------------
# Low-level REST
# ---------------------------------------------------------------------------

def _api(method: str, path: str, body: dict | None = None,
         params: dict | None = None, timeout: int = 30, raw_envelope: bool = False,
         _retried: bool = False):
    """One Asana REST call. Returns the unwrapped `data` payload (or the whole
    envelope when raw_envelope=True, which is how _paged() sees `next_page`).
    Raises AsanaError with a short human reason — callers turn that into a sentence.
    """
    if not path.startswith("/"):
        path = "/" + path
    url = f"{_BASE}{path}"
    if params:
        clean = {k: v for k, v in params.items() if v not in (None, "")}
        if clean:
            url += ("&" if "?" in path else "?") + urllib.parse.urlencode(clean)
    data = json.dumps({"data": body}).encode() if body is not None else None
    req = urllib.request.Request(url, data=data, method=method.upper())
    req.add_header("Authorization", f"Bearer {_pat()}")
    req.add_header("Accept", "application/json")
    if data is not None:
        req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            raw = r.read()
    except urllib.error.HTTPError as e:
        detail = ""
        try:
            err = json.loads(e.read() or b"{}")
            errs = err.get("errors") or []
            detail = "; ".join(str(x.get("message", "")) for x in errs)[:200]
        except Exception:  # noqa: BLE001
            pass
        if e.code == 429:
            wait = 0.0
            try:
                wait = float(e.headers.get("Retry-After") or 0)
            except Exception:  # noqa: BLE001
                wait = 0.0
            if not _retried and wait <= 20:
                time.sleep(max(wait, 1.0))
                return _api(method, path, body, params, timeout, raw_envelope,
                            _retried=True)
            raise AsanaError(
                "Asana is rate-limiting us (429) — try again in "
                f"{int(wait) or 60} seconds.", 429)
        if e.code == 401:
            raise AsanaError("Asana rejected the token (401) — check ASANA_PAT.", 401)
        if e.code == 402:
            raise AsanaError(
                "that needs a paid Asana plan (402) — it's not on this free "
                "workspace.", 402)
        if e.code == 403:
            why = f": {detail}" if detail else ""
            raise AsanaError(f"Asana says no permission for that (403){why}.", 403)
        if e.code == 404:
            raise AsanaError("Asana couldn't find that item (404).", 404)
        raise AsanaError(f"Asana error {e.code}: {detail or 'request failed'}", e.code)
    except urllib.error.URLError as e:
        raise AsanaError(f"couldn't reach Asana: {str(e.reason)[:120]}")
    payload = json.loads(raw) if raw else {}
    if raw_envelope:
        return payload
    return payload.get("data", payload)


def _paged(path: str, params: dict, cap: int = 300) -> list[dict]:
    """GET a collection, following `next_page.offset` (which lives OUTSIDE data)."""
    out: list[dict] = []
    p = dict(params)
    p.setdefault("limit", 100)
    for _ in range(10):                       # hard stop; cap ends it first normally
        env = _api("GET", path, params=p, raw_envelope=True) or {}
        rows = env.get("data") or []
        out += rows
        nxt = env.get("next_page") or {}
        offset = nxt.get("offset") if isinstance(nxt, dict) else None
        if not offset or len(out) >= cap:
            break
        p["offset"] = offset
    return out[:cap]


def _fail(tool: str, e: Exception) -> str:
    return f"{tool} failed: {str(e)[:200]}"


def _remember(name: str, gid: str) -> None:
    """Cache a task we just created (see _task_gid). Bounded — these processes
    (the voice engine, the Railway service) run for days."""
    if not name or not gid:
        return
    if len(_RECENT) >= 200:
        _RECENT.pop(next(iter(_RECENT)), None)      # drop the oldest
    _RECENT[name.strip().lower()] = str(gid)


def _list(val) -> list[str]:
    """'a, b' | ['a','b'] → ['a','b']."""
    if not val:
        return []
    if isinstance(val, (list, tuple)):
        items = list(val)
    else:
        items = str(val).split(",")
    return [str(x).strip() for x in items if str(x).strip()]


def _truthy(v) -> bool:
    if isinstance(v, str):
        return v.strip().lower() in ("1", "true", "yes", "y", "done", "complete")
    return bool(v)


# ---------------------------------------------------------------------------
# Resolvers — humans say names, Asana wants gids
# ---------------------------------------------------------------------------

def _ws_info() -> dict:
    """{'gid','name','is_org'} for the workspace we operate in (cached)."""
    if _WS_CACHE.get("gid"):
        return dict(_WS_CACHE)
    want = os.environ.get("ASANA_WORKSPACE", "").strip()
    spaces = _api("GET", "/workspaces",
                  params={"opt_fields": "name,is_organization", "limit": 100}) or []
    if not spaces:
        raise AsanaError("that Asana account has no workspaces.")
    pick = None
    if want.isdigit():
        pick = next((w for w in spaces if str(w.get("gid")) == want), None)
    elif want:
        low = want.lower()
        pick = next((w for w in spaces
                     if str(w.get("name", "")).lower() == low), None)
        pick = pick or next((w for w in spaces
                             if low in str(w.get("name", "")).lower()), None)
    pick = pick or spaces[0]
    _WS_CACHE["gid"] = str(pick.get("gid"))
    _WS_CACHE["name"] = str(pick.get("name", ""))
    _WS_CACHE["is_org"] = "1" if pick.get("is_organization") else ""
    return dict(_WS_CACHE)


def _workspace() -> str:
    """Gid of the workspace we operate in (cached)."""
    return _ws_info()["gid"]


def _match(items: list[dict], ref: str) -> dict | None:
    """Exact (case-insensitive) name match, then a contains-match."""
    low = ref.lower()
    hit = next((i for i in items if str(i.get("name", "")).lower() == low), None)
    return hit or next((i for i in items
                        if low in str(i.get("name", "")).lower()), None)


def _project_gid(ref: str, archived: bool | None = False) -> str | None:
    """Resolve a project name (or gid) → gid. None if not found."""
    ref = str(ref or "").strip()
    if not ref:
        return None
    if ref.isdigit():
        return ref
    params: dict = {"workspace": _workspace(), "opt_fields": "name", "limit": 100}
    if archived is not None:
        params["archived"] = "true" if archived else "false"
    projs = _paged("/projects", params)
    hit = _match(projs, ref)
    if not hit and archived is False:          # maybe it's archived — look wider
        projs = _paged("/projects", {"workspace": _workspace(),
                                     "opt_fields": "name", "limit": 100})
        hit = _match(projs, ref)
    return str(hit["gid"]) if hit else None


def _task_gid(ref: str) -> str | None:
    """Resolve a task name (or gid) → gid. None if not found.

    Three tiers, because Asana's typeahead TEXT index lags several minutes
    (verified against the live API): a task created 2 seconds ago is invisible to
    `typeahead?query=…`, which would break "create it, now comment on it".
      1. gids and tasks we created in this process (instant),
      2. typeahead by name (the normal path for anything older),
      3. typeahead with an EMPTY query = the workspace's most-RECENT tasks, which
         is not text-indexed and therefore does see brand-new tasks.
    """
    ref = str(ref or "").strip()
    if not ref:
        return None
    if ref.isdigit():
        return ref
    cached = _RECENT.get(ref.lower())
    if cached:
        return cached
    ws = _workspace()
    hits = _api("GET", f"/workspaces/{ws}/typeahead", params={
        "resource_type": "task", "query": ref, "count": 20,
        "opt_fields": "name"}) or []
    hit = _match(hits, ref)
    if not hit:
        fresh = _api("GET", f"/workspaces/{ws}/typeahead", params={
            "resource_type": "task", "query": "", "count": 100,
            "opt_fields": "name"}) or []
        hit = _match(fresh, ref)
    hit = hit or (hits[0] if hits else None)
    if hit:
        _remember(ref, str(hit["gid"]))
        return str(hit["gid"])
    return None


def _section_gid(project_gid: str, ref: str) -> str | None:
    """Resolve a section (board column) name → gid inside one project."""
    ref = str(ref or "").strip()
    if not ref or not project_gid:
        return None
    if ref.isdigit():
        return ref
    secs = _api("GET", f"/projects/{project_gid}/sections",
                params={"opt_fields": "name", "limit": 100}) or []
    hit = _match(secs, ref)
    return str(hit["gid"]) if hit else None


def _team_gid(ref: str) -> str | None:
    ref = str(ref or "").strip()
    if not ref:
        return None
    if ref.isdigit():
        return ref
    teams = _api("GET", f"/workspaces/{_workspace()}/teams",
                 params={"opt_fields": "name", "limit": 100}) or []
    hit = _match(teams, ref)
    return str(hit["gid"]) if hit else None


def _tag_gid(ref: str, create: bool = False) -> str | None:
    ref = str(ref or "").strip()
    if not ref:
        return None
    if ref.isdigit():
        return ref
    tags = _paged(f"/workspaces/{_workspace()}/tags", {"opt_fields": "name"})
    hit = _match(tags, ref)
    if hit:
        return str(hit["gid"])
    if create:
        made = _api("POST", f"/workspaces/{_workspace()}/tags",
                    {"name": ref}, params={"opt_fields": "name"}) or {}
        return str(made.get("gid")) if made.get("gid") else None
    return None


def _me_gid() -> str:
    """My own user gid (cached) — needed to filter 'assigned to me' client-side."""
    if _WS_CACHE.get("me"):
        return _WS_CACHE["me"]
    me = _api("GET", "/users/me", params={"opt_fields": "name"}) or {}
    _WS_CACHE["me"] = str(me.get("gid", ""))
    _WS_CACHE["me_name"] = str(me.get("name", ""))
    return _WS_CACHE["me"]


def _user(ref: str) -> str:
    """A user reference Asana accepts anywhere a user is expected.

    Asana takes a gid, an EMAIL, or the literal 'me' interchangeably — so there
    is no email→gid lookup to build. Only a spoken NAME ('Sara') needs resolving.
    """
    ref = str(ref or "").strip()
    if not ref or ref.lower() in ("me", "myself", "i", "ahmed", "mine"):
        return "me"
    if ref.isdigit() or "@" in ref:
        return ref
    try:
        hits = _api("GET", f"/workspaces/{_workspace()}/typeahead", params={
            "resource_type": "user", "query": ref, "count": 10,
            "opt_fields": "name,email"}) or []
        hit = _match(hits, ref)
        if hit:
            return str(hit["gid"])
    except Exception:  # noqa: BLE001 — fall through, let Asana judge the string
        pass
    return ref


# ---------------------------------------------------------------------------
# Formatting — short, speakable lines (never markdown / JSON)
# ---------------------------------------------------------------------------

def _due(t: dict) -> str:
    d = t.get("due_on") or (t.get("due_at") or "")[:10]
    return f", due {d}" if d else ""


def _who(t: dict) -> str:
    a = t.get("assignee") or {}
    name = a.get("name") if isinstance(a, dict) else None
    return f" [{name}]" if name else ""


def _task_line(t: dict) -> str:
    done = " (done)" if t.get("completed") else ""
    return f"- {t.get('name', '(untitled)')}{_due(t)}{_who(t)}{done}"


def _names(seq) -> str:
    return ", ".join(str((x or {}).get("name", "")) for x in (seq or [])
                     if isinstance(x, dict) and x.get("name"))


# ---------------------------------------------------------------------------
# WORKSPACES / PROJECTS
# ---------------------------------------------------------------------------

def workspaces() -> str:
    if not configured():
        return _UNCONFIGURED
    try:
        spaces = _api("GET", "/workspaces",
                      params={"opt_fields": "name,is_organization"}) or []
        if not spaces:
            return "That Asana account has no workspaces."
        default = _workspace()
        lines = []
        for w in spaces:
            kind = "organization" if w.get("is_organization") else "personal workspace"
            tag = " (default)" if str(w.get("gid")) == default else ""
            lines.append(f"- {w.get('name')} — {kind}{tag}")
        return f"{len(spaces)} Asana workspace(s):\n" + "\n".join(lines)
    except Exception as e:  # noqa: BLE001
        return _fail("asana_workspaces", e)


def projects(query: str = "", archived: bool = False) -> str:
    if not configured():
        return _UNCONFIGURED
    try:
        projs = _paged("/projects", {
            "workspace": _workspace(),
            "archived": "true" if _truthy(archived) else "false",
            "opt_fields": _F_PROJECT})
        q = str(query or "").strip().lower()
        if q:
            projs = [p for p in projs if q in str(p.get("name", "")).lower()]
        if not projs:
            return (f"No Asana projects matching {query!r}." if q
                    else "No projects in that Asana workspace yet.")
        lines = []
        for p in projs:
            st = p.get("current_status") or {}
            title = st.get("title") if isinstance(st, dict) else None
            owner = (p.get("owner") or {}).get("name") if isinstance(
                p.get("owner"), dict) else None
            lines.append(f"- {p.get('name')}" + _due(p)
                         + (f" [{owner}]" if owner else "")
                         + (f" — {title}" if title else ""))
        label = "archived Asana project(s)" if _truthy(archived) \
            else "Asana project(s)"
        return f"{len(lines)} {label}:\n" + "\n".join(lines)
    except Exception as e:  # noqa: BLE001
        return _fail("asana_projects", e)


def project_create(name: str, team: str = "", notes: str = "", due_on: str = "",
                   privacy: str = "") -> str:
    """Create a project. In an ORGANIZATION a team is required (Asana rule);
    in a personal workspace it isn't."""
    if not configured():
        return _UNCONFIGURED
    name = str(name or "").strip()
    if not name:
        return "asana_project_create failed: need a project name."
    try:
        ws = _ws_info()
        body: dict = {"name": name}
        if notes:
            body["notes"] = str(notes)
        if due_on:
            body["due_on"] = str(due_on).strip()[:10]
        if privacy:
            body["privacy_setting"] = str(privacy).strip()
        tid = _team_gid(team) if team else None
        if team and not tid:
            return (f"No Asana team called {team!r} — say which team it belongs "
                    "to (asana_api GET /workspaces/{gid}/teams lists them).")
        if tid:
            out = _api("POST", f"/teams/{tid}/projects", body,
                       params={"opt_fields": "name,permalink_url"}) or {}
        else:
            if ws.get("is_org"):
                teams = _api("GET", f"/workspaces/{ws['gid']}/teams",
                             params={"opt_fields": "name", "limit": 100}) or []
                have = ", ".join(str(t.get("name")) for t in teams[:10])
                return ("That Asana workspace is an organization, so a new project "
                        "must belong to a team — tell me which team"
                        + (f" (available: {have})." if have else "."))
            body["workspace"] = ws["gid"]
            out = _api("POST", "/projects", body,
                       params={"opt_fields": "name,permalink_url"}) or {}
        where = f" in team {team}" if tid else ""
        return f"Created Asana project {out.get('name', name)!r}{where} (id {out.get('gid')})."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_project_create", e)


def project_update(project: str, name: str = "", notes: str = "", due_on: str = "",
                   owner: str = "", archive=None) -> str:
    """Rename / re-note / re-date / re-own a project, or ARCHIVE it
    (archive=True → PUT {archived:true}; reversible — this is NOT a delete)."""
    if not configured():
        return _UNCONFIGURED
    try:
        pid = _project_gid(project, archived=None)
        if not pid:
            return f"No Asana project matching {project!r}."
        body: dict = {}
        if name:
            body["name"] = str(name).strip()
        if notes:
            body["notes"] = str(notes)
        if due_on:
            body["due_on"] = str(due_on).strip()[:10]
        if owner:
            body["owner"] = _user(owner)
        if archive is not None:
            body["archived"] = _truthy(archive)
        if not body:
            return ("Nothing to change — give a name, notes, due_on, owner, or "
                    "archive=true.")
        out = _api("PUT", f"/projects/{pid}", body,
                   params={"opt_fields": "name,archived"}) or {}
        pname = out.get("name", project)
        if "archived" in body:
            return (f"{'Archived' if body['archived'] else 'Un-archived'} the Asana "
                    f"project {pname!r}. (Archive is reversible — nothing deleted.)")
        return f"Updated Asana project {pname!r} ({', '.join(body.keys())})."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_project_update", e)


def project_members(project: str, add: str = "", remove: str = "",
                    access_level: str = "") -> str:
    """List / add / remove the people on a project ("invite people to a project").
    add / remove take gids, EMAILS, or 'me' (comma-separated)."""
    if not configured():
        return _UNCONFIGURED
    try:
        pid = _project_gid(project, archived=None)
        if not pid:
            return f"No Asana project matching {project!r}."
        adds = [_user(x) for x in _list(add)]
        rems = [_user(x) for x in _list(remove)]
        said: list[str] = []
        if adds:
            if access_level:
                for m in adds:
                    _api("POST", "/memberships",
                         {"parent": pid, "member": m,
                          "access_level": str(access_level).strip()})
                said.append(f"added {len(adds)} as {access_level}")
            else:
                _api("POST", f"/projects/{pid}/addMembers", {"members": adds})
                said.append(f"added {', '.join(adds)}")
        if rems:
            _api("POST", f"/projects/{pid}/removeMembers", {"members": rems})
            said.append(f"removed {', '.join(rems)}")
        rows = _api("GET", f"/projects/{pid}/project_memberships", params={
            "opt_fields": "user.name,user.email,access_level", "limit": 100}) or []
        who = []
        for r in rows:
            u = r.get("user") or {}
            lvl = r.get("access_level")
            who.append(f"- {u.get('name') or u.get('email') or '?'}"
                       + (f" ({lvl})" if lvl else ""))
        head = ("Project " + str(project) + ": " + "; ".join(said) + ".") if said \
            else f"{len(who)} member(s) on {project}:"
        return head + ("\n" + "\n".join(who) if who else "")
    except Exception as e:  # noqa: BLE001
        return _fail("asana_project_members", e)


def project_status(project: str, text: str, status_type: str = "on_track",
                   title: str = "") -> str:
    """Post a project status update (on_track / at_risk / off_track / on_hold /
    complete)."""
    if not configured():
        return _UNCONFIGURED
    text = str(text or "").strip()
    if not text:
        return "asana_project_status failed: need the status text."
    try:
        pid = _project_gid(project, archived=None)
        if not pid:
            return f"No Asana project matching {project!r}."
        st = str(status_type or "on_track").strip().lower().replace(" ", "_")
        ok = ("on_track", "at_risk", "off_track", "on_hold", "complete")
        if st not in ok:
            return f"status_type must be one of: {', '.join(ok)}."
        body: dict = {"parent": pid, "text": text, "status_type": st}
        if title:
            body["title"] = str(title).strip()
        _api("POST", "/status_updates", body, params={"opt_fields": "title"})
        return f"Posted a status update on {project} — {st.replace('_', ' ')}."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_project_status", e)


# ---------------------------------------------------------------------------
# TASKS
# ---------------------------------------------------------------------------

def task(task: str) -> str:  # noqa: A002 — the tool's arg really is called task
    """Everything about ONE task: notes, assignee, due, section, tags, subtasks,
    custom fields and the last few comments."""
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        try:
            t = _api("GET", f"/tasks/{tid}", params={
                "opt_fields": _F_TASK_FULL + ",dependencies.name"}) or {}
        except AsanaError:                      # dependencies are paid-tier
            t = _api("GET", f"/tasks/{tid}", params={"opt_fields": _F_TASK_FULL}) or {}
        lines = [f"{t.get('name', '(untitled)')}"
                 + (" — DONE" if t.get("completed") else "")]
        who = (t.get("assignee") or {}).get("name")
        if who:
            lines.append(f"Assignee: {who}")
        if t.get("due_on") or t.get("due_at"):
            lines.append(f"Due: {t.get('due_on') or (t.get('due_at') or '')[:16]}")
        if t.get("start_on"):
            lines.append(f"Starts: {t['start_on']}")
        proj = _names(t.get("projects"))
        if proj:
            lines.append(f"Project: {proj}")
        for m in (t.get("memberships") or []):
            sec = (m.get("section") or {}).get("name")
            pr = (m.get("project") or {}).get("name")
            if sec:
                lines.append(f"Section: {sec}" + (f" (in {pr})" if pr else ""))
        par = (t.get("parent") or {}).get("name")
        if par:
            lines.append(f"Subtask of: {par}")
        tags = _names(t.get("tags"))
        if tags:
            lines.append(f"Tags: {tags}")
        foll = _names(t.get("followers"))
        if foll:
            lines.append(f"Followers: {foll}")
        deps = _names(t.get("dependencies"))
        if deps:
            lines.append(f"Waiting on: {deps}")
        for cf in (t.get("custom_fields") or []):
            val = cf.get("display_value")
            if val:
                lines.append(f"{cf.get('name')}: {val}")
        if t.get("notes"):
            body = str(t["notes"]).strip().replace("\n", " ")
            lines.append("Notes: " + (body[:400] + ("…" if len(body) > 400 else "")))
        subs = _api("GET", f"/tasks/{tid}/subtasks", params={
            "opt_fields": "name,completed", "limit": 50}) or []
        if subs:
            lines.append(f"Subtasks ({len(subs)}):")
            lines += [f"  - {s.get('name')}" + (" (done)" if s.get("completed") else "")
                      for s in subs[:10]]
        stories = _api("GET", f"/tasks/{tid}/stories", params={
            "opt_fields": "text,created_at,created_by.name,resource_subtype",
            "limit": 100}) or []
        cmts = [s for s in stories if s.get("resource_subtype") == "comment_added"]
        if cmts:
            lines.append(f"Comments ({len(cmts)}), latest:")
            for s in cmts[-3:]:
                by = (s.get("created_by") or {}).get("name", "someone")
                when = str(s.get("created_at", ""))[:10]
                txt = str(s.get("text", "")).strip().replace("\n", " ")[:160]
                lines.append(f"  - {by} ({when}): {txt}")
        return "\n".join(lines)
    except Exception as e:  # noqa: BLE001
        return _fail("asana_task", e)


def tasks(project: str = "", mine: bool = True, section: str = "",
          completed: bool = False, limit: int = 50) -> str:
    if not configured():
        return _UNCONFIGURED
    try:
        try:
            limit = max(1, min(int(limit or 50), 100))
        except Exception:  # noqa: BLE001
            limit = 50
        want_done = _truthy(completed)
        pid = None
        if project:
            pid = _project_gid(project, archived=None)
            if not pid:
                return f"No Asana project matching {project!r}."
        if section:
            if not pid:
                return "Give a project too — a section only exists inside a project."
            sid = _section_gid(pid, section)
            if not sid:
                return f"No section called {section!r} in {project}."
            params = {"opt_fields": _F_TASK, "limit": 100}
            if not want_done:
                params["completed_since"] = "now"
            rows = _paged(f"/sections/{sid}/tasks", params, cap=limit * 3)
            label = f"task(s) in {section} ({project})"
        elif pid:
            params = {"project": pid, "opt_fields": _F_TASK, "limit": 100}
            if not want_done:
                params["completed_since"] = "now"
            rows = _paged("/tasks", params, cap=limit * 3)
            label = f"task(s) in {project}"
        elif _truthy(mine):
            params = {"assignee": "me", "workspace": _workspace(),
                      "opt_fields": _F_TASK, "limit": 100}
            if not want_done:
                params["completed_since"] = "now"
            rows = _paged("/tasks", params, cap=limit * 3)
            label = "task(s) assigned to you"
        else:
            return "Give a project name, or set mine=true for your own tasks."
        rows = [t for t in rows if bool(t.get("completed")) == want_done]
        state = "completed" if want_done else "open"
        if not rows:
            return f"No {state} {label}."
        lines = [_task_line(t) for t in rows[:limit]]
        more = f"\n…and {len(rows) - limit} more." if len(rows) > limit else ""
        return f"{len(rows)} {state} {label}:\n" + "\n".join(lines) + more
    except Exception as e:  # noqa: BLE001
        return _fail("asana_tasks", e)


def task_create(name: str, notes: str = "", project: str = "", assignee: str = "me",
                due_on: str = "", section: str = "", parent: str = "") -> str:
    if not configured():
        return _UNCONFIGURED
    name = str(name or "").strip()
    if not name:
        return "asana_task_create failed: need a task name."
    try:
        body: dict = {"name": name}
        if notes:
            body["notes"] = str(notes)
        if due_on:
            body["due_on"] = str(due_on).strip()[:10]
        if assignee:
            body["assignee"] = _user(assignee)
        pid = None
        if project:
            pid = _project_gid(project, archived=None)
            if not pid:
                return f"No Asana project matching {project!r} — task not created."
        par = _task_gid(parent) if parent else None
        if parent and not par:
            return f"No Asana task matching {parent!r} — subtask not created."
        if par:
            # Subtasks do NOT inherit the parent's project — add it explicitly below.
            out = _api("POST", f"/tasks/{par}/subtasks", body,
                       params={"opt_fields": "name"}) or {}
            tid = str(out.get("gid"))
            if pid:
                _api("POST", f"/tasks/{tid}/addProject", {"project": pid})
        else:
            if pid:
                body["projects"] = [pid]      # create-only field — fine here
            else:
                body["workspace"] = _workspace()
            out = _api("POST", "/tasks", body, params={"opt_fields": "name"}) or {}
            tid = str(out.get("gid"))
        _remember(name, tid)            # typeahead won't see it for minutes
        if section and pid:
            sid = _section_gid(pid, section)
            if sid:
                _api("POST", f"/sections/{sid}/addTask", {"task": tid})
            else:
                return (f"Created {name!r} in {project} (id {tid}) — but there's no "
                        f"section called {section!r}, so it's in the default one.")
        where = f" in {project}" if project else ""
        under = f" under {parent}" if parent else ""
        col = f" ({section})" if section and pid else ""
        when = f", due {body['due_on']}" if body.get("due_on") else ""
        return f"Created Asana task {name!r}{where}{col}{under}{when} (id {tid})."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_task_create", e)


def task_update(task: str, name: str = "", notes: str = "", due_on: str = "",  # noqa: A002
                start_on: str = "", assignee: str = "", complete=None) -> str:
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        body: dict = {}
        if name:
            body["name"] = str(name).strip()
        if notes:
            body["notes"] = str(notes)
        if due_on:
            body["due_on"] = str(due_on).strip()[:10]
        if start_on:
            body["start_on"] = str(start_on).strip()[:10]
            # Asana rejects a start date on a task with no due date (400: "You must
            # provide `due_on` or `due_at` when setting `start_on`"). Carry the
            # task's existing due date through, or say what's missing.
            if "due_on" not in body:
                cur = _api("GET", f"/tasks/{tid}",
                           params={"opt_fields": "due_on,due_at"}) or {}
                have = cur.get("due_on") or (cur.get("due_at") or "")[:10]
                if not have:
                    return ("Asana needs a due date before it will take a start date "
                            "— give me due_on as well.")
                body["due_on"] = have
        if assignee:
            body["assignee"] = _user(assignee)
        if complete is not None:
            body["completed"] = _truthy(complete)
        if not body:
            return ("Nothing to update — give a name, notes, due_on, start_on, "
                    "assignee, or complete.")
        out = _api("PUT", f"/tasks/{tid}", body, params={"opt_fields": "name"}) or {}
        if body.get("name"):                    # renamed — keep the cache truthful
            _RECENT.pop(str(task).strip().lower(), None)
            _remember(body["name"], tid)
        return f"Updated Asana task {out.get('name', task)!r} ({', '.join(body.keys())})."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_task_update", e)


def task_move(task: str, project: str = "", section: str = "",  # noqa: A002
              remove_from: str = "") -> str:
    """Move a task. A real move: adding it to a project REMOVES it from the ones it
    was in (Asana's addProject alone would leave it in both)."""
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        if not project and not section and not remove_from:
            return "Say where to move it — a project and/or a section."
        cur = _api("GET", f"/tasks/{tid}",
                   params={"opt_fields": "name,projects.name"}) or {}
        tname = cur.get("name", task)
        was = [p for p in (cur.get("projects") or []) if p.get("gid")]
        said: list[str] = []
        pid = None
        if project:
            pid = _project_gid(project, archived=None)
            if not pid:
                return f"No Asana project matching {project!r} — nothing moved."
            body: dict = {"project": pid}
            if section:
                sid = _section_gid(pid, section)
                if sid:
                    body["section"] = sid
                else:
                    return f"No section called {section!r} in {project} — nothing moved."
            _api("POST", f"/tasks/{tid}/addProject", body)
            said.append(f"added to {project}" + (f" / {section}" if section else ""))
        elif section:
            # Same project, different column.
            if not was:
                return "That task isn't in a project, so it has no sections."
            host = was[0]
            sid = _section_gid(str(host["gid"]), section)
            if not sid:
                return f"No section called {section!r} in {host.get('name')}."
            _api("POST", f"/sections/{sid}/addTask", {"task": tid})
            said.append(f"moved to {section} in {host.get('name')}")
        # Removals — explicit one, else every OTHER project (that's what "move" means)
        drop: list[dict] = []
        if remove_from:
            rid = _project_gid(remove_from, archived=None)
            if not rid:
                return f"No Asana project matching {remove_from!r}."
            drop = [{"gid": rid, "name": remove_from}]
        elif pid:
            drop = [p for p in was if str(p.get("gid")) != pid]
        for p in drop:
            _api("POST", f"/tasks/{tid}/removeProject", {"project": str(p["gid"])})
        if drop:
            said.append("removed from " + ", ".join(str(p.get("name") or p["gid"])
                                                    for p in drop))
        return f"{tname}: " + "; ".join(said) + "."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_task_move", e)


def task_complete(task: str) -> str:  # noqa: A002
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        out = _api("PUT", f"/tasks/{tid}", {"completed": True},
                   params={"opt_fields": "name"}) or {}
        return f"Marked {out.get('name', task)!r} complete in Asana."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_task_complete", e)


def task_delete(task: str) -> str:  # noqa: A002
    """Delete a task — it goes to Asana's Deleted Items and is recoverable for 30 days."""
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        got = _api("GET", f"/tasks/{tid}", params={"opt_fields": "name"}) or {}
        tname = got.get("name", task)
        _api("DELETE", f"/tasks/{tid}")
        for k in (str(task).strip().lower(), str(tname).strip().lower()):
            _RECENT.pop(k, None)
        return (f"Deleted the Asana task {tname!r}. It's in Deleted Items and can be "
                "restored for 30 days.")
    except Exception as e:  # noqa: BLE001
        return _fail("asana_task_delete", e)


def task_people(task: str, assignee: str = "", add_followers: str = "",  # noqa: A002
                remove_followers: str = "", add_tag: str = "",
                remove_tag: str = "") -> str:
    """Who and what is on a task: assignee, followers, tags. No args = just report."""
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        said: list[str] = []
        if assignee:
            _api("PUT", f"/tasks/{tid}", {"assignee": _user(assignee)})
            said.append(f"assigned to {assignee}")
        adds = [_user(x) for x in _list(add_followers)]
        if adds:
            _api("POST", f"/tasks/{tid}/addFollowers", {"followers": adds})
            said.append(f"following: +{', '.join(adds)}")
        rems = [_user(x) for x in _list(remove_followers)]
        if rems:
            _api("POST", f"/tasks/{tid}/removeFollowers", {"followers": rems})
            said.append(f"un-followed: {', '.join(rems)}")
        for name in _list(add_tag):
            gid = _tag_gid(name, create=True)   # make the tag if it's new
            if not gid:
                return f"Couldn't find or create the Asana tag {name!r}."
            _api("POST", f"/tasks/{tid}/addTag", {"tag": gid})
            said.append(f"tagged {name}")
        for name in _list(remove_tag):
            gid = _tag_gid(name)
            if gid:
                _api("POST", f"/tasks/{tid}/removeTag", {"tag": gid})
                said.append(f"untagged {name}")
        t = _api("GET", f"/tasks/{tid}", params={
            "opt_fields": "name,assignee.name,followers.name,tags.name"}) or {}
        who = (t.get("assignee") or {}).get("name")
        bits = [f"assignee {who}" if who else "unassigned"]
        foll = _names(t.get("followers"))
        if foll:
            bits.append(f"followers {foll}")
        tags = _names(t.get("tags"))
        if tags:
            bits.append(f"tags {tags}")
        head = ("Done — " + "; ".join(said) + ". ") if said else ""
        return head + f"{t.get('name', task)}: " + " · ".join(bits)
    except Exception as e:  # noqa: BLE001
        return _fail("asana_task_people", e)


def subtasks(task: str, add: str = "") -> str:  # noqa: A002
    """List a task's subtasks, or create them (add = comma-separated names)."""
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        made: list[str] = []
        for nm in _list(add):
            sub = _api("POST", f"/tasks/{tid}/subtasks", {"name": nm},
                       params={"opt_fields": "name"}) or {}
            _remember(nm, str(sub.get("gid") or ""))
            made.append(nm)
        subs = _api("GET", f"/tasks/{tid}/subtasks", params={
            "opt_fields": "name,completed,due_on,assignee.name", "limit": 100}) or []
        head = (f"Added {len(made)} subtask(s) to {task}: {', '.join(made)}.\n"
                if made else "")
        if not subs:
            return head or f"{task} has no subtasks."
        lines = [_task_line(s) for s in subs[:40]]
        return head + f"{len(subs)} subtask(s) on {task}:\n" + "\n".join(lines)
    except Exception as e:  # noqa: BLE001
        return _fail("asana_subtasks", e)


# ---------------------------------------------------------------------------
# COMMENTS
# ---------------------------------------------------------------------------

def comment(task: str, text: str) -> str:  # noqa: A002
    if not configured():
        return _UNCONFIGURED
    text = str(text or "").strip()
    if not text:
        return "asana_comment failed: need something to say."
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        _api("POST", f"/tasks/{tid}/stories", {"text": text})
        return f"Commented on Asana task {task!r}."
    except Exception as e:  # noqa: BLE001
        return _fail("asana_comment", e)


def comments(task: str, limit: int = 10) -> str:  # noqa: A002
    """Read a task's comment thread (system events like 'X assigned this' are skipped)."""
    if not configured():
        return _UNCONFIGURED
    try:
        tid = _task_gid(task)
        if not tid:
            return f"No Asana task matching {task!r}."
        try:
            limit = max(1, min(int(limit or 10), 50))
        except Exception:  # noqa: BLE001
            limit = 10
        stories = _paged(f"/tasks/{tid}/stories", {
            "opt_fields": "text,created_at,created_by.name,resource_subtype,is_pinned"})
        cmts = [s for s in stories if s.get("resource_subtype") == "comment_added"]
        if not cmts:
            return f"No comments on {task}."
        lines = []
        for s in cmts[-limit:]:
            by = (s.get("created_by") or {}).get("name", "someone")
            when = str(s.get("created_at", ""))[:10]
            txt = str(s.get("text", "")).strip().replace("\n", " ")
            pin = " (pinned)" if s.get("is_pinned") else ""
            lines.append(f"- {by}, {when}{pin}: {txt[:300]}")
        head = f"{len(cmts)} comment(s) on {task}"
        head += f", latest {len(lines)}:" if len(cmts) > len(lines) else ":"
        return head + "\n" + "\n".join(lines)
    except Exception as e:  # noqa: BLE001
        return _fail("asana_comments", e)


# ---------------------------------------------------------------------------
# SEARCH — premium /tasks/search, transparent typeahead fallback on 402
# ---------------------------------------------------------------------------

def _search_fallback(query: str, pid: str | None, who: str, due_before: str,
                     due_after: str, want_done) -> str:
    """Free-tier path: typeahead (name-match only) or a project listing, then
    filter client-side."""
    q = str(query or "").strip()
    ws = _workspace()
    rows: list[dict] = []
    if pid:
        params = {"project": pid, "opt_fields": _F_TASK, "limit": 100}
        if want_done is False:
            params["completed_since"] = "now"
        rows = _paged("/tasks", params)
        if q:
            rows = [t for t in rows if q.lower() in str(t.get("name", "")).lower()]
    else:
        rows = _api("GET", f"/workspaces/{ws}/typeahead", params={
            "resource_type": "task", "query": q, "count": 100,
            "opt_fields": _F_TASK}) or []
        # Asana's typeahead index lags 15-45s (measured on the live API), so a task
        # created moments ago is invisible to ANY search. Tasks WE created this
        # session are known by gid — pull them in so "find the task I just made"
        # works. (A project listing has no such lag, hence only this branch.)
        if q:
            seen = {str(t.get("gid")) for t in rows}
            for nm, gid in list(_RECENT.items()):
                if q.lower() not in nm or gid in seen:
                    continue
                try:
                    fresh = _api("GET", f"/tasks/{gid}",
                                 params={"opt_fields": _F_TASK})
                except AsanaError:
                    continue                    # deleted elsewhere — skip it
                if fresh:
                    rows.append(fresh)
    if who:
        low = who.strip().lower()
        if low in ("me", "myself", "i", "ahmed", "mine"):
            mine = _me_gid()
            rows = [t for t in rows
                    if str((t.get("assignee") or {}).get("gid", "")) == mine]
        else:
            rows = [t for t in rows
                    if low in str((t.get("assignee") or {}).get("name", "")).lower()
                    or low == str((t.get("assignee") or {}).get("gid", ""))]
    if due_before:
        rows = [t for t in rows if t.get("due_on") and t["due_on"] <= due_before]
    if due_after:
        rows = [t for t in rows if t.get("due_on") and t["due_on"] >= due_after]
    if want_done is not None:
        rows = [t for t in rows if bool(t.get("completed")) == want_done]
    if not rows:
        extra = ""
        if not pid:
            projs = _api("GET", f"/workspaces/{_workspace()}/typeahead", params={
                "resource_type": "project", "query": q, "count": 5,
                "opt_fields": "name"}) or []
            if projs:
                extra = ("\nMatching projects: "
                         + ", ".join(str(p.get("name")) for p in projs) + ".")
        return f"Nothing in Asana matches {query!r}.{extra}"
    lines = [_task_line(t) for t in rows[:30]]
    more = f"\n…and {len(rows) - 30} more." if len(rows) > 30 else ""
    return (f"{len(rows)} Asana task(s) matching {query!r} "
            "(free-tier search: matches names, not descriptions):\n"
            + "\n".join(lines) + more)


def search(query: str = "", project: str = "", assignee: str = "",
           due_before: str = "", due_after: str = "", completed=None) -> str:
    """Find tasks. Uses Asana's premium search when the plan has it, else falls back
    to typeahead. NOTE: the search index lags 10-60s, so a task created seconds ago
    may not show — read it with asana_task instead."""
    if not configured():
        return _UNCONFIGURED
    query = str(query or "").strip()
    if not query and not project and not assignee and not due_before and not due_after:
        return "asana_search failed: give a phrase, a project, or a filter."
    try:
        want_done = None if completed is None else _truthy(completed)
        pid = None
        if project:
            pid = _project_gid(project, archived=None)
            if not pid:
                return f"No Asana project matching {project!r}."
        who = _user(assignee) if assignee else ""
        params: dict = {"opt_fields": _F_TASK, "limit": 100,
                        "sort_by": "modified_at"}
        if query:
            params["text"] = query
        if pid:
            params["projects.any"] = pid
        if who:
            params["assignee.any"] = who
        if due_before:
            params["due_on.before"] = str(due_before).strip()[:10]
        if due_after:
            params["due_on.after"] = str(due_after).strip()[:10]
        if want_done is not None:
            params["completed"] = "true" if want_done else "false"
        try:
            rows = _api("GET", f"/workspaces/{_workspace()}/tasks/search",
                        params=params) or []
        except AsanaError as e:
            if e.code in (402, 403):            # not a premium workspace
                return _search_fallback(query, pid,
                                        str(assignee or "").strip(),
                                        str(due_before or "")[:10],
                                        str(due_after or "")[:10], want_done)
            raise
        if not rows:
            return f"Nothing in Asana matches {query!r}."
        lines = [_task_line(t) for t in rows[:30]]
        more = f"\n…and {len(rows) - 30} more." if len(rows) > 30 else ""
        return (f"{len(rows)} Asana task(s) matching {query!r}:\n"
                + "\n".join(lines) + more)
    except Exception as e:  # noqa: BLE001
        return _fail("asana_search", e)


# ---------------------------------------------------------------------------
# ESCAPE HATCH — any endpoint the curated tools don't cover
# ---------------------------------------------------------------------------

def raw(method: str, path: str, body_json: str = "", params_json: str = "") -> str:
    """Call ANY Asana REST endpoint. Returns raw JSON (for the model, not for speech)."""
    if not configured():
        return _UNCONFIGURED
    body = None
    params = None
    if body_json:
        try:
            body = json.loads(body_json)
        except Exception as e:  # noqa: BLE001
            return f"body_json isn't valid JSON: {str(e)[:120]}"
    if params_json:
        try:
            params = json.loads(params_json)
        except Exception as e:  # noqa: BLE001
            return f"params_json isn't valid JSON: {str(e)[:120]}"
    try:
        out = _api(str(method).upper(), str(path), body, params)
        s = json.dumps(out, ensure_ascii=False)
        return s[:4000] + (" …[truncated]" if len(s) > 4000 else "")
    except Exception as e:  # noqa: BLE001
        return _fail("asana_api", e)