app.py

"""
app.py — Fossil Agent as a Hugging Face *Docker* Space (Chainlit chat UI)
=============================================================================
Chainlit frontend for the fossil digitization agent. The user attaches
photographs of a specimen — or picks a case from the example gallery — and asks
for a 3D reconstruction. Follow-up questions reuse the same specimen.

How this differs from the CXR Space it is modelled on
-----------------------------------------------------
**A specimen is a folder of views, not one image.** Meshy's `multi-image-to-3d`
triangulates geometry across 1-4 photographs of the same fossil and invents
whatever no view shows. So the unit of work here is a *set*: attachments are
written into a per-turn specimen folder, and each gallery card arms an image
SET, not a single file.

**The payoff artifact is a mesh, not a picture.** The right-hand panel renders
PNGs (coordinate canvas, keypoint rounds, SAM3 overlay/cutout) *and* GLB meshes
in a real 3D viewer. An iframe would not do.

**Reconstruction costs money.** Meshy is billed per task, so the 3D switch is
OFF by default and the run reports the request it *would* have sent. The toggles
are stored in ADK **session state**, not process env — a Space serves many users
from one process, and a global would let one person's switch spend another
person's credits.

Framework
---------
The **OpenAI Agents SDK**, same as the CXR Space. `Runner.run_streamed()` returns
synchronously and the run is not complete until `stream_events()` is fully
consumed; tool calls surface as `run_item_stream_event`s and the answer streams as
`raw_response_event` token deltas.

Per-run state (the Meshy switch, the texture switch, the collected cutouts) lives
on a `FossilContext` passed as `context=`. It never enters the model's context
window — which matters, because a Space serves many users from one process and a
process-global `FOSSIL_MESHY_ENABLED` would let one person's toggle spend another
person's credits.

Repo layout expected (this file at the Space root):
    Dockerfile
    app.py                 <- this file
    requirements.txt
    README.md              <- carries the HF Docker Space header (sdk: docker)
    chainlit.md
    .chainlit/config.toml
    public/elements/       ArtifactFrame.jsx  FossilGallery.jsx
                           TurnTimer.jsx      TimerStop.jsx
    examples/              examples.csv + one folder of views per specimen
    config/ helpers/ tools/ schemas/ fossil_agents/ skills/   <- the project

Secrets (Space -> Settings -> Secrets):
    OPENAI_API_KEY    required for the default model (gpt-5.6-luna)
    MESHY_API_KEY     required only if 3D reconstruction is enabled
    HF_TOKEN          gated facebook/sam3, and RMBG-2.0 background removal

Variables (all optional):
    FOSSIL_MODEL_ID       default openai/gpt-5.6-luna
    FOSSIL_BG_BACKEND     "local" (GrabCut, no token) or "hf" (RMBG-2.0)
    FOSSIL_MESHY_TEXTURE  default for the texture switch
"""

from __future__ import annotations

import asyncio
import base64
import csv
import hashlib
import json
import mimetypes
import os
import re
import shutil
import sys
import time
import uuid
import zipfile
from pathlib import Path

ROOT = Path(__file__).resolve().parent
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

import chainlit as cl  # noqa: E402
from chainlit.input_widget import Switch  # noqa: E402

from agents import Runner  # noqa: E402

from config.settings import MESHY_TEXTURE, ORCHESTRATOR_MODEL_ID, describe_routing  # noqa: E402
from fossil_agents.orchestrator.orchestrator_agent import build_orchestrator  # noqa: E402
from helpers.specimen import MAX_VIEWS, discover_views  # noqa: E402
from schemas.context import FossilContext  # noqa: E402

# Per-session working root: uploaded views land here, and output/ artifacts too.
WORK_ROOT = Path(os.environ.get("FOSSIL_WORK_DIR", "/tmp/fossil_work"))
WORK_ROOT.mkdir(parents=True, exist_ok=True)

# The agent writes artifacts under ./output/<specimen>/ relative to CWD.
OUTPUT_ROOT = ROOT / "output"

