"""FastAPI service for ``xa``.
``build_api(...)`` returns a mountable ``FastAPI`` app that reimplements
the ``edualc`` route surface on top of the ``xa`` primitives. The service
is deliberately a function (not a module that imports-and-runs) so it can
be embedded into larger hosts (enlace / tw_platform) or launched
standalone via ``xa serve``.
FastAPI is an optional dependency. Import ``xa.service`` only when you
have installed ``xa[service]`` (pulls in ``fastapi`` + ``uvicorn``).
"""
from __future__ import annotations
import base64
import hashlib
import hmac
import os
import random
import re
import secrets
import signal
import string
import threading
import time
from pathlib import Path
from typing import Any, Callable, Optional
from xa import archive as arch
from xa import claude_cli as ccli
from xa import claude_fs as cfs
from xa import sessions as sess
from xa import store as st
from xa import tmux as tm
try: # pragma: no cover - optional dep
from pydantic import BaseModel
class CreateReq(BaseModel):
name: Optional[str] = None
cwd: Optional[str] = None
model: Optional[str] = None # per-session claude --model (no flag if empty)
effort: Optional[str] = None # per-session claude --effort (no flag if empty)
# True (default): block until the bridge URL appears — the
# original synchronous wire contract, preserved for API users.
# False: return as soon as the tmux session exists and poll
# GET /sessions for the URL (what the bundled webui does).
wait: bool = True
class DeleteReq(BaseModel):
captcha_token: Optional[str] = None
captcha_answer: Optional[str] = None
class ResumeReq(BaseModel):
name: Optional[str] = None
model: Optional[str] = None
effort: Optional[str] = None
class LabelReq(BaseModel):
label: Optional[str] = None # null or "" clears
class HideReq(BaseModel):
hidden: bool = True
except ImportError: # pragma: no cover
pass
# --------------------------------------------------------------------------- #
# auth: reference HTTP Basic implementation (swappable)
# --------------------------------------------------------------------------- #
[docs]
def make_basic_auth(username: str, password: str) -> Callable[..., Any]:
"""Build a FastAPI dependency that enforces HTTP Basic auth.
Returns a callable suitable for ``Depends(...)``. Uses
``secrets.compare_digest`` to avoid timing leaks.
"""
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBasic, HTTPBasicCredentials
basic = HTTPBasic(realm="xa")
def _require_auth(creds: HTTPBasicCredentials = Depends(basic)) -> str:
user_ok = secrets.compare_digest(creds.username, username)
pass_ok = secrets.compare_digest(creds.password, password)
if not (user_ok and pass_ok):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid credentials",
headers={"WWW-Authenticate": 'Basic realm="xa"'},
)
return creds.username
return _require_auth
[docs]
def allow_all() -> str:
"""No-op auth dependency. Useful when the service is behind an external
auth layer (enlace, reverse-proxy mTLS) that already gated the request.
FastAPI inspects the signature to decide what to inject; keep this
nullary so it isn't treated as query-parameter binding.
"""
return "anonymous"
# --------------------------------------------------------------------------- #
# captcha: stateless HMAC-signed challenge (ported from edualc)
# --------------------------------------------------------------------------- #
[docs]
class Captcha:
"""Stateless 4-letter captcha with an HMAC-signed token.
The signing key never leaves the server. Tokens are
``b64(CHALLENGE.EXPIRY).SIG`` where SIG = HMAC-SHA256(key, payload).
Multi-worker safe; no server-side state. Pass one instance of this
class to :func:`build_api` to enable captcha-gated deletes.
"""
def __init__(self, *, key: str, ttl_sec: int = 120) -> None:
self._key = key.encode()
self._ttl_sec = ttl_sec
def _sign(self, payload: bytes) -> str:
sig = hmac.new(self._key, payload, hashlib.sha256).digest()
return base64.urlsafe_b64encode(sig).rstrip(b"=").decode()
[docs]
def issue(self) -> tuple[str, str, int]:
"""Return ``(token, challenge, ttl_sec)``."""
challenge = "".join(random.choices(string.ascii_uppercase, k=4))
expiry = int(time.time()) + self._ttl_sec
payload = f"{challenge}.{expiry}".encode()
token = (
base64.urlsafe_b64encode(payload).rstrip(b"=").decode()
+ "."
+ self._sign(payload)
)
return token, challenge, self._ttl_sec
def check(self, token: str, answer: str) -> bool:
try:
b64_payload, sig = token.rsplit(".", 1)
padded = b64_payload + "=" * (-len(b64_payload) % 4)
payload = base64.urlsafe_b64decode(padded.encode())
challenge, expiry_s = payload.decode().split(".")
expiry = int(expiry_s)
except (ValueError, UnicodeDecodeError):
return False
if not hmac.compare_digest(sig, self._sign(payload)):
return False
if time.time() > expiry:
return False
return secrets.compare_digest(challenge, (answer or "").strip().upper())
# --------------------------------------------------------------------------- #
# build_api
# --------------------------------------------------------------------------- #
_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]{1,48}$")
# Model / effort values ride into a shell command (shlex-quoted, so this
# is defense-in-depth plus early typo feedback). Permissive enough for
# real model ids — dots, colons, slashes, brackets (e.g. "opus[1m]").
_OPT_VALUE_RE = re.compile(r"^[A-Za-z0-9._:\[\]@/-]{1,64}$")
[docs]
def build_api(
*,
auth: Callable[..., Any] = allow_all,
events_store: Optional[st.JsonLinesStore] = None,
pane_store: Optional[st.FileStore] = None,
captcha: Optional[Captcha] = None,
claude_home: Path = cfs.DEFAULT_CLAUDE_HOME,
claude_bin: str = ccli.DEFAULT_CLAUDE_BIN,
session_prefix: str = "xa-",
title: str = "xa",
version: str = "0.1",
include_webui: bool = False,
default_folder: Optional[Path] = None,
):
"""Return a ``FastAPI`` app exposing ``xa``'s session + archive surface.
The caller composes this app into whatever process they have: mount
under ``/api/xa/`` inside enlace, or run standalone via ``uvicorn``.
"""
from dataclasses import asdict
from fastapi import Body, Depends, FastAPI, HTTPException
from fastapi.responses import PlainTextResponse
# Use ``is None`` — not ``or`` — because an empty ``JsonLinesStore``
# is falsy (``__len__`` == 0) which would silently substitute the
# default store and lose the caller's one.
events = events_store if events_store is not None else st.default_events_store()
panes = pane_store if pane_store is not None else st.default_pane_store()
# Default working directory the webui prefills into the "new session"
# dialog and the folder chooser opens at. Resolved lazily so tests /
# callers can point at a tmp dir via the keyword arg.
_default_folder = (
Path(default_folder).expanduser() if default_folder else Path.home()
)
app = FastAPI(title=title, version=version)
def _session_dict(s: sess.Session, overlay_map: Optional[dict] = None) -> dict:
d = asdict(s)
d["transcript_path"] = (
str(d["transcript_path"]) if d["transcript_path"] else None
)
# Apply a label/hidden overlay if present. Lookup order goes from
# most session-specific to least: archive id (= claude session id
# for transcript-only sessions) → claude_session_id → tmux_name.
# tmux names get reused, so they're last to avoid an old session's
# label leaking onto a new one of the same name.
ov = {}
if overlay_map:
for key in (s.id, s.claude_session_id, s.tmux_name):
if key and key in overlay_map:
ov = overlay_map[key]
break
d["label"] = ov.get("label")
d["hidden"] = bool(ov.get("hidden", False))
return d
def _record_dict(r: arch.ArchiveRecord) -> dict:
return asdict(r)
_ARCHIVE_ID_RE = re.compile(r"^[0-9a-f]{6,64}$")
def _resolve_session(id: str) -> Optional[sess.Session]:
"""Find a Session by tmux name, claude session id (full or prefix),
or edualc archive id.
``sess.get_session`` only knows about claude session ids and tmux
names; archive ids live in the events store. This helper bridges
the two so /info, /label and /resume all accept the same id forms
the webui already shows on its session cards.
"""
s = sess.get_session(id, claude_home=claude_home)
if s is not None:
return s
if not _ARCHIVE_ID_RE.match(id):
return None
# Translate archive id → claude_session_id via the event log.
cs_id: Optional[str] = None
for rec in arch.records(events, panes):
if rec.id == id and rec.claude_session_id:
cs_id = rec.claude_session_id
break
if cs_id is None:
return None
return sess.get_session(cs_id, claude_home=claude_home)
def _generate_name() -> str:
existing = {t.name for t in tm.list_sessions()}
for _ in range(50):
stub = "".join(random.choices(string.ascii_lowercase + string.digits, k=5))
candidate = f"{session_prefix}{stub}"
if candidate not in existing:
return candidate
raise HTTPException(500, "Could not generate a unique session name")
# --------------------------------------------------------------------- #
# sessions
# --------------------------------------------------------------------- #
@app.get("/sessions")
def list_sessions(
_: str = Depends(auth),
project: Optional[str] = None,
state: Optional[str] = None,
include_forks: bool = True,
limit: int = 100,
) -> dict:
rows = sess.list_sessions(
project=project,
state=state, # type: ignore[arg-type]
include_forks=include_forks,
limit=limit or None,
claude_home=claude_home,
)
# Freshen the archive before reporting so dead sessions show up.
try:
arch.reconcile(events, panes, tm.list_sessions(), claude_home=claude_home)
except Exception:
pass
overlay_map = arch.overlays(events)
return {"sessions": [_session_dict(r, overlay_map) for r in rows]}
@app.post("/sessions")
def create_session(req: CreateReq, _: str = Depends(auth)) -> dict:
name = req.name or _generate_name()
if not _NAME_RE.match(name):
raise HTTPException(
400, "Invalid name (allowed: letters, digits, _.-, max 48)"
)
for label, value in (("model", req.model), ("effort", req.effort)):
if value and not _OPT_VALUE_RE.match(value):
raise HTTPException(400, f"Invalid {label} value: {value!r}")
existing = {t.name for t in tm.list_sessions()}
if name in existing:
raise HTTPException(409, f"Session '{name}' already exists")
cwd = req.cwd or str(Path.home())
if not Path(cwd).is_dir():
raise HTTPException(400, f"cwd does not exist: {cwd}")
try:
# Fast part: tmux session created, pane piped, `created` event
# emitted. Fails loudly here (name clash, bad cwd, no tmux).
pending = ccli.prepare_spawn(
name,
cwd=cwd,
claude_bin=claude_bin,
claude_home=claude_home,
claude_name=name,
model=req.model or None,
effort=req.effort or None,
archive_store=events,
pane_store=panes,
)
except (FileNotFoundError, RuntimeError) as e:
raise HTTPException(500, str(e))
if req.wait:
result = ccli.complete_spawn(pending)
return {
"name": result.name,
"cwd": result.cwd,
"status": "complete",
"url": result.url,
"url_source": result.url_source,
"claude_session_id": result.claude_session_id,
"warning": result.warning,
"attention": result.attention,
}
# Slow part (URL wait, prompt dismissal, url_acquired event) runs in
# the background. All completion state lands in shared substrates
# (tmux, ~/.claude/sessions/, the events log), so any worker — or a
# webui poll of /sessions — observes it; no in-process registry.
threading.Thread(
target=ccli.complete_spawn,
args=(pending,),
daemon=True,
name=f"xa-spawn-{name}",
).start()
return {
"name": name,
"cwd": cwd,
"status": "starting",
"url": None,
"url_source": None,
"claude_session_id": None,
# Every create response honors the url-or-warning invariant.
"warning": pending.warning
or (
"Session starting — remote-control URL not ready yet; "
"poll GET /sessions for the url."
),
"attention": None,
}
@app.delete("/sessions/{name}")
def delete_session(
name: str,
req: DeleteReq = Body(default=DeleteReq()),
_: str = Depends(auth),
) -> dict:
if not _NAME_RE.match(name):
raise HTTPException(400, "Invalid session name")
if captcha is not None:
if not captcha.check(req.captcha_token or "", req.captcha_answer or ""):
raise HTTPException(
400, "Captcha failed — request a new one and try again"
)
def _tmux_kill(target: str) -> dict:
try:
tm.kill_session(target)
except RuntimeError as e:
raise HTTPException(500, str(e))
return {"killed": target}
existing = {t.name for t in tm.list_sessions()}
if name in existing:
return _tmux_kill(name)
# Not a raw tmux name — resolve like /info does (claude session id,
# full or prefix, or archive id) and kill whatever backs it.
try:
s = _resolve_session(name)
except LookupError as e:
raise HTTPException(400, str(e))
if s is not None and s.state == "live":
# Kill the whole tmux session only when it is DEDICATED to
# this claude (single pane, claude verified in it) — a claude
# living in one window of a shared workspace must not take
# the workspace down with it.
if (
s.tmux_name
and s.tmux_name in existing
and ccli.tmux_session_dedicated_to(s.tmux_name, s.live_pid)
):
return _tmux_kill(s.tmux_name)
# Otherwise signal exactly the one claude — identity-verified
# so a recycled pid can never direct a kill at a bystander.
if s.live_pid and ccli.pid_is_claude(s.live_pid):
try:
os.kill(s.live_pid, signal.SIGTERM)
except ProcessLookupError:
pass # already gone — fall through to 404
except OSError as e:
raise HTTPException(500, f"kill pid {s.live_pid} failed: {e}")
else:
return {"killed": f"pid {s.live_pid}"}
raise HTTPException(
404,
f"No live session matching '{name}' — it may have already "
f"exited; refresh the list.",
)
@app.get("/sessions/{id}/info")
def session_info(id: str, _: str = Depends(auth)) -> dict:
try:
s = _resolve_session(id)
except LookupError as e:
raise HTTPException(400, str(e))
if s is None:
raise HTTPException(404, f"No session matching '{id}'")
out = _session_dict(s)
if s.transcript_path:
out["forensics"] = asdict(cfs.transcript_forensics(s.transcript_path))
out["forensics"]["transcript_path"] = (
str(out["forensics"]["transcript_path"])
if out["forensics"]["transcript_path"]
else None
)
if s.state == "live" and (s.tmux_pane or s.tmux_name):
# Prefer the exact claude pane — a bare session name targets
# the *active* pane, which may be a different window.
out["pane_tail"] = tm.capture_pane(s.tmux_pane or s.tmux_name, lines=80)
return out
@app.get("/sessions/{id}/diagnose")
def diagnose(id: str, _: str = Depends(auth), tail_kb: int = 16) -> dict:
"""Single-stop "what happened" for a session.
Accepts the same id forms as /info: tmux name, claude session id
(full or unique prefix), or archive id. Combines the existing
forensics, pane tail, and a synthesized human-readable hint. The
hint draws on the same signals ``archive.classify_death`` uses,
so it stays consistent with the death reason shown elsewhere.
"""
try:
s = _resolve_session(id)
except LookupError as e:
raise HTTPException(400, str(e))
if s is None:
raise HTTPException(404, f"No session matching '{id}'")
out: dict = {"session": _session_dict(s)}
# Transcript forensics — works for live and archived sessions both.
forensics_obj: Optional[cfs.TranscriptForensics] = None
if s.transcript_path:
try:
forensics_obj = cfs.transcript_forensics(s.transcript_path)
except OSError:
forensics_obj = None
if forensics_obj is not None:
fdict = asdict(forensics_obj)
fdict["transcript_path"] = (
str(fdict["transcript_path"]) if fdict["transcript_path"] else None
)
out["forensics"] = fdict
# Locate the matching archive record (lookup by claude_session_id
# via the records table — that's the join key the events store
# already exposes).
archive_rec: Optional[arch.ArchiveRecord] = None
if s.claude_session_id:
for rec in arch.records(events, panes):
if rec.claude_session_id == s.claude_session_id:
archive_rec = rec
break
# Pane tail. Live: capture from tmux. Archived: read from pane store.
oom_markers: tuple[str, ...] = ()
pane_tail: Optional[str] = None
if s.state == "live" and (s.tmux_pane or s.tmux_name):
pane_tail = tm.capture_pane(s.tmux_pane or s.tmux_name, lines=80)
elif archive_rec is not None and archive_rec.id in panes:
cap = max(1024, tail_kb * 1024)
try:
pane_tail = panes[archive_rec.id][-cap:].decode(
"utf-8", errors="replace"
)
except KeyError:
pane_tail = None
if pane_tail:
out["pane_tail"] = pane_tail
oom_markers = tuple(m for m in arch._OOM_PANE_MARKERS if m in pane_tail)
if oom_markers:
out["oom_signals"] = list(oom_markers)
if archive_rec is not None:
out["archive_id"] = archive_rec.id
out["gone_reason"] = archive_rec.gone_reason
out["gone"] = archive_rec.gone
# Synthesize the hint — consistent with classify_death's verdict.
out["hint"] = arch.synthesize_diagnosis(
state=s.state,
reason=archive_rec.gone_reason if archive_rec else None,
forensics=forensics_obj,
oom_markers=oom_markers,
)
return out
@app.post("/sessions/{id}/resume")
def resume(id: str, req: ResumeReq, _: str = Depends(auth)) -> dict:
try:
s = _resolve_session(id)
except LookupError as e:
raise HTTPException(400, str(e))
if s is None or not s.claude_session_id:
raise HTTPException(404, f"No resumable session matching '{id}'")
for label, value in (("model", req.model), ("effort", req.effort)):
if value and not _OPT_VALUE_RE.match(value):
raise HTTPException(400, f"Invalid {label} value: {value!r}")
try:
result = sess.resume(
s,
name=req.name,
claude_bin=claude_bin,
claude_home=claude_home,
model=req.model or None,
effort=req.effort or None,
)
except (ValueError, FileNotFoundError, RuntimeError) as e:
raise HTTPException(500, str(e))
return {
"name": result.name,
"cwd": result.cwd,
"url": result.url,
"url_source": result.url_source,
"claude_session_id": result.claude_session_id,
"warning": result.warning,
}
@app.post("/sessions/{id}/remote-control")
def enable_remote_control(id: str, _: str = Depends(auth)) -> dict:
"""Make a live, bridgeless, tmux-hosted session remote-reachable.
Sends ``/remote-control`` into the pane (once, prompts permitting)
and waits briefly for the bridge URL. When it can't succeed —
logged-out claude, not tmux-hosted — the response says exactly why
and how to fix it by hand.
"""
try:
s = _resolve_session(id)
except LookupError as e:
raise HTTPException(400, str(e))
if s is None:
raise HTTPException(404, f"No session matching '{id}'")
if s.state != "live":
raise HTTPException(409, "Session is not live")
if s.url:
return {
"url": s.url,
"url_source": s.url_source,
"attention": None,
"hint": None,
"already_enabled": True,
}
if not (s.tmux_pane or s.tmux_name):
raise HTTPException(
409,
"Session is not tmux-hosted — attach the terminal that "
"started it and run /remote-control there.",
)
# Target the exact claude pane when known: sending keystrokes at
# the bare session name would type into whatever pane is active.
url, src, attention = ccli.request_remote_control(
s.tmux_pane or s.tmux_name, claude_home=claude_home
)
return {
"url": url,
"url_source": src,
"attention": attention,
"hint": ccli.attention_hint(attention, tmux_name=s.tmux_name),
"already_enabled": False,
}
@app.patch("/sessions/{id}/label")
def set_label(
id: str,
req: LabelReq = Body(default=LabelReq()),
_: str = Depends(auth),
) -> dict:
"""Set (or clear) a display label for a session.
Accepts any of: tmux session name, archive id, claude session id.
For a live session, also renames the tmux session so lookups by
the new name work natively. For archived / transcript-only
sessions, the label is kept as an overlay only.
"""
label = (req.label or "").strip()
if label and not _NAME_RE.match(label):
raise HTTPException(400, "Label must match [A-Za-z0-9_.-]{1,48}")
# Try to find a matching Session (live or transcript-only). If
# none, we still accept the request and record the overlay —
# archive-only records are identified by their hex id.
try:
s = _resolve_session(id)
except LookupError as e:
raise HTTPException(400, str(e))
keys_to_label: list[str] = [id]
if s is not None:
if s.state == "live" and s.tmux_name and label:
try:
tm.rename_session(s.tmux_name, label)
except RuntimeError as e:
raise HTTPException(500, f"tmux rename failed: {e}")
if s.id not in keys_to_label:
keys_to_label.append(s.id)
if s.claude_session_id and s.claude_session_id not in keys_to_label:
keys_to_label.append(s.claude_session_id)
if s.tmux_name and s.tmux_name not in keys_to_label:
keys_to_label.append(s.tmux_name)
for key in keys_to_label:
arch.append_label(events, id=key, label=label or None)
return {"id": id, "label": label or None}
@app.post("/archive/{archive_id}/hide")
def hide(
archive_id: str,
req: HideReq = Body(default=HideReq()),
_: str = Depends(auth),
) -> dict:
if not re.fullmatch(r"[0-9a-f]{6,64}", archive_id):
raise HTTPException(400, "Invalid archive id")
arch.append_hidden(events, id=archive_id, hidden=req.hidden)
return {"id": archive_id, "hidden": req.hidden}
@app.delete("/archive/{archive_id}/hide")
def unhide(archive_id: str, _: str = Depends(auth)) -> dict:
if not re.fullmatch(r"[0-9a-f]{6,64}", archive_id):
raise HTTPException(400, "Invalid archive id")
arch.append_hidden(events, id=archive_id, hidden=False)
return {"id": archive_id, "hidden": False}
# --------------------------------------------------------------------- #
# archive
# --------------------------------------------------------------------- #
@app.get("/archive")
def archive_list(_: str = Depends(auth), limit: int = 100) -> dict:
try:
arch.reconcile(events, panes, tm.list_sessions(), claude_home=claude_home)
except Exception:
pass
recs = arch.records(events, panes)
if limit:
recs = recs[:limit]
return {"sessions": [_record_dict(r) for r in recs]}
@app.get("/archive/{archive_id}/forensics")
def archive_forensics(archive_id: str, _: str = Depends(auth)) -> dict:
if not re.fullmatch(r"[0-9a-f]{6,64}", archive_id):
raise HTTPException(400, "Invalid archive id")
rec = next((r for r in arch.records(events, panes) if r.id == archive_id), None)
if rec is None:
raise HTTPException(404, "No such archived session")
out = _record_dict(rec)
if rec.cwd and rec.claude_session_id:
path = cfs.transcript_path(
rec.cwd, rec.claude_session_id, claude_home=claude_home
)
if path is not None:
out["transcript_forensics"] = asdict(cfs.transcript_forensics(path))
out["transcript_forensics"]["transcript_path"] = (
str(out["transcript_forensics"]["transcript_path"])
if out["transcript_forensics"]["transcript_path"]
else None
)
return out
@app.get("/archive/{archive_id}/log")
def archive_log(archive_id: str, _: str = Depends(auth), tail_kb: int = 64):
if not re.fullmatch(r"[0-9a-f]{6,64}", archive_id):
raise HTTPException(400, "Invalid archive id")
if archive_id not in panes:
raise HTTPException(404, "No pane log for that session")
data = panes[archive_id]
if tail_kb and len(data) > tail_kb * 1024:
data = data[-tail_kb * 1024 :]
return PlainTextResponse(
data.decode("utf-8", errors="replace"),
media_type="text/plain; charset=utf-8",
)
# --------------------------------------------------------------------- #
# captcha + health
# --------------------------------------------------------------------- #
if captcha is not None:
@app.get("/captcha")
def _captcha(_: str = Depends(auth)) -> dict:
token, challenge, ttl = captcha.issue()
return {"token": token, "challenge": challenge, "ttl_sec": ttl}
# --------------------------------------------------------------------- #
# filesystem browsing (for the webui folder chooser)
# --------------------------------------------------------------------- #
@app.get("/fs/default")
def fs_default(_: str = Depends(auth)) -> dict:
return {"path": str(_default_folder)}
@app.get("/fs/list")
def fs_list(
_: str = Depends(auth),
path: Optional[str] = None,
show_hidden: bool = False,
) -> dict:
"""List directory entries for the folder chooser.
No path-confinement: xa already grants code-execution via spawned
claude sessions, so restricting the browser's view would be
security theater. Auth is the only boundary.
"""
target = Path(path).expanduser() if path else _default_folder
try:
target = target.resolve()
except (OSError, RuntimeError) as e:
raise HTTPException(400, f"Cannot resolve path: {e}")
if not target.exists():
raise HTTPException(404, f"No such path: {target}")
if not target.is_dir():
raise HTTPException(400, f"Not a directory: {target}")
entries: list[dict] = []
try:
children = list(target.iterdir())
except PermissionError as e:
raise HTTPException(403, str(e))
for child in children:
if not show_hidden and child.name.startswith("."):
continue
try:
is_dir = child.is_dir()
except OSError:
# Broken symlink or unreadable — skip silently.
continue
entries.append({"name": child.name, "path": str(child), "is_dir": is_dir})
entries.sort(key=lambda e: (not e["is_dir"], e["name"].lower()))
parent = str(target.parent) if target.parent != target else None
return {"path": str(target), "parent": parent, "entries": entries}
@app.get("/health")
def health() -> dict:
return {"ok": True}
# --------------------------------------------------------------------- #
# optional webui
# --------------------------------------------------------------------- #
#
# The bundled static UI lives at ``xa/webui/`` and is served at the
# mount root when ``include_webui=True``. It talks to the API on the
# same origin — works under ``xa serve`` standalone and under any
# reverse-proxy mount that includes the same prefix for both.
if include_webui:
from fastapi.staticfiles import StaticFiles
webui_root = Path(__file__).parent / "webui"
if webui_root.is_dir():
# `html=True` makes `/` serve `index.html` naturally.
app.mount(
"/", StaticFiles(directory=str(webui_root), html=True), name="webui"
)
return app