VIEW_ARTIFACT_ACTION = "view_artifact"
SHOW_STEPS_ACTION = "show_steps"
ARM_EXAMPLE_ACTION = "arm_example"
CLEAR_EXAMPLE_ACTION = "clear_example"

EXAMPLES_DIR = ROOT / "examples"
EXAMPLES_CSV = EXAMPLES_DIR / "examples.csv"

IMAGE_SUFFIXES = {".jpg", ".jpeg", ".png"}   # exactly what Meshy accepts
MESH_SUFFIXES = {".glb"}                     # what the 3D viewer can render
# Meshy returns several formats per task. Only GLB renders in the browser, but the
# others are the ones you actually take to Blender / MeshLab / a printer, so they
# are all offered for download.
DOWNLOAD_SUFFIXES = {".glb", ".obj", ".stl", ".mtl", ".usdz", ".fbx", ".ply"}
ARTIFACT_SUFFIXES = IMAGE_SUFFIXES | DOWNLOAD_SUFFIXES

# The artifacts the pipeline makes on the way to the mesh: the coordinate canvas,
# the analyst's keypoint rounds, the background matte, the SAM3 mask/overlay/cutout.
# They are the debugging surface, not the product — so they are collected but kept
# out of the way, and revealed only on request (a button, or just asking).
_INTERMEDIATE_MARKERS = (
    "_canvas.png", "_keypoints_round", "_contact_sheet.png", "_cutout_white.png",
    "_cutout_rgba.png", "_alpha.png", "_sam3_mask.png", "_sam3_overlay.png",
    "_sam3_cutout.png",
)


def _is_intermediate(path: Path) -> bool:
    """Working image, not a deliverable. Meshes and Meshy's own render are the product."""
    if path.suffix.lower() in DOWNLOAD_SUFFIXES:
        return False
    if path.parent.name == "mesh":       # Meshy's preview.png
        return False
    return any(m in path.name for m in _INTERMEDIATE_MARKERS)


INTRO = (
    "### Fossil Agent\n\n"
    "**Attach 1–4 photographs of one fossil from different angles**, or pick a specimen "
    "from the gallery below, then ask for a 3D model.\n\n"
    "Several angles matter: the reconstruction triangulates geometry *across* the views, "
    "and infers whatever no view shows. With one photograph, the entire far side of the "
    "specimen is guessed. With four, it is observed.\n\n"
    "Each view is background-removed, then the vision analyst looks at it, picks its own "
    "SAM3 prompt points, checks them, and segments the specimen. The cutouts are "
    "reconstructed together into a mesh.\n\n"
    "The finished mesh opens in the panel on the right — orbit it, and download it in "
    "`.glb`, `.obj` or `.stl`. Everything else the run produced comes down in a single "
    "archive.\n\n"
    "Ask for the overlay, the mask, or the keypoints at any time and they will open on "
    "the right."
)


# ── Right-hand panel ──────────────────────────────────────────────────────────

def _artifact_props(path: Path) -> dict:
    """Base64 the artifact so the panel renders without a static-file route.

    Meshes are a few MB; inlining them keeps the Space free of served paths at the
    cost of some memory. Images are small.
    """
    mime = mimetypes.guess_type(path.name)[0] or "application/octet-stream"
    if path.suffix.lower() == ".glb":
        mime = "model/gltf-binary"
    return {
        "kind": "mesh" if path.suffix.lower() in MESH_SUFFIXES else "image",
        "title": path.name,
        "filename": path.name,   # names the panel's download button
        "src": f"data:{mime};base64,{base64.b64encode(path.read_bytes()).decode('ascii')}",
    }


async def _show_in_sidebar(art: dict, aid: str) -> None:
    await cl.ElementSidebar.set_title(art["label"])
    await cl.ElementSidebar.set_elements(
        [cl.CustomElement(name="ArtifactFrame", props=art["props"], display="inline")],
        key=f"art-{aid}",
    )


# Asking for the working images in plain language should work as well as the button.
_WANTS_STEPS_RE = re.compile(
    r"\b(overlays?|masks?|key ?points?|canvas|cutouts?|segmentations?|mattes?|"
    r"working images?|intermediates?|steps?|workings?)\b",
    re.IGNORECASE,
)


def _wants_steps(text: str) -> bool:
    return bool(_WANTS_STEPS_RE.search(text or ""))


@cl.action_callback(SHOW_STEPS_ACTION)
async def on_show_steps(action: cl.Action):
    """Reveal the working images for a turn, on request."""
    registry: dict = cl.user_session.get("artifact_registry") or {}
    aids = (action.payload or {}).get("artifact_ids") or []
    arts = [(a, registry[a]) for a in aids if a in registry]
    if not arts:
        await cl.Message(content="Those working images are no longer in this session.").send()
        return

    await cl.Message(
        content="**Working images** — the pipeline's steps for that specimen. The SAM3 "
                "overlay is the one worth checking: it shows whether the cast shadow "
                "stayed out of the mask.",
        actions=[cl.Action(
            name=VIEW_ARTIFACT_ACTION,
            payload={"artifact_id": aid},
            label=art["label"],
            icon="image",
            tooltip="Open in the panel on the right",
        ) for aid, art in arts],
    ).send()


@cl.action_callback(VIEW_ARTIFACT_ACTION)
async def on_view_artifact(action: cl.Action):
    registry: dict = cl.user_session.get("artifact_registry") or {}
    aid = (action.payload or {}).get("artifact_id")
    art = registry.get(aid)
    if not art:
        await cl.ElementSidebar.set_title("Unavailable")
        await cl.ElementSidebar.set_elements(
            [cl.Text(content="That artifact is no longer available in this session.")])
        return
    await _show_in_sidebar(art, aid)


# ── Example gallery (image SETS, one per specimen) ────────────────────────────

def _load_examples() -> list[dict]:
    """examples/examples.csv — one row per specimen, with a SET of views.

    Columns: id, title, folder, prompt, source
    `folder` is a directory of 1-4 photographs, resolved against the Space root.
    A specimen is a set, so the gallery cannot key on a single file.
    """
    cached = getattr(_load_examples, "_cache", None)
    if cached is not None:
        return cached

    rows: list[dict] = []
    if EXAMPLES_CSV.exists():
        with EXAMPLES_CSV.open(newline="", encoding="utf-8") as fh:
            for row in csv.DictReader(fh):
                rel = (row.get("folder") or "").strip()
                folder = (ROOT / rel).resolve() if rel else None
                views: list[str] = []
                if folder and folder.is_dir():
                    try:
                        views = discover_views(folder)[:MAX_VIEWS]
                    except (FileNotFoundError, ValueError):
                        views = []
                rows.append({
                    "id": (row.get("id") or "").strip(),
                    "title": (row.get("title") or row.get("id") or "Specimen").strip(),
                    "prompt": (row.get("prompt") or
                               "Digitize this specimen and reconstruct a 3D mesh.").strip(),
                    "folder": str(folder) if folder else "",
                    "views": views,
                    "source": (row.get("source") or "").strip(),
                })
    _load_examples._cache = rows  # type: ignore[attr-defined]
    return rows


def _thumb(path: str, box: int = 320) -> str:
    """Small base64 thumbnail. Specimen photos are 12MP; inlining them raw would
    push megabytes of base64 per card into the first message."""
    from PIL import Image

    img = Image.open(path).convert("RGB")
    img.thumbnail((box, box), Image.LANCZOS)
    import io

    buf = io.BytesIO()
    img.save(buf, format="JPEG", quality=80)
    return "data:image/jpeg;base64," + base64.b64encode(buf.getvalue()).decode("ascii")


def _gallery_items() -> list[dict]:
    items: list[dict] = []
    for e in _load_examples():
        if not e["views"]:
            continue
        items.append({
            "id": e["id"],
            "title": e["title"],
            "prompt": e["prompt"],
            "n_views": len(e["views"]),
            "thumbs": [_thumb(v) for v in e["views"]],
            "source": e["source"],
        })
    return items


@cl.action_callback(ARM_EXAMPLE_ACTION)
async def on_arm_example(action: cl.Action):
    """Arm a specimen for the next send. Deliberately does NOT run the agent —
    the card prefills the prompt so the user can edit it first."""
    ex_id = (action.payload or {}).get("example_id")
    ex = next((e for e in _load_examples() if e["id"] == ex_id), None)
    if not ex or not ex["views"]:
        return
    cl.user_session.set("pending_example", ex)


@cl.action_callback(CLEAR_EXAMPLE_ACTION)
async def on_clear_example(action: cl.Action):
    cl.user_session.set("pending_example", None)


# ── Session ───────────────────────────────────────────────────────────────────

@cl.on_chat_start
async def on_chat_start():
    settings = await cl.ChatSettings([
        Switch(
            id="meshy_enabled",
            label="Reconstruct 3D mesh",
            initial=True,
        ),
        Switch(
            id="include_texture",
            label="Include texture (off = geometry only, faster)",
            initial=bool(MESHY_TEXTURE),
        ),
    ]).send()

    work_dir = WORK_ROOT / uuid.uuid4().hex
    work_dir.mkdir(parents=True, exist_ok=True)

    notes: list[str] = []
    async with cl.Step(name="Preparing the fossil agent…", type="tool"):
        routing = describe_routing(ORCHESTRATOR_MODEL_ID)
        if not routing["api_key_present"]:
            notes.append("⚠️ No OPENAI_API_KEY — add it in the Space secrets.")

        agent = build_orchestrator()

        # Per-run state, NOT process env. See the module docstring.
        context = FossilContext(
            meshy_enabled=bool(settings.get("meshy_enabled", True)),
            include_texture=bool(settings.get("include_texture", MESHY_TEXTURE)),
        )

    cl.user_session.set("agent", agent)
    cl.user_session.set("context", context)
    cl.user_session.set("conversation", None)     # None until the first turn runs
    cl.user_session.set("work_dir", work_dir)
    cl.user_session.set("started_at", time.time())
    cl.user_session.set("seen_artifacts", {})     # path -> (mtime_ns, size)
    cl.user_session.set("artifact_registry", {})  # id -> {label, props, path}
    cl.user_session.set("pending_example", None)
    cl.user_session.set("specimen_dir", None)
    cl.user_session.set("model_id", routing["FOSSIL_MODEL_ID"])

    await cl.Message(content=INTRO + (("\n\n" + "\n".join(notes)) if notes else "")).send()

    items = _gallery_items()
    if items:
        await cl.Message(
            content="**Example gallery** — each card is a specimen with its own set of "
                    "views. Picking one prefills its prompt; edit it, then send.",
            elements=[cl.CustomElement(
                name="FossilGallery", props={"examples": items}, display="inline")],
        ).send()


@cl.on_settings_update
async def on_settings_update(settings):
    """Push the switches onto this chat's FossilContext, where the tools read them."""
    context: FossilContext = cl.user_session.get("context")
    if context is None:
        return
    context.meshy_enabled = bool(settings.get("meshy_enabled", True))
    context.include_texture = bool(settings.get("include_texture", MESHY_TEXTURE))


# ── Live step labels ──────────────────────────────────────────────────────────

FRIENDLY_TOOL = {
    "inspect_specimen_tool":   "Inspecting the specimen's views",
    "select_views_tool":       "Choosing which views to use",
    "record_view_choice_tool": "Recording the view choice",
    "remove_background_tool":  "Removing the background",
    "select_keypoints_tool":   "Vision analyst — selecting keypoints",
    "check_keypoints_tool":    "Checking the chosen keypoints",
    "segment_fossil_sam3_tool": "Segmenting with SAM3",
    "generate_3d_model_tool":  "Reconstructing 3D mesh (Meshy)",
}


def _compact(value, limit: int = 700) -> str:
    if value is None:
        return ""
    s = value if isinstance(value, str) else json.dumps(value, default=str)
    s = s.strip()
    return s if len(s) <= limit else s[:limit] + " …"


def _working_md(activity: str) -> str:
    return f"🦴 **Digitizing the specimen…**\n\n⏳ _{activity}_"


def _meshy_md(status: str, percent: int) -> str:
    """Live Meshy progress.

    Reconstruction polls for minutes. Without a moving number this is
    indistinguishable from a hang, and the temptation is to reload the page —
    which abandons a task you have already paid for.
    """
    pct = max(0, min(100, int(percent or 0)))
    filled = round(pct / 5)
    bar = "█" * filled + "░" * (20 - filled)
    label = {
        "PENDING": "queued at Meshy",
        "IN_PROGRESS": "reconstructing geometry",
        "SUCCEEDED": "done — downloading the mesh",
        "FAILED": "failed",
    }.get(status, status.lower() or "working")
    return (f"🧊 **Reconstructing the 3D mesh…**\n\n"
            f"`{bar}` **{pct}%** — _{label}_\n\n"
            f"_This takes a few minutes. Leave this tab open — reloading will lose it._")


# ── Specimen resolution ───────────────────────────────────────────────────────

def _attached_images(message: cl.Message) -> list[str]:
    return [
        el.path for el in (message.elements or [])
        if (getattr(el, "mime", "") or "").startswith("image")
        and getattr(el, "path", None)
        and Path(el.path).suffix.lower() in IMAGE_SUFFIXES
    ]


def _make_specimen_dir(views: list[str], work_dir: Path, name: str) -> tuple[str, list[str]]:
    """Copy the views into a folder — the pipeline's unit of work.

    Returns (folder, dropped). Meshy takes at most 4 views; beyond that the agent
    would have to choose, and rather than make that call silently here we keep the
    first 4 and say what was dropped. (The agent's own view-selector handles the
    >4 case when a folder is passed on the CLI.)
    """
    folder = work_dir / name
    folder.mkdir(parents=True, exist_ok=True)
    kept, dropped = views[:MAX_VIEWS], views[MAX_VIEWS:]
    for i, v in enumerate(kept, 1):
        shutil.copy2(v, folder / f"{i:02d}_{Path(v).name}")
    return str(folder), dropped


# ── Artifact collection ───────────────────────────────────────────────────────

_MISSING = object()


_MIME = {
    ".glb": "model/gltf-binary",
    ".obj": "model/obj",
    ".stl": "model/stl",
    ".mtl": "text/plain",
    ".ply": "application/octet-stream",
    ".usdz": "model/vnd.usdz+zip",
    ".fbx": "application/octet-stream",
    ".png": "image/png",
    ".jpg": "image/jpeg",
    ".jpeg": "image/jpeg",
}


def _aid_for(path: str) -> str:
    return "a" + hashlib.sha1(path.encode()).hexdigest()[:10]


_LABELS = [
    ("_sam3_cutout.png", "🪨 SAM3 cutout (this is what Meshy sees)"),
    ("_sam3_overlay.png", "🎯 SAM3 mask overlay"),
    ("_sam3_mask.png", "⬜ SAM3 mask"),
    ("_canvas.png", "📐 Coordinate canvas (what the analyst reads)"),
    ("_keypoints_round", "📍 Analyst's keypoints"),
    ("_contact_sheet.png", "🖼️ Contact sheet"),
    ("_cutout_white.png", "✂️ Background removed"),
    (".glb", "🧊 3D mesh"),
]


def _label_for(path: Path) -> str:
    for suffix, label in _LABELS:
        if suffix in path.name:
            return f"{label} — {path.stem}"
    return path.name


def _collect_new_artifacts(seen: dict) -> list[Path]:
    """Sweep output/ for artifacts produced or rewritten this session.

    Dedup by (mtime, size): the keypoint preview is rewritten on each revision
    round, and a rewritten file should re-surface rather than be swallowed.
    """
    started = float(cl.user_session.get("started_at") or 0.0)
    if not OUTPUT_ROOT.is_dir():
        return []

    fresh: list[Path] = []
    for fp in OUTPUT_ROOT.rglob("*"):
        if not fp.is_file() or fp.suffix.lower() not in ARTIFACT_SUFFIXES:
            continue
        try:
            st = fp.stat()
        except OSError:
            continue
        if st.st_mtime < started - 1.0:
            continue  # a previous session's leftovers in a shared container
        sig = (st.st_mtime_ns, st.st_size)
        key = str(fp)
        if seen.get(key, _MISSING) != sig:
            seen[key] = sig
            fresh.append(fp)

    # Meshes last: the mesh is the payoff, so it should be the panel's final state.
    fresh.sort(key=lambda p: (p.suffix.lower() == ".glb", p.stat().st_mtime))
    return fresh


def _register(new: list[Path], registry: dict, turn_seen: set):
    """Register artifacts, keeping the deliverables and the working images apart.

    Returns (actions, files, last_final_aid, intermediate_aids).

    `actions`/`files` carry only the DELIVERABLES — the mesh, in every format Meshy
    returned. The working images (canvas, keypoints, matte, SAM3 overlay) are
    registered so they can still be opened, but they get no button unless asked for.
    `last` deliberately tracks only a final artifact, so the right-hand panel lands
    on the mesh rather than on whichever PNG happened to be written last.
    """
    actions, files, last, intermediates = [], [], None, []
    for fp in new:
        aid = _aid_for(str(fp))
        try:
            props = _artifact_props(fp)
        except OSError:
            continue
        registry[aid] = {"label": _label_for(fp), "props": props, "path": str(fp)}

        if str(fp) in turn_seen:
            continue
        turn_seen.add(str(fp))

        if _is_intermediate(fp):
            # Registered so "show me the overlay" still works, but it gets no chip and
            # no file attachment. Twenty PNG chips per run is a wall of machinery the
            # user did not ask for and cannot act on.
            intermediates.append(aid)
            continue

        if fp.suffix.lower() not in DOWNLOAD_SUFFIXES:
            # e.g. Meshy's preview.png — a flat render of the mesh you can already
            # orbit. Into the archive, not the chat.
            intermediates.append(aid)
            continue

        last = aid
        if fp.suffix.lower() in MESH_SUFFIXES:
            actions.append(cl.Action(
                name=VIEW_ARTIFACT_ACTION,
                payload={"artifact_id": aid},
                label="🧊 View 3D mesh",
                icon="box",
                tooltip="Open in the panel on the right",
            ))
        if fp.suffix.lower() in DOWNLOAD_SUFFIXES:
            files.append(cl.File(
                name=fp.name, path=str(fp),
                mime=_MIME.get(fp.suffix.lower(), "application/octet-stream"),
                display="inline"))
    return actions, files, last, intermediates


def _build_results_archive(work_dir: Path) -> Path | None:
    """Zip everything this run produced: meshes, cutouts, masks, keypoints, canvases.

    The working images are deliberately absent from the chat — nobody wants twenty PNG
    chips. But they are the only way to audit a mesh after the fact: whether the mask
    kept the cast shadow out, where the analyst put its points, what the coordinate
    canvas looked like. Hidden and discarded are very different things, and this archive
    is the difference.
    """
    started = float(cl.user_session.get("started_at") or 0.0)
    if not OUTPUT_ROOT.is_dir():
        return None

    members = [fp for fp in OUTPUT_ROOT.rglob("*")
               if fp.is_file() and fp.stat().st_mtime >= started - 1.0]
    if not members:
        return None

    archive = Path(work_dir) / "fossil_results.zip"
    with zipfile.ZipFile(archive, "w", zipfile.ZIP_DEFLATED) as zf:
        for fp in members:
            zf.write(fp, fp.relative_to(OUTPUT_ROOT))
    return archive


# ── One turn ──────────────────────────────────────────────────────────────────

@cl.on_message
async def on_message(message: cl.Message):
    work_dir = cl.user_session.get("work_dir")
    if work_dir is None:
        await cl.Message(content="Session not initialized — refresh to start a new chat.").send()
        return

    specimen_dir: str | None = None
    dropped: list[str] = []

    views = _attached_images(message)
    if views:
        specimen_dir, dropped = _make_specimen_dir(
            views, Path(work_dir), f"upload_{uuid.uuid4().hex[:8]}")

    pending = cl.user_session.get("pending_example")
    if specimen_dir is None and pending:
        specimen_dir = pending["folder"]
        await cl.Message(
            content=f"_Specimen: **{pending['title']}** — {len(pending['views'])} view(s)._",
            elements=[cl.Image(name=Path(v).name, path=v, display="inline")
                      for v in pending["views"]],
        ).send()
    cl.user_session.set("pending_example", None)

    if dropped:
        await cl.Message(
            content=f"⚠️ Meshy accepts at most **{MAX_VIEWS} views**; using the first "
                    f"{MAX_VIEWS} and ignoring {len(dropped)}. If a dropped view showed "
                    f"a face the others don't — a broken cross-section, say — reattach "
                    f"it in place of a redundant angle. Which four you pick matters more "
                    f"than how many you had."
        ).send()

    if specimen_dir:
        cl.user_session.set("specimen_dir", specimen_dir)
    else:
        specimen_dir = cl.user_session.get("specimen_dir")

    if not specimen_dir:
        await cl.Message(
            content="Attach 1–4 photographs of one fossil from different angles, or pick "
                    "a specimen from the gallery."
        ).send()
        return

    await _run_turn((message.content or "").strip(), specimen_dir, bool(views or pending))


async def _run_turn(prompt: str, specimen_dir: str, is_new_specimen: bool) -> None:
    agent = cl.user_session.get("agent")
    context: FossilContext = cl.user_session.get("context")
    conversation = cl.user_session.get("conversation")
    seen = cl.user_session.get("seen_artifacts")
    registry = cl.user_session.get("artifact_registry") or {}

    text = prompt or "Digitize this specimen and reconstruct a 3D mesh."
    if is_new_specimen:
        text += f"\n\nSpecimen (folder of views): {specimen_dir}"
    text += (
        f"\n\n[3D reconstruction is "
        f"{'ENABLED' if context.meshy_enabled else 'DISABLED'} for this run; texture "
        f"{'on' if context.include_texture else 'off'}. Do not try to override this.]"
    )
    run_input = text if conversation is None else conversation + [
        {"role": "user", "content": text}]

    t0 = time.monotonic()
    timer_id = uuid.uuid4().hex[:8]
    answer = cl.Message(content="", elements=[cl.CustomElement(
        name="TurnTimer", props={"id": timer_id}, display="inline")])
    await answer.send()
    answer.content = _working_md("starting the pipeline")
    await answer.update()

    streaming = False
    open_steps: list[list] = []      # [call_id, Step, tool_name]
    turn_actions: list = []
    turn_files: list = []
    turn_seen: set = set()
    turn_intermediates: list = []   # registered, but not shown unless asked for
    sidebar_target: list = [None]

    def _match(call_id):
        for i, entry in enumerate(open_steps):
            if call_id is not None and entry[0] == call_id:
                return open_steps.pop(i)
        return open_steps.pop(0) if open_steps else None

    async def _surface():
        new = _collect_new_artifacts(seen)
        if not new:
            return
        acts, fls, last, inters = _register(new, registry, turn_seen)
        turn_actions.extend(acts)
        turn_files.extend(fls)
        turn_intermediates.extend(inters)
        cl.user_session.set("seen_artifacts", seen)
        cl.user_session.set("artifact_registry", registry)
        if last:
            sidebar_target[0] = last

    # The Meshy tool runs in a worker thread (the Agents SDK dispatches sync tools
    # via to_thread), so the event loop stays free — which is what makes a live
    # progress bar possible at all. This watcher just mirrors what that thread
    # publishes onto the context.
    stop_watch = asyncio.Event()

    async def _watch_meshy():
        last = None
        while not stop_watch.is_set():
            if context.meshy_status and not streaming:
                now = (context.meshy_status, context.meshy_percent)
                if now != last:
                    last = now
                    answer.content = _meshy_md(*now)
                    await answer.update()
            await asyncio.sleep(1.5)

    watcher = asyncio.create_task(_watch_meshy())

    try:
        # run_streamed() returns synchronously; the run is not complete until
        # stream_events() has been fully consumed.
        result = Runner.run_streamed(
            agent, run_input, context=context, max_turns=40)

        async for ev in result.stream_events():
            if ev.type == "raw_response_event":
                data = getattr(ev, "data", None)
                if getattr(data, "type", "") == "response.output_text.delta":
                    if not streaming:
                        streaming = True
                        answer.content = ""      # drop the working placeholder
                    await answer.stream_token(getattr(data, "delta", "") or "")
                continue

            if ev.type != "run_item_stream_event":
                continue

            # No per-tool Step chips. Chainlit renders each one as a "Used <tool>"
            # row, and a four-view run produces twenty of them — a wall of machinery
            # nobody reads. One status line that says what is happening right now is
            # more informative and far less noise.
            if ev.name == "tool_called":
                raw = getattr(ev.item, "raw_item", None)
                tname = getattr(raw, "name", None) or "tool"
                if not streaming:
                    answer.content = _working_md(FRIENDLY_TOOL.get(tname, tname))
                    await answer.update()

            elif ev.name == "tool_output":
                pass

                # A tool may have just written an artifact — surface it now rather
                # than making the user wait for the whole turn.
                await _surface()
    except Exception as e:  # noqa: BLE001 — keep the chat alive
        stop_watch.set()
        answer.content = f"❌ Run failed: {e}"
        answer.elements = [cl.CustomElement(
            name="TimerStop", props={"id": timer_id}, display="inline")]
        await answer.update()
        return

    stop_watch.set()
    await watcher

    await _surface()

    if sidebar_target[0] and sidebar_target[0] in registry:
        await _show_in_sidebar(registry[sidebar_target[0]], sidebar_target[0])

    # If the user explicitly ASKED to see the working images, show them. Otherwise
    # they stay out of the chat entirely and live in the results archive.
    if turn_intermediates and _wants_steps(prompt):
        for aid in turn_intermediates:
            art = registry.get(aid)
            if art:
                turn_actions.append(cl.Action(
                    name=VIEW_ARTIFACT_ACTION,
                    payload={"artifact_id": aid},
                    label=art["label"],
                    icon="image",
                    tooltip="Open in the panel on the right",
                ))

    content = str(result.final_output)
    footer = []

    meshes = [f for f in turn_files
              if getattr(f, "name", "").lower().endswith(tuple(DOWNLOAD_SUFFIXES))]
    if meshes:
        footer.append(
            "🧊 *Mesh ready — open it in the panel on the right and orbit it. Look at "
            "the faces your photographs didn't show; Meshy inferred those.*")

    if not context.meshy_enabled:
        footer.append("🧊 *3D reconstruction is off for this session — turn it on in settings (⚙️).*")

    # Everything the run produced — meshes AND the working images — in one download.
    # The intermediates are hidden from the chat, not thrown away: they are how you
    # audit a mesh after the fact (did the mask eat the cast shadow?), and Meshy does
    # not refund a task that succeeded on a bad cutout.
    archive = _build_results_archive(Path(cl.user_session.get("work_dir")))
    if archive:
        turn_files.append(cl.File(
            name="fossil_results.zip", path=str(archive),
            mime="application/zip", display="inline"))
        footer.append(
            "📦 *`fossil_results.zip` — everything else the run produced: the SAM3 "
            "cutouts and overlays, the analyst's keypoints, the coordinate canvases.*")

    footer.append(f"⏱ *Took {time.monotonic() - t0:.0f}s.*")
    content += "\n\n---\n" + "\n\n".join(footer)

    turn_files.append(cl.CustomElement(
        name="TimerStop", props={"id": timer_id}, display="inline"))
    answer.content = content
    answer.elements = turn_files
    answer.actions = turn_actions
    await answer.update()

    # Persist history so follow-ups reuse the specimen and its artifacts.
    cl.user_session.set("conversation", result.to_input_list())