handler.star

# Copyright (c) ClaceIO, LLC
# SPDX-License-Identifier: Apache-2.0
load("openrun.in", "openrun")
load("openrun_admin.in", "openrun_admin")
load("build.in", "build")
load("utils.star", "utils")

# Route handlers for the console. Each screen has a *_data function which
# builds the full page context; action handlers run the mutation and re-render
# the same context with a Flash/FlashError message. Mutation results must read
# ret.error BEFORE the *_data call: an unread plugin error fails the next
# plugin call. The error_handler in app.star is the fallback when that is
# missed.


# ---------- Overview ----------


def ov_can(perms, perm):
    # True when the caller holds perm (or the admin super-user permission)
    return bool(perms.get(perm) or perms.get("admin"))


def ov_running(entries):
    # Count of container entries in the running state
    running = 0
    for entry in entries:
        if entry["state"] == "running":
            running += 1
    return running


def ov_uptime(secs):
    # Compact uptime for the stat tile: "42m", "3h 12m", "12d 4h"
    secs = int(secs)
    if secs < 3600:
        return "%dm" % (secs // 60)
    if secs < 86400:
        return "%dh %dm" % (secs // 3600, (secs % 3600) // 60)
    return "%dd %dh" % (secs // 86400, (secs % 86400) // 3600)


def ov_trim_time(t):
    # Timestamps are time.time values, passed through to the templates; the
    # go zero time (unset) maps to ""
    return utils.nonzero_time(t)


def ov_apps_tile(perms):
    # Big apps tile: totals, prod/dev/declarative split, pending promotion
    # count and a by-spec mini bar chart. Uses plain list_apps (the
    # check_approval audit sweep is too slow for a landing page; the
    # needs-approval chip lazy-loads via overview_approvals_handler).
    # Tiles the user cannot read return None and are OMITTED from the
    # page (overview shows only accessible info, unlike the
    # disabled-with-tooltip convention for action controls)
    if not ov_can(perms, "app:read"):
        return None
    ret = openrun.list_apps(include_internal=True)
    if ret.error:
        return {"Error": ret.error}
    rows = build_app_rows(ret.value)

    dev = 0
    promote = 0
    spec_counts = {}
    for row in rows:
        if row["is_dev"]:
            dev += 1
        if row["staging"] and row["staging"]["ahead"]:
            promote += 1
        spec = row["spec"] or "custom"
        spec_counts[spec] = spec_counts.get(spec, 0) + 1

    specs = sorted(spec_counts.items(), key=lambda kv: (-kv[1], kv[0]))
    top = specs[:4]
    other = 0
    for _, count in specs[4:]:
        other += count
    max_count = other
    for _, count in top:
        max_count = max(max_count, count)
    by_spec = []
    for name, count in top:
        by_spec.append({"label": name, "count": count,
                        "pct": count * 100 // max_count, "other": False})
    if other:
        by_spec.append({"label": "other", "count": other,
                        "pct": other * 100 // max_count, "other": True})

    return {
        "Total": len(rows),
        "Prod": len(rows) - dev,
        "Dev": dev,
        "Promote": promote,
        "BySpec": by_spec,
    }


def ov_syncs_tile(perms):
    # Sync health: one status dot per sync entry, failing entries first
    if not ov_can(perms, "sync:read"):
        return None
    ret = openrun.list_sync()
    if ret.error:
        return {"Error": ret.error}

    entries = []
    failing = 0
    disabled = 0
    for entry in ret.value:
        state = entry["status"]["state"]  # Enabled / Disabled / Failing
        if state == "Failing":
            failing += 1
        elif state == "Disabled":
            disabled += 1
        entries.append({
            "id": entry["id"],
            "repo": entry["path"],
            "state": state,
            "last_exec": ov_trim_time(entry["status"]["last_execution_time"]),
            "error": entry["status"].get("error") or "",
        })
    order = {"Failing": 0, "Enabled": 1, "Disabled": 2}
    entries = sorted(entries, key=lambda e: (order.get(e["state"], 3), e["repo"]))
    return {
        "Total": len(entries),
        "Failing": failing,
        "Disabled": disabled,
        "Ok": len(entries) - failing - disabled,
        "Syncs": entries[:12],
        "More": max(0, len(entries) - 12),
    }


def ov_services_tile(perms):
    # Services and bindings counts (tiny table scans)
    tile = {}
    if ov_can(perms, "service:read"):
        ret = openrun.list_services()
        if not ret.error:
            tile["Services"] = len(ret.value)
    if ov_can(perms, "binding:read"):
        ret = openrun.list_bindings()
        if not ret.error:
            bindings = ret.value
            auto = 0
            for b in bindings:
                if b["path"].startswith("/auto/"):
                    auto += 1
            tile["Bindings"] = len(bindings)
            tile["Auto"] = auto
    if "Services" not in tile and "Bindings" not in tile:
        return None
    return tile


def ov_server_tile(perms):
    # Server identity chips + metadata replication state, all in-memory
    # server-side (server_info never touches the DB or external services).
    # config:basic_read is implied by config:read, but get_permissions
    # reports held permissions literally, so both are checked
    if not (ov_can(perms, "config:basic_read") or perms.get("config:read")):
        return None
    ret = openrun.server_info()
    if ret.error:
        return {"Error": ret.error}
    info = ret.value
    repl = []
    for entry in info["metadata_replication"]:
        repl.append({
            "target": entry["target"],
            "state": entry["state"],
            "last_sync": ov_trim_time(entry.get("last_sync") or ""),
            "error": entry.get("error") or "",
        })
    commit = info["commit"]
    return {
        "Version": info["version"],
        # short_sha would mangle the "dev_build" placeholder of unreleased
        # binaries into "dev_bui"
        "Commit": commit if commit == "dev_build" else utils.short_sha(commit),
        "Uptime": ov_uptime(info["uptime_secs"]),
        "MetadataDB": info["metadata_db_type"],
        "AuditDB": info["audit_db_type"],
        "Runtime": info.get("container_runtime") or "",
        "IsLeader": info["is_leader"],
        "MetadataRepl": repl,
        # The MCP mini tile: the management endpoint and the app actions
        # endpoint state (api.app_mcp.served is enabled AND RBAC enforced)
        "Api": info.get("api"),
    }


def ov_activity_tile(perms, scope):
    # Recent activity ticker: the last few audit events. The default
    # "system" scope lists management operations only; "all" includes
    # http/action/custom events (the scope chips re-render via the
    # /overview/activity fragment)
    if not ov_can(perms, "audit:read"):
        return None
    event_type = "system" if scope != "all" else ""
    # start_date is a required arg in the plugin signature (empty = no
    # filter). 100 events, ~20 visible - the ticker box scrolls the rest
    ret = openrun.list_audit_events(event_type=event_type, start_date="", limit=100)
    if ret.error:
        return {"Error": ret.error}
    events = []
    for entry in ret.value:
        status = entry.get("status") or ""
        is_error = bool(status) and status != "Success" and \
            not status.startswith("2") and not status.startswith("3")
        events.append({
            "user": entry.get("user_id") or "",
            "operation": entry.get("operation") or entry.get("event_type") or "",
            "target": entry.get("target") or "",
            "time": ov_trim_time(entry.get("create_time") or ""),
            "is_error": is_error,
        })
    return {"Events": events, "Scope": "all" if scope == "all" else "system"}


def ov_repl_tile(perms, server):
    # Replication tile: the metadata state (from server_info, so visible
    # with config:basic_read) plus the lazy-loaded app binding rows
    # (app:read - the server filters rows to the caller's readable apps).
    # None - and no tile - when the user can see neither. The
    # "not configured" metadata note renders only when the caller could
    # actually read the metadata state (ShowMetadata), never as a guess
    show_bindings = ov_can(perms, "app:read")
    show_metadata = bool(server) and not server.get("Error")
    metadata = server["MetadataRepl"] if show_metadata else []
    if not show_metadata and not show_bindings:
        return None
    return {"Metadata": metadata, "ShowMetadata": show_metadata,
            "ShowBindings": show_bindings}


def overview_data(req):
    # Overview home page: fleet counts and health at a glance. Tiles the
    # user cannot read - or whose feature is disabled for this install -
    # are omitted entirely (each ov_*_tile returns None). Every plugin
    # call here must be cheap (in-memory or one small table read):
    # container state and binding replication cost external calls, so
    # those tiles lazy-load via the /overview/containers and
    # /overview/replication fragments
    perms = utils.get_perms()
    server = ov_server_tile(perms)
    data = {
        "Title": "Overview",
        "Nav": "overview",
        "Perms": perms,
        "Apps": ov_apps_tile(perms),
        "Syncs": ov_syncs_tile(perms),
        "ServicesTile": ov_services_tile(perms),
        "Server": server,
        "Repl": ov_repl_tile(perms, server),
        "Activity": ov_activity_tile(perms, "system"),
        "ShowContainers": bool(perms.get("feature:container") and
                               ov_can(perms, "container:read")),
    }
    data["Empty"] = not (data["Apps"] or data["Syncs"] or data["ServicesTile"] or
                         data["Server"] or data["Repl"] or data["Activity"] or
                         data["ShowContainers"])
    return data


def ov_container_kind_row(label, ctype):
    # One "X of Y running" stat row for a special container listing
    ret = openrun.list_containers(type=ctype)
    if ret.error:
        return {"label": label, "error": ret.error}
    return {"label": label, "running": ov_running(ret.value),
            "total": len(ret.value)}


def overview_containers_handler(req):
    # Lazy containers tile: list_containers shells out to the container
    # daemon, so it renders as a skeleton first and loads here
    perms = utils.get_perms()
    data = {"Perms": perms, "Loaded": True}
    if not perms.get("feature:container"):
        # Only reachable by a direct fragment request: the page does not
        # render the lazy loader when the container feature is disabled
        data["Error"] = "containers are disabled for this install"
        return data
    if not ov_can(perms, "container:read"):
        data["Error"] = "requires the container:read or admin permission"
        return data
    ret = openrun.list_containers()
    if ret.error:
        data["Error"] = ret.error
        return data

    # The managed list includes the litestream replication sidecars (they
    # carry the app.id label); split them out by their -ls name suffix -
    # the same heuristic the replication status API uses (ContainerInfo
    # does not expose labels)
    apps_total = 0
    apps_running = 0
    ls_total = 0
    ls_running = 0
    sc_total = 0
    sc_running = 0
    job_total = 0
    job_running = 0
    runtime = ""
    for entry in ret.value:
        runtime = entry["runtime"]
        is_running = entry["state"] == "running"
        if entry["name"].endswith("-ls"):
            ls_total += 1
            ls_running += 1 if is_running else 0
        elif entry.get("job"):
            # Job run containers (both runtimes): finished ones stay until
            # their run record is retired, so most are stopped
            job_total += 1
            job_running += 1 if is_running else 0
        elif entry.get("role") == "sidecar":
            # App sidecars are separate labeled containers on docker/podman
            # (in-pod on kubernetes, counted with their pod)
            sc_total += 1
            sc_running += 1 if is_running else 0
        else:
            apps_total += 1
            apps_running += 1 if is_running else 0
    data["Total"] = apps_total
    data["Running"] = apps_running

    rows = []
    if perms.get("feature:builder"):
        # Builder agent sandboxes and (on kubernetes) kaniko build pods:
        # one more daemon call each, fine in this lazy fragment
        rows.append(ov_container_kind_row("builder agents", "agent"))
        if runtime == "kubernetes":
            rows.append(ov_container_kind_row("kaniko builds", "kaniko"))
    if runtime != "kubernetes":
        rows.append({"label": "app sidecars", "running": sc_running,
                     "total": sc_total})
    rows.append({"label": "litestream sidecars", "running": ls_running,
                 "total": ls_total})
    rows.append({"label": "job runs", "running": job_running,
                 "total": job_total})
    data["Rows"] = rows
    return data


def overview_activity_handler(req):
    # Re-renders the activity tile when the System/All scope chips change
    perms = utils.get_perms()
    scope = utils.query_param(req, "scope") or "system"
    return {"Perms": perms, "Activity": ov_activity_tile(perms, scope)}


def overview_approvals_handler(req):
    # Lazy needs-approval chip on the apps tile: check_approval audits
    # staging per app (cached server-side by app version + binding
    # generation), too slow for the overview first paint
    perms = utils.get_perms()
    data = {"Perms": perms, "Loaded": True}
    if not ov_can(perms, "app:read"):
        return data
    ret = openrun.list_apps(include_internal=True, check_approval=True)
    if ret.error:
        return data
    approval = 0
    for row in build_app_rows(ret.value):
        if row["needs_approval"]:
            approval += 1
    data["Approval"] = approval
    return data


# Replication states that mean attention is needed; everything else
# (healthy, idle, syncing, pending) is a working or quiet-but-fine state
REPL_UNHEALTHY_STATES = ("sidecar_down", "misconfigured", "error")


def overview_replication_handler(req):
    # Lazy binding replication summary: replication_status sweeps the
    # replica store (S3) and the container daemon on a cache miss (30s
    # server-side cache), so the tile shows the free metadata state first
    # and this summary loads here. Only PROD app entries are counted
    # (staged replication state is on the app detail page); the server
    # filters entries per caller (app rows to readable apps via app:read,
    # metadata rows to config:basic_read) - metadata rows are dropped here
    # regardless, they already rendered with the server tile
    perms = utils.get_perms()
    data = {"Perms": perms, "Loaded": True}
    if not ov_can(perms, "app:read"):
        data["Error"] = "requires the app:read or admin permission"
        return data
    ret = openrun.replication_status()
    if ret.error:
        data["Error"] = ret.error
        return data
    # An app is unhealthy when ANY of its prod replication targets is in a
    # bad state (an app has at most one sqlite binding today, but keep the
    # fold safe if that changes)
    app_healthy = {}
    for entry in ret.value:
        if entry["kind"] != "app" or entry.get("env") != "prod":
            continue
        bad = entry["state"] in REPL_UNHEALTHY_STATES
        for app_path in entry.get("app_paths") or []:
            app_healthy[app_path] = app_healthy.get(app_path, True) and not bad
    healthy = len([1 for ok in app_healthy.values() if ok])
    unhealthy = len(app_healthy) - healthy
    data["Total"] = len(app_healthy)
    data["HealthyText"] = "%d app%s healthy" % (healthy, "" if healthy == 1 else "s")
    data["UnhealthyText"] = "%d app%s unhealthy" % (unhealthy, "" if unhealthy == 1 else "s")
    data["HasUnhealthy"] = unhealthy > 0
    return data


def replication_data(req):
    # Replication detail page (/replication, linked from the overview
    # replication tile): one row per PROD app replication target with the
    # full entry detail (state, sidecar, last sync, replica size, target
    # and the per-database file breakdown), searchable by app path. The
    # server filters entries to the caller's readable apps (app:read),
    # same as the overview tile; staged state stays on the app detail page
    query = utils.query_param(req, "query").lower()
    perms = utils.get_perms()
    data = {
        "Title": "Replication",
        "Nav": "overview",
        "Query": query,
        "Perms": perms,
        "Total": 0,
        "Unhealthy": 0,
        "Rows": [],
    }
    if not ov_can(perms, "app:read"):
        data["FlashError"] = "requires the app:read or admin permission"
        return data
    ret = openrun.replication_status()
    if ret.error:
        data["FlashError"] = ret.error
        return data

    # Resolve each prod app's litestream sidecar container for the row
    # drill-down (docker/podman: separate "-ls" containers, prod told apart
    # by app_prd_ in the name; none on kubernetes where the sidecar runs
    # inside the app pod). list_containers is only granted on container
    # feature installs - errors are ignored and the rows render without
    # the drill-down, matching the missing /containers routes
    sidecars = {}
    cont_ret = openrun.list_containers()
    cont_error = cont_ret.error
    if not cont_error:
        for c in cont_ret.value:
            if c["app_path"] and c["name"].endswith("-ls") and "app_prd_" in c["name"]:
                sidecars[c["app_path"]] = c["id"]

    rows = []
    for entry in ret.value:
        if entry["kind"] != "app" or entry.get("env") != "prod":
            continue
        # A target replicates one binding; apps sharing the binding share
        # the entry - render one row per app so the search works by app
        for app_path in entry.get("app_paths") or []:
            data["Total"] += 1
            bad = entry["state"] in REPL_UNHEALTHY_STATES
            if bad:
                data["Unhealthy"] += 1
            if query and query not in app_path.lower():
                continue
            files = []
            for f in entry.get("files") or []:
                files.append({
                    "path": f["path"],
                    "size": utils.human_size(int(f.get("size") or 0)),
                    "last_sync": ov_trim_time(f.get("last_sync") or ""),
                })
            sidecar = entry.get("sidecar_running")
            size = int(entry.get("replica_size") or 0)
            rows.append({
                "container_id": sidecars.get(app_path) or "",
                "app_path": app_path,
                "state": entry["state"],
                "unhealthy": bad,
                "target": entry["target"],
                "sidecar": "" if sidecar == None else ("running" if sidecar else "down"),
                "last_sync": ov_trim_time(entry.get("last_sync") or ""),
                "size": utils.human_size(size) if size else "",
                "files": files,
                "error": entry.get("error") or "",
            })
    # Failing targets first, then by app path
    data["Rows"] = sorted(rows, key=lambda r: (not r["unhealthy"], r["app_path"]))
    return data


# ---------- Apps ----------


def build_app_rows(all_apps):
    # Folds staging entries into their main app's row and returns the row
    # dicts rendered by the shared app_table template. all_apps must come
    # from list_apps with include_internal=True
    staging_by_main = {}
    for entry in all_apps:
        if entry["is_stage"]:
            staging_by_main[entry["main_app"]] = entry

    rows = []
    for entry in all_apps:
        if entry["main_app"]:
            continue

        stage = staging_by_main.get(entry["id"])
        # The staging app carries the most recent sync state (prod picks it
        # up on promote); fall back to the prod app's value. get() keeps this
        # working against servers older than the applied_sync_id field
        sync_id = (stage.get("applied_sync_id", "") if stage else "") or entry.get("applied_sync_id", "")

        staging = None
        if stage:
            staging = {
                "version": stage["version"],
                "git_sha": utils.short_sha(stage["git_sha"]),
                "git_message": stage["git_message"],
                # staging has a version prod does not have yet
                "ahead": stage["version_mismatch"],
            }

        rows.append({
            "name": entry["name"],
            "path": entry["path"],
            "url": entry["url"],
            "auth": entry["auth"],
            "is_dev": entry.get("is_dev") or False,
            # Only set when list_apps ran with check_approval=True (apps page);
            # the backend mirrors the staging app's audit onto the main app
            "needs_approval": entry.get("needs_approval") or False,
            "is_git": bool(entry["git_branch"]),
            # Declarative means a sync source last applied the app. Git
            # presence is not the signal: image/proxy spec apps have no git
            "sync_id": sync_id,
            "is_declarative": bool(sync_id),
            # "-" is the placeholder for apps with no source (image/proxy specs)
            "source": entry["source"] if entry["source"] != "-" else "",
            "source_url": entry["source_url"],
            "git_branch": entry["git_branch"],
            "spec": entry.get("spec") or "",
            "version": entry["version"],
            "git_sha": utils.short_sha(entry["git_sha"]),
            "git_message": entry["git_message"],
            "staging": staging,
            "created_by": entry.get("created_by") or "",
            "update_time": entry.get("update_time") or "",
            "update_user": entry.get("update_user") or "",
        })
    return rows


def apps_data(req):
    # Apps list page: apps grouped by their managing sync, plus unmanaged.
    # The promote/approval tabs show the apps waiting on that action as a
    # flat list, with the row action switched to Promote/Approve
    query = utils.query_param(req, "query")
    filter = utils.query_param(req, "filter")  # "", "declarative" or "imperative"
    tab = utils.query_param(req, "tab")  # "", "promote" or "approval"

    # include_internal picks up staging/preview apps; staging entries are
    # folded into their main app's row instead of being listed separately.
    # check_approval adds the needs_approval flag (cached server-side).
    # A search containing ":", starting with "/", or equal to "all"
    # switches to app path GLOB matching (domain:path form, e.g.
    # "example.com:/**", "/utils/**", "all" - the same globs the CLI
    # takes); anything else is a substring search over
    # name/path/source/user
    stripped = query.strip()
    glob = ""
    if ":" in stripped or stripped.startswith("/") or stripped.lower() == "all":
        glob = stripped
    list_ret = openrun.list_apps(query="" if glob else query, path=glob,
                                 include_internal=True, check_approval=True)
    list_error = list_ret.error
    all_apps = list_ret.value if not list_error else []

    # Sync entries the user can read. Apps whose sync entry is not visible
    # (no sync:read) are shown in the unmanaged section instead
    syncs = {}
    sync_ret = openrun.list_sync()
    for entry in (sync_ret.value if not sync_ret.error else []):
        syncs[entry["id"]] = {
            "id": entry["id"],
            "repo": entry["path"],
            "branch": entry["metadata"]["git_branch"],
            "state": entry["status"]["state"],  # Enabled / Disabled / Failing
            "last_exec": utils.nonzero_time(entry["status"]["last_execution_time"]),
        }

    grouped = {}  # sync id -> app rows, for apps last applied by a live sync
    unmanaged = []  # created/last updated imperatively
    tab_apps = []  # rows for the active promote/approval tab
    total = 0
    declarative_count = 0
    promote_count = 0
    approval_count = 0
    for app in build_app_rows(all_apps):
        total += 1
        if app["is_declarative"]:
            declarative_count += 1
        if (filter == "declarative" and not app["is_declarative"]) or \
           (filter == "imperative" and app["is_declarative"]):
            continue

        # The tab badge counts follow the active declarative/imperative
        # filter, matching what the tab tables list
        if app["staging"] and app["staging"]["ahead"]:
            promote_count += 1
        if app["needs_approval"]:
            approval_count += 1

        if tab == "promote" and app["staging"] and app["staging"]["ahead"]:
            tab_apps.append(app)
        elif tab == "approval" and app["needs_approval"]:
            tab_apps.append(app)

        if app["sync_id"] and app["sync_id"] in syncs:
            grouped.setdefault(app["sync_id"], []).append(app)
        else:
            unmanaged.append(app)

    # Most recently updated apps first, groups ordered by their most
    # recently updated app
    groups = []
    for sync_id in grouped:
        apps = utils.sort_recent(grouped[sync_id], "update_time", "path")
        groups.append({
            "sync": syncs[sync_id],
            "apps": apps,
            "newest": apps[0]["update_time"] if apps else "",
            "repo": syncs[sync_id]["repo"],
        })

    data = {
        "Title": "Apps",
        "Nav": "apps",
        "Query": query,
        "Filter": filter,
        "Tab": tab,
        "TabApps": utils.sort_recent(tab_apps, "update_time", "path"),
        "Groups": utils.sort_recent(groups, "newest", "repo"),
        "Unmanaged": utils.sort_recent(unmanaged, "update_time", "path"),
        "Total": total,
        "DeclarativeCount": declarative_count,
        "ImperativeCount": total - declarative_count,
        "PromoteCount": promote_count,
        "ApprovalCount": approval_count,
        "Perms": utils.get_perms(),
    }
    if list_error:
        # A malformed glob search (":" queries) renders the empty list with
        # the parse error instead of crashing to the error page
        data["FlashError"] = "Invalid app path glob: %s" % list_error
    return data


def load_versions(path):
    # Returns the version list for the app at path, newest first
    ret = openrun.list_versions(path)
    if ret.error:
        return [], ret.error

    versions = []
    for entry in ret.value["versions"] or []:
        vm = (entry.get("Metadata") or {}).get("version_metadata") or {}
        versions.append({
            "version": entry["Version"],
            "previous": entry.get("PreviousVersion") or 0,
            "active": entry.get("Active") or False,
            "user": entry.get("UserId") or "",
            "create_time": utils.nonzero_time(entry.get("CreateTime")),
            "git_sha": utils.short_sha(vm.get("git_commit") or ""),
            "git_message": vm.get("git_message") or "",
            "git_branch": vm.get("git_branch") or "",
        })
    return sorted(versions, key=lambda v: v["version"], reverse=True), ""


def resolve_env_path(path, env):
    # The prod app path is the external identifier; staging actions resolve
    # the linked staging app's path through get_app
    if env == "stage":
        app_ret = openrun.get_app(path)
        if not app_ret.error:
            return app_ret.value["stage_path"]
    return path


def version_options(path, app):
    # The env-qualified version choices for the Compare/Files selects:
    # staging versions first, newest first. Values are "env:version" specs;
    # also returns the active version per env
    options = []
    stage_versions, stage_err = load_versions(app["stage_path"])
    prod_versions, prod_err = load_versions(path)
    active = {"stage": 0, "prod": 0}
    for v in stage_versions:
        version = int(v["version"])
        if v["active"]:
            active["stage"] = version
        options.append({
            "value": "stage:%d" % version,
            "label": "v%d · staging%s" % (version, " (active)" if v["active"] else ""),
        })
    for v in prod_versions:
        version = int(v["version"])
        if v["active"]:
            active["prod"] = version
        options.append({
            "value": "prod:%d" % version,
            "label": "v%d · prod%s" % (version, " (active)" if v["active"] else ""),
        })
    return options, active, (stage_err or prod_err)


def version_spec_label(spec):
    # Display label for an "env:version" spec
    env, _, version = spec.partition(":")
    env_label = "staging" if env == "stage" else "prod"
    return ("v%s · %s" % (version, env_label)) if version else env_label


def diff_rows(diff):
    # Normalize export_app_diff rows for the template: line number 0 means
    # no line on that side (spacer row)
    rows = []
    for row in diff["rows"] or []:
        left_line = int(row["left_line"])
        right_line = int(row["right_line"])
        rows.append({
            "kind": row["kind"],
            "left_line": ("%d" % left_line) if left_line else "",
            "left_text": row["left_text"],
            "right_line": ("%d" % right_line) if right_line else "",
            "right_text": row["right_text"],
        })
    return rows


def export_lines(text):
    # Split an export output into display lines for the numbered pane
    return text.rstrip("\n").split("\n")


def detail_config_data(data, req):
    # Config tab: the prod export output. When staging runs a different
    # version, the view automatically becomes the prod <-> staging diff -
    # the one-look answer to what is pending promotion
    app = data["App"]
    path = data["Path"]
    cfg = {"Error": "", "Single": True, "Label": "", "Badge": "", "Lines": [],
           "Rows": [], "Changed": 0, "LeftLabel": "", "RightLabel": ""}
    data["Config"] = cfg

    if app["is_dev"]:
        ret = openrun.export_app(path)
        error = ret.error
        if error:
            cfg["Error"] = error
            return
        cfg["Label"] = "Dev app declaration"
        cfg["Lines"] = export_lines(ret.value)
        return

    prod_version = int(app["version"])
    if not app["staged_changes"]:
        ret = openrun.export_app(path)
        error = ret.error
        if error:
            cfg["Error"] = error
            return
        cfg["Label"] = "Prod · v%d · staging in sync" % prod_version
        cfg["Lines"] = export_lines(ret.value)
        return

    _, active, _ = version_options(path, app)
    ret = openrun.export_app_diff(path, "prod:", "stage:")
    error = ret.error
    if error:
        cfg["Error"] = error
        return
    stage_label = ("v%d" % active["stage"]) if active["stage"] else "version"
    if not int(ret.value["changed"]):
        # Staging is ahead but the declarative config is identical: the
        # pending change is source-only. Two identical panes would read as
        # a bug, so show the single config with an explaining badge
        prod_ret = openrun.export_app(path)
        prod_error = prod_ret.error
        if prod_error:
            cfg["Error"] = prod_error
            return
        cfg["Label"] = "Prod · v%d" % prod_version
        cfg["Badge"] = "staging %s is ahead - source changed, config identical" % stage_label
        cfg["Lines"] = export_lines(prod_ret.value)
        return
    cfg["Single"] = False
    cfg["Label"] = "Prod · v%d" % prod_version
    cfg["Badge"] = "staging %s differs - showing changes" % stage_label
    cfg["LeftLabel"] = "prod · v%d" % prod_version
    cfg["RightLabel"] = ("staging · v%d" % active["stage"]) if active["stage"] else "staging"
    cfg["Rows"] = diff_rows(ret.value)
    cfg["Changed"] = int(ret.value["changed"])


def detail_compare_data(data, req):
    # Compare tab: any two versions side by side as export outputs. Defaults
    # to prod active vs staging active when they differ, else the previous
    # prod version vs the active one
    app = data["App"]
    path = data["Path"]
    cmp = {"Error": "", "Options": [], "From": "", "To": "", "Rows": [],
           "Changed": 0, "Same": False, "LeftLabel": "", "RightLabel": ""}
    data["Compare"] = cmp

    options, active, err = version_options(path, app)
    if err and not options:
        cmp["Error"] = err
        return
    cmp["Options"] = options

    from_spec = utils.query_param(req, "from")
    to_spec = utils.query_param(req, "to")
    if not from_spec or not to_spec:
        if app["staged_changes"] and active["stage"]:
            from_spec = "prod:%d" % active["prod"]
            to_spec = "stage:%d" % active["stage"]
        else:
            prod_opts = [o["value"] for o in options if o["value"].startswith("prod:")]
            to_spec = prod_opts[0] if prod_opts else ""
            from_spec = prod_opts[1] if len(prod_opts) > 1 else to_spec
    cmp["From"] = from_spec
    cmp["To"] = to_spec
    if not from_spec or not to_spec:
        cmp["Error"] = "No versions to compare yet"
        return

    ret = openrun.export_app_diff(path, from_spec, to_spec)
    error = ret.error
    if error:
        cmp["Error"] = error
        return
    cmp["Rows"] = diff_rows(ret.value)
    cmp["Changed"] = int(ret.value["changed"])
    cmp["Same"] = cmp["Changed"] == 0
    cmp["LeftLabel"] = version_spec_label(from_spec)
    cmp["RightLabel"] = version_spec_label(to_spec)


def detail_files_data(data, req):
    # Files tab: one version's file tree with the builder-files explorer.
    # Defaults to the staging active version when it differs from prod
    app = data["App"]
    path = data["Path"]
    ft = {"Error": "", "Options": [], "Selected": "", "Label": "", "Tree": [],
          "Count": 0, "TotalSize": "", "DownloadUrl": "", "Endpoint": ""}
    data["FilesTab"] = ft

    options, active, err = version_options(path, app)
    if err and not options:
        ft["Error"] = err
        return
    ft["Options"] = options

    sel = utils.query_param(req, "v")
    if not sel:
        if app["staged_changes"] and active["stage"]:
            sel = "stage:%d" % active["stage"]
        elif active["prod"]:
            sel = "prod:%d" % active["prod"]
    ft["Selected"] = sel
    env, _, version = sel.partition(":")
    if env not in ("prod", "stage") or not version:
        ft["Error"] = "No version selected"
        return
    ft["Label"] = version_spec_label(sel)

    resolved = resolve_env_path(path, env)
    ret = openrun.list_version_files(resolved, version=version)
    error = ret.error
    if error:
        ft["Error"] = error
        return
    names = []
    total = 0
    for entry in ret.value["files"] or []:
        names.append(entry["Name"])
        total += int(entry["Size"])
    ft["Tree"] = build_file_tree(sorted(names))
    ft["Count"] = len(names)
    ft["TotalSize"] = utils.human_size(total)
    ft["DownloadUrl"] = "%s/apps/files/download?path=%s&version=%s&env=%s" % (
        req.AppPath, path, version, env)
    ft["Endpoint"] = "%s/apps/version_file?app=%s&env=%s&version=%s" % (
        req.AppPath, path, env, version)


def apps_detail_config_download_handler(req):
    # GET: the prod export output as an apply-ready .star download
    path = utils.query_param(req, "path")
    ret = openrun.export_app(path)
    if ret.error:
        data = apps_detail_data(req)
        data["FlashError"] = "Export failed: %s" % ret.error
        return data
    header = ("# Declarative config for %s, generated by the OpenRun console\n" +
              "# Apply on an openrun instance with: openrun apply --approve <file>\n\n") % path
    name = path.strip("/").replace("/", "_").replace(":", "_") or "app"
    return ace.response(header + ret.value, download=name + ".star",
                        content_type="text/plain")


def apps_version_file_handler(req):
    # TEXT route: raw content of one version file, rendered by the Files
    # tab's <builder-files> viewer (client side syntax highlighting). The
    # component sets the path query parameter to the selected file
    app_path = utils.query_param(req, "app").strip()
    env = utils.query_param(req, "env").strip() or "prod"
    version = utils.query_param(req, "version").strip()
    name = utils.query_param(req, "path").strip()
    ret = openrun.get_version_file(resolve_env_path(app_path, env),
                                   version=version, name=name)
    if ret.error:
        return "error: %s" % ret.error
    return ret.value


ENV_ORDER = {"prod": "0", "stage": "1", "preview": "2", "dev": "3"}


def container_time_tip(c):
    # Tooltip for a container row: when it started (running) or when it
    # started and stopped (exited). Times come from list_containers
    # times=True; empty when the runtime does not report them
    started = short_ts(c.get("started_at"))
    tip = c["name"]
    if started:
        tip += " · started " + started
    if c["state"] != "running":
        stopped = short_ts(c.get("finished_at"))
        if stopped:
            tip += " · stopped " + stopped
    return tip


def short_ts(t):
    # RFC3339 runtime timestamps (UTC, nanosecond precision) trimmed to
    # seconds for tooltips: "2026-08-23 09:12:33 UTC"
    t = t or ""
    if len(t) < 19 or t.startswith("0001-"):
        return ""
    return t[:10] + " " + t[11:19] + " UTC"


def env_container_rows(app_conts, env):
    # Rows of one environment's containers table on the app detail page:
    # per type (the app container, each sidecar by name) the most recently
    # started running container and the most recently stopped one, if any.
    # On docker/podman sidecars are separate containers (role label); on
    # kubernetes they run inside the app pods - one pod status call per
    # running pod resolves their states, linking to the pod detail
    groups = {}
    order = []

    def add(key, c):
        if key not in groups:
            groups[key] = []
            order.append(key)
        groups[key].append(c)

    for c in app_conts:
        if c["env"] != env or c["name"].endswith("-ls"):
            continue
        if c.get("job"):
            # Job run containers are listed on the Jobs tab, per run
            continue
        if c.get("role") == "sidecar":
            add("sidecar:" + c["sidecar"], c)
            continue
        add("app", c)
        if c.get("sidecars") and c["state"] == "running":
            status = openrun.container_kubernetes_status(c["id"])
            error = status.error
            states = {}
            if not error:
                for pc in status.value.get("containers") or []:
                    if pc.get("sidecar"):
                        states[pc["name"]] = "running" if pc["state"] == "running" and pc.get("ready", True) else pc["state"]
            for name in c["sidecars"]:
                e = dict(c.items())
                e["role"] = "sidecar"
                e["sidecar"] = name
                e["state"] = states.get(name) or "unknown"
                add("sidecar:" + name, e)

    rows = []
    for key in sorted(order, key=lambda k: "" if k == "app" else k):
        entries = groups[key]
        running = sorted([c for c in entries if c["state"] == "running"],
                         key=lambda c: c.get("started_at") or c.get("created_at") or "", reverse=True)
        stopped = sorted([c for c in entries if c["state"] != "running"],
                         key=lambda c: (c.get("finished_at") or "") + (c.get("created_at") or ""), reverse=True)
        for c in running[:1] + stopped[:1]:
            rows.append({
                "id": c["id"],
                "name": c["name"],
                "kind": "sidecar" if c.get("role") == "sidecar" else "app",
                "sidecar": c.get("sidecar") or "",
                "state": c["state"],
                # Last update: the start for running containers, the stop
                # for stopped ones (RFC3339, rendered relative)
                "last": (c.get("started_at") if c["state"] == "running" else c.get("finished_at")) or c.get("created_at") or "",
                "tip": container_time_tip(c),
            })
    return rows


def app_container_sort_key(entry):
    # Sort containers: running first, then prod/stage/preview/dev, then name
    running = "0" if entry["state"] == "running" else "1"
    return running + ENV_ORDER.get(entry["env"], "9") + entry["name"]


def apps_detail_data(req):
    # App detail page, a tabbed inspector: Overview (summary card), Config
    # (export output, auto prod<->staging diff), Compare (any two versions
    # as export diffs), Files (version file explorer) and, for apps with
    # jobs, Jobs (effective jobs, run history, run output). Only the active
    # tab's data is built; the Overview tab's per-environment panel
    # (params, permissions, containers, versions) loads lazily via the
    # envpanel fragment, with the env selection url-persisted (?env=)
    path = utils.query_param(req, "path")
    tab = utils.query_param(req, "tab")
    if tab not in ("config", "compare", "files", "jobs"):
        tab = "overview"
    data = {
        "Title": "App detail",
        "Nav": "apps",
        "Path": path,
        "Tab": tab,
        "Error": "",
        "App": None,
        # Set after a staging-only reload/update, prompts for promotion
        "AskPromote": utils.query_param(req, "staged"),
        # App permissions evaluated against this app, including the owner rule
        "Perms": utils.get_perms(path),
        "HelpUrl": utils.docs_link("/docs/applications/lifecycle/"),
    }

    ret = openrun.get_app(path)
    if ret.error:
        data["Error"] = ret.error
        return data

    app = ret.value
    data["App"] = app

    # The environment shown by the Overview tab's lazy panel: prod by
    # default, url-persisted (the env tabs push ?env=, the version switch
    # posts carry it); dev apps have a single dev view
    env = utils.query_param(req, "env")
    if app["is_dev"]:
        env = "dev"
    elif env not in ("prod", "stage"):
        env = "prod"
    data["PanelEnv"] = env

    if app["is_dev"] and tab in ("compare", "files"):
        # Dev apps have no versions; those tabs render disabled
        tab = "overview"
        data["Tab"] = tab
    if tab in ("config", "compare", "files") and not data["Perms"].get("app:read_detail"):
        # app:read without app:read_detail: the config, version and file
        # views are not offered (the server denies their APIs)
        tab = "overview"
        data["Tab"] = tab
    if tab == "config":
        detail_config_data(data, req)
        return data
    if tab == "compare":
        detail_compare_data(data, req)
        return data
    if tab == "files":
        detail_files_data(data, req)
        return data
    if tab == "jobs":
        detail_jobs_data(data, req)
        return data

    # Resolve the sync entry which manages this app, if any
    if app["applied_sync_id"]:
        sync_ret = openrun.list_sync()
        for entry in (sync_ret.value if not sync_ret.error else []):
            if entry["id"] == app["applied_sync_id"]:
                data["Sync"] = {
                    "repo": entry["path"],
                    "branch": entry["metadata"]["git_branch"],
                }

    return data


# ---------- App detail: Jobs tab ----------

JOB_RUN_LIMIT = 50


def job_stage_of(app_id):
    # prod / stage / preview / dev from a run's app id prefix (the Go
    # JobRun.Stage() method is not carried over the plugin boundary)
    if app_id.startswith("app_stg_"):
        return "stage"
    if app_id.startswith("app_pre_"):
        return "preview"
    if app_id.startswith("app_dev_"):
        return "dev"
    return "prod"


def job_duration(started, ended):
    # "1m 5s" style run duration; empty while the run is active or when the
    # times are unknown. The plugin passes time.time values, whose
    # difference is a duration
    if not started or not ended:
        return ""
    secs = int((ended - started).seconds)
    if secs < 0:
        return ""
    if secs < 60:
        return "%ds" % secs
    if secs < 3600:
        return "%dm %ds" % (secs // 60, secs % 60)
    return "%dh %dm" % (secs // 3600, (secs % 3600) // 60)


def job_run_row(run):
    # One run of the runs table / the logs pane header. started_at stays a
    # time.time value (the templates render it relative); the zero ended_at
    # of an active run maps to "" and leaves the duration empty
    started = utils.nonzero_time(run.get("started_at"))
    ended = utils.nonzero_time(run.get("ended_at"))
    exit_code = run.get("exit_code")
    return {
        "id": run["id"],
        "job": run["job_name"],
        "trigger": run.get("trigger") or "manual",
        "stage": job_stage_of(run.get("app_id") or ""),
        "status": run.get("status") or "",
        "active": run.get("status") == "running",
        "actor": run.get("actor") or "",
        "started_at": started,
        "duration": job_duration(started, ended),
        "exit_code": exit_code,
        "has_exit_code": exit_code != None,
        "message": run.get("message") or "",
        "image": run.get("image") or "",
        "container_name": run.get("container_name") or "",
        "args": run.get("args") or {},
        "forced": run.get("forced") or False,
    }


def job_row(info):
    # One effective job of the Jobs table: the spec summary plus the next
    # scheduled run and the last run. Rows of the stage instance appear only
    # when the staged spec differs from prod's (list_jobs semantics); their
    # run dialog targets the staging instance
    spec = info.get("spec") or {}
    trigger = spec.get("trigger") or {}
    ttype = trigger.get("type") or "manual"
    enabled = spec.get("enabled")
    if enabled == None:
        enabled = True
    image = spec.get("image") or ""
    if image.startswith("image:"):
        image = image[len("image:"):]
    argv = list(spec.get("command") or []) + list(spec.get("args") or [])
    last = info.get("last_run")
    return {
        "name": spec.get("name") or "",
        "description": spec.get("description") or "",
        "stage": info.get("stage") or "prod",
        "origin": info.get("origin") or "",
        "trigger": ttype,
        "schedule": trigger.get("schedule") or "",
        "timezone": trigger.get("timezone") or "",
        "enabled": enabled,
        "timeout": spec.get("timeout") or "1h",
        "is_run": bool(spec.get("run")),
        "executor": spec.get("run") or " ".join(argv),
        "shell": spec.get("shell") or False,
        "image": image,
        "params": list(spec.get("params") or []),
        "next_run": info.get("next_run") or "",
        "last_run": job_run_row(last) if last else None,
        "warnings": list(info.get("warnings") or []),
    }


def list_job_rows(path):
    # The Jobs table rows of an app; returns (rows, error)
    ret = openrun.list_jobs(path)
    error = ret.error
    if error:
        return [], error
    return [job_row(info) for info in ret.value.get("jobs") or []], ""


def job_names(rows):
    # Distinct job names of the Jobs table, for the runs filter select
    names = []
    for row in rows:
        if row["name"] and row["name"] not in names:
            names.append(row["name"])
    return names


def job_runs_data(req, path, perms, names):
    # The runs table: newest first across the prod and stage instances,
    # filtered by ?job= and ?status=. Command runs link to their container
    # detail when the containers feature is on and the container still
    # exists (matched by the run id label; one list_containers call per
    # render, skipped when no listed run had a container - run jobs)
    job = utils.query_param(req, "job")
    status = utils.query_param(req, "status")
    runs = {"Job": job, "Status": status, "Names": names, "Rows": [], "Error": "", "Active": False}
    ret = openrun.list_job_runs(path, job=job, status=status, limit=JOB_RUN_LIMIT)
    error = ret.error
    if error:
        runs["Error"] = error
        return runs
    rows = [job_run_row(run) for run in ret.value.get("runs") or []]
    containers = {}
    if perms.get("feature:container") and any([row["container_name"] for row in rows]):
        cont_ret = openrun.list_containers(path=path)
        cont_error = cont_ret.error
        if not cont_error:
            for c in cont_ret.value:
                if c.get("job_run"):
                    containers[c["job_run"]] = c["id"]
    for row in rows:
        row["container_id"] = containers.get(row["id"]) or ""
        if row["active"]:
            runs["Active"] = True
    runs["Rows"] = rows
    return runs


def job_logs_data(run_id):
    # The logs pane of one run: the run header plus its output (read from
    # the container while it exists; a run job's result message otherwise)
    ret = openrun.job_logs(run_id)
    error = ret.error
    if error:
        return {"RunId": run_id, "Error": error, "Run": None, "Logs": ""}
    return {"RunId": run_id, "Error": "", "Run": job_run_row(ret.value["run"]),
            "Logs": ret.value.get("logs") or ""}


def detail_jobs_data(data, req):
    # Jobs tab: the effective jobs (prod or dev instance, plus the stage
    # instance's jobs that differ while a change is staged), the run
    # history of both instances and, with ?run=, the logs pane of one run
    # (the container detail page of a job container links here)
    path = data["Path"]
    rows, error = list_job_rows(path)
    data["Jobs"] = {"Rows": rows, "Error": error}
    if not data["Perms"].get("app:read_detail"):
        # app:read lists the jobs; the run history and output need
        # app:read_detail (the pane omits those sections)
        data["Runs"] = None
        data["Logs"] = None
        return
    data["Runs"] = job_runs_data(req, path, data["Perms"], job_names(rows))
    run_id = utils.query_param(req, "run")
    data["Logs"] = job_logs_data(run_id) if run_id else None


def apps_detail_jobs_runs_handler(req):
    # Runs table fragment: the job/status filter selects and the poll while
    # a run is active re-render #job-runs through here
    path = utils.query_param(req, "path")
    perms = utils.get_perms(path)
    rows, error = list_job_rows(path)
    if error:
        return {"Error": error}
    return {"Path": path, "Perms": perms, "Runs": job_runs_data(req, path, perms, job_names(rows))}


def apps_detail_jobs_logs_handler(req):
    # Logs pane fragment for one run (?run=), swapped into #job-logs
    return {"Logs": job_logs_data(utils.query_param(req, "run"))}


def apps_detail_jobs_run_handler(req):
    # POST: start a manual run from the Jobs tab run dialog. The dialog
    # posts the job, the target instance (stage checkbox), force and the
    # run arguments as arg_<param> fields (arg_names lists the params; an
    # empty value keeps the app's configured param value). The run is not
    # waited for: the runs table polls while it is active
    path, error_data = require_app_path(req, apps_detail_data)
    if error_data:
        return error_data
    job = utils.query_param(req, "job").strip()
    if not job:
        data = apps_detail_data(req)
        data["FlashError"] = "Job name is required"
        return data
    args = {}
    for name in utils.query_param_list(req, "arg_names"):
        value = utils.query_param(req, "arg_" + name)
        if value != "":
            args[name] = value
    ret = openrun_admin.run_job(path, job, stage=bool(utils.query_param(req, "stage")),
                                force=bool(utils.query_param(req, "force")), args=args)
    error = ret.error
    data = apps_detail_data(req)
    if error:
        data["FlashError"] = "Run of job %s failed: %s" % (job, error)
    else:
        run = ret.value["run"]
        data["Flash"] = "Started run %s of job %s on %s" % (run["id"], job, job_stage_of(run.get("app_id") or ""))
    return data


def apps_detail_jobs_cancel_handler(req):
    # POST: cancel an active run (executing on this node) from the runs table
    run_id = utils.query_param(req, "run").strip()
    if not run_id:
        data = apps_detail_data(req)
        data["FlashError"] = "Run id is required"
        return data
    ret = openrun_admin.cancel_job(run_id)
    error = ret.error
    return utils.flash_result(apps_detail_data(req), error,
                              "Cancel requested for run %s" % run_id, "Cancel failed")


def apps_detail_envpanel_handler(req):
    # Lazy per-environment panel on the detail page's Overview tab: the
    # parameters, requested permissions, containers and versions of one
    # environment. env=prod (default) or stage; dev apps render a single
    # dev view. Loaded via hx-trigger=load so the page's first paint stays
    # cheap; the panel's env tabs push ?env= on the page url
    path = utils.query_param(req, "path")
    env = utils.query_param(req, "env")
    if env not in ("prod", "stage"):
        env = "prod"
    data = {
        "Path": path,
        "Env": env,
        "Error": "",
        "Perms": utils.get_perms(path),
    }
    if not data["Perms"].get("app:read_detail"):
        # The panel is not rendered for app:read only callers (the page
        # omits its loader); the params, audit, containers and versions
        # calls below would each be denied
        data["Error"] = "requires the app:read_detail permission"
        return data

    ret = openrun.get_app(path)
    if ret.error:
        data["Error"] = ret.error
        return data
    app = ret.value
    data["App"] = app
    if app["is_dev"]:
        env = "dev"
        data["Env"] = env

    # Parameters of this environment's app: staging params can differ from
    # prod until the staged changes are promoted. EnvVersion is the env's
    # active version, named by the panel's indicator line
    params = app["params"]
    data["EnvVersion"] = app["version"]
    if env == "stage":
        stage_ret = openrun.get_app(app["stage_path"], include_internal=True)
        if not stage_ret.error:
            params = stage_ret.value["params"]
            data["EnvVersion"] = stage_ret.value["version"]
    data["ParamsText"] = utils.params_to_text(params)

    # Needs-approval alert: approvals apply to staging first, so the alert
    # uses the staging audit (cached server-side via check_approval) and
    # shows on both environment tabs
    la_ret = openrun.list_apps(path=path, check_approval=True)
    if not la_ret.error:
        for row in la_ret.value:
            if row["path"] == path:
                data["NeedsApproval"] = row.get("needs_approval") or False

    # The plugin permissions requested by this environment's app code and
    # whether they are approved for it (dev apps are audited directly)
    if env == "dev":
        audit_ret = openrun.audit_app(path)
    else:
        audit_ret = openrun.audit_app(path, env=env)
    if audit_ret.error:
        data["AuditError"] = audit_ret.error
    else:
        audit = audit_ret.value
        data["Audit"] = utils.review_from_dryrun({"approve_results": [audit]})
        data["EnvNeedsApproval"] = audit.get("needs_approval") or False

    # Containers of this environment: per type (app container, each
    # sidecar name) the running container and the last stopped one. The
    # litestream sidecars (-ls suffix) are excluded: the Overview tab's
    # Replication row links to them. The path filter makes the call
    # allowed with app:read_detail on the app alone (no container:read)
    cont_ret = openrun.list_containers(path=path, times=True)
    if not cont_ret.error:
        app_conts = [c for c in cont_ret.value if c["app_path"] == path]
        data["Containers"] = env_container_rows(app_conts, env)

    if env != "dev":
        # Dev apps serve directly from disk, no versions are tracked
        versions, err = load_versions(path if env == "prod" else app["stage_path"])
        data["Versions"] = versions
        data["VersionsError"] = err
    return data


def apps_detail_replication_handler(req):
    # Lazy replication status for the app detail Overview card:
    # replication_status sweeps the replica store on a server cache miss,
    # so it must not block the detail page's first paint. Renders nothing
    # when the app has no replicated sqlite binding
    path = utils.query_param(req, "path")
    data = {"Perms": utils.get_perms(path), "Loaded": True, "Path": path}
    ret = openrun.get_app(path)
    error = ret.error
    if error:
        # No access or app gone: render nothing (the page itself already
        # surfaced the error)
        return data
    stage_path = ret.value.get("stage_path") or ""

    rs = openrun.replication_status()
    error = rs.error
    if error:
        data["Error"] = error
        return data

    # The litestream sidecars are separate labeled containers on docker/
    # podman (-ls name suffix, prod vs staged told apart by the app id in
    # the container name); each status chip links to its sidecar's
    # container detail when one exists (on kubernetes the sidecar runs
    # inside the app pod, so there is no separate container to link).
    # The path filter is allowed with app:read_detail on the app alone
    sidecars = {}
    cont_ret = openrun.list_containers(path=path)
    cont_error = cont_ret.error
    if not cont_error:
        for c in cont_ret.value:
            if c["app_path"] != path or not c["name"].endswith("-ls"):
                continue
            sidecars["prod" if "app_prd_" in c["name"] else "staged"] = c["id"]

    rows = []
    for entry in rs.value:
        if entry["kind"] != "app":
            continue
        env = entry.get("env") or ""
        paths = entry.get("app_paths") or []
        if (env == "prod" and path in paths) or \
           (env == "staged" and stage_path and stage_path in paths):
            rows.append({
                "env": "prod" if env == "prod" else "stage",
                "state": entry["state"],
                "last_sync": ov_trim_time(entry.get("last_sync") or ""),
                "error": entry.get("error") or "",
                "container_id": sidecars.get(env) or "",
            })
    # prod first, matching the containers chip order
    data["Rows"] = sorted(rows, key=lambda r: r["env"] != "prod")
    return data


def apps_switch_handler(req):
    # POST: switch the active version for prod or staging
    path = utils.query_param(req, "path")
    env = utils.query_param(req, "env") or "prod"
    version = utils.query_param(req, "version")

    ret = openrun_admin.switch_version(resolve_env_path(path, env), version)
    error = ret.error
    return utils.flash_result(apps_detail_data(req), error,
                        "Switched %s to v%s" % (env, version), "Version switch failed")


def require_app_path(req, data_fn):
    # The app write plugin APIs take path globs; a missing path must not
    # silently become an empty glob (which would match every app). Returns
    # (path, None) or ("", error page data)
    path = utils.query_param(req, "path").strip()
    if not path:
        data = data_fn(req)
        data["FlashError"] = "App path is required"
        return "", data
    return path, None


def promote_app_result(req, data_fn, path):
    # Promote the staging app to prod and re-render the page via data_fn
    # with the result flash. Shared by the detail page, the apps list
    # pending-promotion tab and the builder session page
    ret = openrun_admin.promote_apps(path)
    error = ret.error
    data = data_fn(req)
    if error:
        data["FlashError"] = "Promote failed: %s" % error
    elif not ret.value.get("promote_results"):
        data["Flash"] = "Nothing to promote, prod matches staging"
    else:
        data["Flash"] = "Promoted %s to prod" % path
    return data


def apps_promote_handler(req):
    # POST: promote the staging app to prod
    path, error_data = require_app_path(req, apps_detail_data)
    if error_data:
        return error_data
    return promote_app_result(req, apps_detail_data, path)


def approve_app_result(req, data_fn, path):
    # Approve the pending plugin permissions (applies to the staging app, or
    # directly for dev apps) and re-render the page via data_fn
    ret = openrun_admin.approve_apps(path, promote=False)
    error = ret.error
    data = data_fn(req)
    if error:
        data["FlashError"] = "Approve failed: %s" % error
    else:
        data["Flash"] = "Approved pending permissions for %s" % path
    return data


def apps_approve_handler(req):
    # POST: approve the requested plugin permissions; promotion is asked as
    # the next step
    path, error_data = require_app_path(req, apps_detail_data)
    if error_data:
        return error_data
    data = approve_app_result(req, apps_detail_data, path)
    if not data.get("FlashError"):
        data["AskPromote"] = "approve"
    return data


def reload_app_options(req, path):
    # Reload with the reload-dialog options. The dialog disables the approve
    # checkbox and promotion radios when the user lacks the permission, and
    # disabled inputs are not submitted: missing params mean no approval and
    # no promotion (the approve/promote flags hard-fail server-side without
    # the permission). promote_mode verify reloads app containers to verify
    # staging before promoting; a failed verification rolls everything back
    mode = utils.query_param(req, "promote_mode") or "none"
    return openrun_admin.reload_apps(path,
                                   approve=bool(utils.query_param(req, "approve")),
                                   promote=mode == "verify" or mode == "noverify",
                                   verify=mode == "verify")


def apps_detail_reload_handler(req):
    # POST: reload from source with the dialog options, staying on the
    # detail page. A staging-only reload prompts for promotion as the next
    # step; a promoting reload is done in one go
    path, error_data = require_app_path(req, apps_detail_data)
    if error_data:
        return error_data
    ret = reload_app_options(req, path)
    error = ret.error
    data = apps_detail_data(req)
    if error:
        data["FlashError"] = "Reload failed: %s" % error
    elif ret.value.get("skipped_results") and not ret.value.get("reload_results"):
        data["Flash"] = "%s is already up to date" % path
    elif ret.value.get("promote_results"):
        data["Flash"] = "Reloaded and promoted to prod"
    elif data.get("App") and data["App"].get("is_dev"):
        data["Flash"] = "Reloaded from source"
    else:
        data["Flash"] = "Staging reloaded from source"
        data["AskPromote"] = "reload"
    return data


def apps_detail_delete_handler(req):
    # POST: delete the app and return to the apps list
    path, error_data = require_app_path(req, apps_detail_data)
    if error_data:
        return error_data
    ret = openrun_admin.delete_apps(path)
    if ret.error:
        data = apps_detail_data(req)
        data["FlashError"] = "Delete failed: %s" % ret.error
        return data
    # The app is gone, go back to the apps list
    return ace.response(apps_detail_data(req), block="detail_content",
                        redirect=req.AppPath + "/apps")


def apps_files_handler(req):
    # The old version-files table page is retired: redirect to the detail
    # page's Files tab, translating the env/version params. The download
    # fragment on this route stays as the zip endpoint
    path = utils.query_param(req, "path")
    version = utils.query_param(req, "version")
    env = utils.query_param(req, "env") or "prod"
    target = "%s/apps/detail?path=%s&tab=files" % (req.AppPath, path)
    if version:
        target += "&v=%s:%s" % (env, version)
    return ace.redirect(target)


def apps_files_download_handler(req):
    # GET: bundle the version's files into a zip and stream it back to the
    # client as an attachment (chunked, no disk/db staging). Errors returned
    # by get_version_zip render the detail page; later producer errors are
    # handled by the framework while writing the download
    path = utils.query_param(req, "path")
    version = utils.query_param(req, "version")
    env = utils.query_param(req, "env") or "prod"

    ret = openrun.get_version_zip(resolve_env_path(path, env), version=version)
    if ret.error:
        # apps_files_handler now returns a redirect, not template data.
        # Render the detail page so the download error remains visible.
        data = apps_detail_data(req)
        data["FlashError"] = "Download failed: %s" % ret.error
        return ace.response(data, "app_detail.go.html")
    return ace.response(ret.value["content"], download=ret.value["name"],
                        content_type="application/zip")


def apps_delete_handler(req):
    # POST: delete an app from the apps list
    path, error_data = require_app_path(req, apps_data)
    if error_data:
        return error_data
    ret = openrun_admin.delete_apps(path)
    error = ret.error
    return utils.flash_result(apps_data(req), error, "Deleted %s" % path, "Delete failed")


def apps_reload_handler(req):
    # POST: reload from the apps list with the dialog options. A promoting
    # reload stays on the list with a flash; a staging-only reload continues
    # on the detail page to review and promote
    path, error_data = require_app_path(req, apps_data)
    if error_data:
        return error_data
    ret = reload_app_options(req, path)

    if ret.error:
        data = apps_data(req)
        data["FlashError"] = "Reload failed: %s" % ret.error
        return data
    if ret.value.get("skipped_results") and not ret.value.get("reload_results"):
        data = apps_data(req)
        data["Flash"] = "%s is already up to date" % path
        return data
    if ret.value.get("promote_results"):
        data = apps_data(req)
        data["Flash"] = "Reloaded %s and promoted to prod" % path
        return data

    # Staging reloaded; continue on the detail page to review and promote
    return ace.response(apps_data(req), block="app_groups",
                        redirect="%s/apps/detail?path=%s&staged=reload" % (req.AppPath, path))


def apps_list_promote_handler(req):
    # POST: promote staging to prod from the pending-promotion tab
    path, error_data = require_app_path(req, apps_data)
    if error_data:
        return error_data
    return promote_app_result(req, apps_data, path)


def apps_list_approve_handler(req):
    # POST: approve the pending plugin permissions from the approval tab; a
    # prod app then shows under pending promotion as the next step
    path, error_data = require_app_path(req, apps_data)
    if error_data:
        return error_data
    return approve_app_result(req, apps_data, path)


def run_sync_action(req, data_fn):
    # Run a sync and show the detailed apply results on the current page
    ret = openrun_admin.run_sync(utils.query_param(req, "sync_id"))
    error = ret.error
    data = data_fn(req)
    if error:
        data["FlashError"] = "Sync failed: %s" % error
    elif ret.value.get("error"):
        data["FlashError"] = "Sync failed: %s" % ret.value["error"]
    else:
        data["SyncResult"] = utils.sync_result_summary(ret.value)
    return data


def apps_sync_handler(req):
    # POST: run a sync from the apps list
    return run_sync_action(req, apps_data)


# ---------- App create / update forms ----------


def auth_options():
    # Valid app auth types from the server: built-ins (default/system/none)
    # plus the configured oauth, saml and client cert auth entries
    ret = openrun.list_auths()
    return ret.value if not ret.error else []


def git_auth_options():
    # The git_auth entries configured on the server plus the
    # security.default_git_auth entry name; the create forms preselect the
    # default. Errors degrade to no options and no default
    ret = openrun.list_git_auths()
    if ret.error:
        return {"entries": [], "default": ""}
    return ret.value


def binding_options():
    # The choices for the app form's service bindings dropdowns: service ids
    # (binding to a service creates an auto binding) and the base/derived
    # binding paths (an app's own auto bindings are not offered). Errors
    # (e.g. no binding:read access) degrade to empty lists
    services = []
    svc_ret = openrun.list_services()
    if not svc_ret.error:
        for entry in svc_ret.value:
            services.append(entry["service_type"] + "/" + entry["name"])

    bindings = []
    list_ret = openrun.list_bindings()
    if not list_ret.error:
        for entry in list_ret.value:
            if not entry["path"].startswith("/auto/"):
                bindings.append(entry["path"])

    return {"services": sorted(services), "bindings": sorted(bindings)}


def posted_bindings(req):
    # The bindings selected on the app form, in row order. Rows left on the
    # placeholder (empty value) are skipped
    return [ref for ref in utils.query_param_list(req, "bindings") if ref]


def app_binding_refs(app):
    # The app's current bindings as form dropdown values: an auto binding
    # path is mapped back to the service source it was created from (the
    # dropdown offers services, not auto binding paths); explicit binding
    # paths stay as-is
    refs = app.get("bindings") or []
    if not refs:
        return []
    sources = {}
    list_ret = openrun.list_bindings()
    if not list_ret.error:
        for entry in list_ret.value:
            sources[entry["path"]] = entry["source"]
    mapped = []
    for ref in refs:
        if ref.startswith("/auto/") and sources.get(ref):
            mapped.append(sources[ref])
        else:
            mapped.append(ref)
    return mapped


def form_values(req):
    # The form fields for the create/update subpages
    return {
        "path": utils.query_param(req, "path"),
        "source_url": utils.query_param(req, "source_url"),
        "spec": utils.query_param(req, "spec"),
        "auth": utils.query_param(req, "auth"),
        "git_branch": utils.query_param(req, "git_branch"),
        "git_auth": utils.query_param(req, "git_auth"),
        "mcp": utils.query_param(req, "mcp"),
        "params_rows": utils.raw_kv_rows(req, "params"),
        "bindings": posted_bindings(req),
        "approve": utils.query_param(req, "approve"),
    }


# The app form's MCP setting choices: the stored mcp document in short form
# (get_app mcp_setting). "custom" (an upstream MCP server or a scoped
# document, set from the CLI) is kept as-is by the update form
MCP_OPTIONS = [
    {"value": "default", "label": "default - actions served as MCP tools at <app>/mcp"},
    {"value": "actions", "label": "actions - serve the app's actions as MCP tools (explicit)"},
    {"value": "disable", "label": "disable - no MCP endpoint, left out of every tool list"},
]


def default_app_auth(perms):
    # security.app_default_auth_type, what an app with auth default resolves
    # to; shown on the app form. Needs config:basic_read (server_info),
    # unknown otherwise
    if not (perms.get("config:basic_read") or perms.get("config:read") or perms.get("admin")):
        return ""
    ret = openrun.server_info()
    if ret.error:
        return ""
    return (ret.value.get("api") or {}).get("app_default_auth") or "none"


def create_form_data(req, values, error):
    # Page context for the app create form
    perms = utils.get_perms()
    return {
        "Title": "New app",
        "Nav": "apps",
        "Mode": "create",
        "Step": "edit",
        "Error": error,
        "Specs": openrun.list_specs().value,
        "AuthOptions": auth_options(),
        "DefaultAuth": default_app_auth(perms),
        "GitAuthOptions": git_auth_options()["entries"],
        "McpOptions": MCP_OPTIONS,
        "BindingOptions": binding_options(),
        "Values": values,
        "Perms": perms,
    }


def approve_step_data(req, values, review, error):
    # Create form context for the post-create approval step
    data = create_form_data(req, values, error)
    data["Step"] = "approve"
    data["Review"] = review
    return data


def apps_create_page_handler(req):
    # App create form page; git auth preselects the server's default entry
    # (error re-renders keep the submitted choice instead)
    values = form_values(req)
    if not values["git_auth"]:
        values["git_auth"] = git_auth_options()["default"]
    return create_form_data(req, values, "")


def apps_create_submit_handler(req):
    # POST: validate (dry run), create, or approve a new app
    values = form_values(req)
    action = utils.query_param(req, "action")

    if action == "approve":
        # The app was created, approve its pending permissions
        ret = openrun_admin.approve_apps(values["path"])
        if ret.error:
            pending = openrun_admin.approve_apps(values["path"], dry_run=True)
            review = {"loads": [], "permissions": []}
            if not pending.error:
                review = utils.review_from_dryrun({"approve_results": pending.value.get("staged_update_results")})
            return approve_step_data(req, values, review, ret.error)
        return form_redirect(req, req.AppPath + "/apps")

    if not values["path"]:
        return create_form_data(req, values, "App path is required")
    if not values["source_url"]:
        return create_form_data(req, values, "Source url is required")

    params, err = utils.parse_kv_rows(req, "params")
    if err:
        return create_form_data(req, values, err)

    auth = values["auth"] if values["auth"] != "default" else ""
    # The MCP setting: default leaves the app on the default behavior
    # (actions served at /mcp when it defines any)
    mcp = values["mcp"] if values["mcp"] not in ("", "default") else ""

    if action == "create":
        # Create the app without approval; if it requests permissions, ask
        # for the approval as the next step
        ret = openrun_admin.create_app(values["path"], values["source_url"],
                               approve=False, auth=auth,
                               spec=values["spec"], git_branch=values["git_branch"],
                               git_auth=values["git_auth"], params=params,
                               bindings=values["bindings"], mcp=mcp)
        if ret.error:
            return create_form_data(req, values, ret.error)
        if utils.needs_approval(ret.value):
            data = approve_step_data(req, values, utils.review_from_dryrun(ret.value), "")
            # The effective login of the new app: it is reachable at its
            # url as soon as it is created, "none" means by anyone
            data["CreatedAuth"] = ret.value.get("auth") or ""
            return data
        return form_redirect(req, req.AppPath + "/apps")

    # Validate: dry run to check the create and gather the requested
    # permissions, nothing is committed
    ret = openrun_admin.create_app(values["path"], values["source_url"],
                           approve=True, dry_run=True, auth=auth,
                           spec=values["spec"], git_branch=values["git_branch"],
                           git_auth=values["git_auth"], params=params,
                           bindings=values["bindings"], mcp=mcp)
    if ret.error:
        return create_form_data(req, values, ret.error)

    data = create_form_data(req, values, "")
    data["Validated"] = True
    data["Review"] = utils.review_from_dryrun(ret.value)
    return data


def update_form_data(req, app, values, error):
    # Page context for the app update form
    return {
        "Title": "Update app",
        "Nav": "apps",
        "Mode": "update",
        "Step": "edit",
        "Error": error,
        "App": app,
        "AuthOptions": auth_options(),
        "McpOptions": MCP_OPTIONS,
        "BindingOptions": binding_options(),
        "Values": values,
        "Perms": utils.get_perms(values.get("path", "")),
    }


def update_form_source(app):
    # Edits are staged, so the update form prefills from and compares
    # against the STAGING app when one exists: prod values would silently
    # drop already-staged changes on the next submit (and make a staged
    # change impossible to revert from the form). Auth stays on the main
    # app - it is a setting, not staged
    if app["is_dev"] or not app.get("stage_path"):
        return app
    # Staging apps are internal; get_app skips them without the flag
    ret = openrun.get_app(app["stage_path"], include_internal=True)
    error = ret.error
    if error:
        return app
    return ret.value


def apps_update_page_handler(req):
    # App update form page, prefilled from the staged configuration
    path = utils.query_param(req, "path")
    ret = openrun.get_app(path)
    if ret.error:
        return update_form_data(req, None, {}, ret.error)

    app = ret.value
    source = update_form_source(app)
    values = {
        "path": app["path"],
        "auth": app["auth"] or "default",
        "mcp": source.get("mcp_setting") or "default",
        "params_rows": utils.kv_rows(source["params"]),
        "bindings": app_binding_refs(source),
    }
    return update_form_data(req, app, values, "")


def apps_update_submit_handler(req):
    # POST: apply param/binding (staged) and auth (direct) changes
    path = utils.query_param(req, "path")
    values = {
        "path": path,
        "auth": utils.query_param(req, "auth"),
        "mcp": utils.query_param(req, "mcp"),
        "params_rows": utils.raw_kv_rows(req, "params"),
        "bindings": posted_bindings(req),
    }

    ret = openrun.get_app(path)
    if ret.error:
        return update_form_data(req, None, values, ret.error)
    app = ret.value
    # Changed-detection runs against the staging app, the same source the
    # form prefilled from - updates land on staging
    source = update_form_source(app)

    params, err = utils.parse_kv_rows(req, "params")
    if err:
        return update_form_data(req, app, values, err)

    params_changed = params != source["params"]
    if params_changed:
        # Params apply to staging; promotion is asked on the detail page
        result = openrun_admin.update_params(path, params, promote=False)
        if result.error:
            return update_form_data(req, app, values, result.error)

    # Compare in dropdown-value space (auto binding paths mapped back to
    # their service source), same as the form prefill
    bindings_changed = values["bindings"] != app_binding_refs(source)
    if bindings_changed:
        # Bindings apply to staging like params; a single "-" clears them all
        result = openrun_admin.update_bindings(path, values["bindings"] or ["-"],
                                               promote=False)
        if result.error:
            return update_form_data(req, app, values, result.error)

    # The MCP document is version controlled like params and bindings:
    # staged, promoted later. "custom" (a document the form does not
    # edit) and an unchanged choice leave it alone
    new_mcp = values["mcp"] or "default"
    mcp_changed = new_mcp != "custom" and new_mcp != (source.get("mcp_setting") or "default")
    if mcp_changed:
        result = openrun_admin.update_mcp(path, new_mcp, promote=False)
        if result.error:
            return update_form_data(req, app, values, result.error)

    new_auth = values["auth"] or "default"
    if new_auth != (app["auth"] or "default"):
        # Auth is an app setting, not version controlled; applies directly
        result = openrun_admin.update_auth(path, new_auth)
        if result.error:
            return update_form_data(req, app, values, result.error)

    if (params_changed or bindings_changed or mcp_changed) and not app.get("is_dev"):
        # Ask about promoting the staged change; dev apps apply directly
        # (they have no staging), so there is nothing to promote
        return form_redirect(req, "%s/apps/detail?path=%s&staged=update" % (req.AppPath, path))
    return form_redirect(req, "%s/apps/detail?path=%s" % (req.AppPath, path))


# ---------- Bindings and services ----------


def bindings_data(req):
    # Bindings page: services table plus base/derived/auto binding tables
    query = utils.query_param(req, "query").lower()

    # Map app id -> app path, to show which app an auto binding belongs to
    app_paths = {}
    for entry in openrun.list_apps(include_internal=True).value:
        app_paths[entry["id"]] = entry["path"]

    base = []
    derived = []
    auto = []
    total = 0
    list_error = ""
    list_ret = openrun.list_bindings()
    if list_ret.error:
        # No binding:read access, show the page with the error instead
        list_error = list_ret.error
    for entry in (list_ret.value if not list_error else []):
        total += 1
        path = entry["path"]
        derived_from = entry["derived_from"]

        created_by = entry.get("created_by") or ""
        # Search matches the binding path, the base binding's path and creator
        if query and query not in path.lower() and query not in derived_from.lower() and \
           query not in created_by.lower():
            continue

        metadata = entry["metadata"]
        staged = entry["staged_metadata"]

        grants = metadata["grants"] or []
        staged_grants = staged["grants"] or []
        config = metadata["config"] or {}
        staged_config = staged["config"] or {}

        binding = {
            "path": path,
            "created_by": created_by,
            "source": entry["source"],
            "service_type": entry["service_type"],
            "service_name": entry["service_name"],
            "derived_from": derived_from,
            "grants": grants,
            "staged_grants": staged_grants,
            # staging has grant/config changes which are not applied to prod yet
            "has_staged": staged_grants != grants or staged_config != config,
            "config_keys": sorted(config.keys()),
            "update_time": utils.nonzero_time(entry["update_time"]),
        }

        if path.startswith("/auto/"):
            # Auto bindings are created for app service references, path is
            # /auto/<app_id>/<service_type>
            app_id = path.split("/")[2] if len(path.split("/")) > 3 else ""
            binding["app_path"] = app_paths.get(app_id, app_id)
            auto.append(binding)
        elif derived_from:
            derived.append(binding)
        else:
            base.append(binding)

    # Service entries, shown above the binding tables. Search matches the
    # service type/name
    services = []
    services_error = ""
    svc_ret = openrun.list_services()
    if svc_ret.error:
        services_error = svc_ret.error
    for entry in (svc_ret.value if not services_error else []):
        service_id = entry["service_type"] + "/" + entry["name"]
        if query and query not in service_id.lower():
            continue
        services.append({
            "id": service_id,
            "service_type": entry["service_type"],
            "name": entry["name"],
            "is_default": entry["is_default"],
            "staging": entry["staging"],
            "config_keys": entry["config_keys"] or [],
            "update_time": utils.nonzero_time(entry["update_time"]),
        })

    return {
        "Title": "Bindings",
        "Nav": "bindings",
        "Query": query,
        "Total": total,
        "Perms": utils.get_perms(),
        "FlashError": list_error,
        "Services": sorted(services, key=lambda svc: svc["id"]),
        "ServicesError": services_error,
        # Most recently updated bindings first
        "Base": utils.sort_recent(base, "update_time", "path"),
        "Derived": utils.sort_recent(derived, "update_time", "path"),
        "Auto": utils.sort_recent(auto, "update_time", "path"),
    }


def bindings_health_data(req):
    # Card health indicator fragments for the bindings page: one aggregate
    # call checks every entry in the group (services / base / derived / auto
    # bindings). The checks dial the real backends, so the page renders a
    # skeleton and loads these async; the server caches results per target,
    # keeping the load-triggered refires on partial swaps cheap. Bindings are
    # checked on BOTH the prod and the staging account.
    group = utils.query_param(req, "group")
    if group == "services":
        ret = openrun.service_health()
    elif group == "base":
        ret = openrun.binding_health(kind="base")
    elif group == "derived":
        ret = openrun.binding_health(kind="derived")
    elif group == "auto":
        ret = openrun.binding_health(kind="auto")
    else:
        return {"Group": group, "Error": "unknown health group %s" % group}

    error = ret.error
    if error:
        return {"Group": group, "Error": error}

    value = ret.value
    total = int(value["total"])
    unhealthy = int(value["unhealthy"])

    # Auto bindings belong to an app (path is /auto/<app_id>/<service_type>);
    # resolve the owning app so failing entries can name and link it. Best
    # effort: without app read access the entries just render pathless.
    app_paths = {}
    if group == "auto" and unhealthy > 0:
        apps_ret = openrun.list_apps(include_internal=True)
        apps_error = apps_ret.error
        if not apps_error:
            for entry in apps_ret.value:
                app_paths[entry["id"]] = entry["path"]

    auto_app_path = lambda path: app_paths.get(
        path.split("/")[2] if len(path.split("/")) > 3 else "", "")

    failures = []
    for entry in value["results"]:
        if group == "services":
            if not entry["healthy"]:
                failures.append({"id": entry["id"], "env": "", "error": entry["error"], "app_path": ""})
        else:
            # A binding row can fail on either env; list each failing env
            app_path = auto_app_path(entry["path"]) if group == "auto" else ""
            if not entry["healthy"]:
                failures.append({"id": entry["path"], "env": "prod", "error": entry["error"],
                                 "app_path": app_path})
            if not entry["staging_healthy"]:
                failures.append({"id": entry["path"], "env": "staging", "error": entry["staging_error"],
                                 "app_path": app_path})

    return {
        "Group": group,
        "Error": "",
        "Empty": total == 0,
        "HasFailures": len(failures) > 0,
        "UnhealthyText": "%d unhealthy" % unhealthy,
        "HealthyTip": "%d checked, all healthy" % total,
        "Failures": failures,
    }


def binding_form_values(req):
    # The form fields for the binding create/update subpages
    return {
        "path": utils.query_param(req, "path"),
        "source": utils.query_param(req, "source"),
        "grants_text": utils.query_param(req, "grants_text"),
        "config_rows": utils.raw_kv_rows(req, "config"),
    }


def binding_form_data(req, mode, values, error):
    # Page context for the binding create/update form
    return {
        "Title": "New binding" if mode == "create" else "Update binding",
        "Nav": "bindings",
        "Mode": mode,
        "Error": error,
        "Values": values,
        "Perms": utils.get_perms(),
    }


def bindings_create_page_handler(req):
    # Binding create form page
    return binding_form_data(req, "create", binding_form_values(req), "")


def bindings_create_submit_handler(req):
    # POST: validate (dry run) or create a binding
    values = binding_form_values(req)
    action = utils.query_param(req, "action")

    if not values["path"]:
        return binding_form_data(req, "create", values, "Binding path is required")
    if not values["source"]:
        return binding_form_data(req, "create", values, "Source is required")

    config, err = utils.parse_kv_rows(req, "config")
    if err:
        return binding_form_data(req, "create", values, err)
    grants = utils.parse_lines(values["grants_text"])

    if action == "validate":
        ret = openrun_admin.create_binding(values["path"], values["source"],
                                         grants=grants, config=config, dry_run=True)
        if ret.error:
            return binding_form_data(req, "create", values, ret.error)
        data = binding_form_data(req, "create", values, "")
        data["Validated"] = True
        return data

    ret = openrun_admin.create_binding(values["path"], values["source"],
                                     grants=grants, config=config)
    if ret.error:
        return binding_form_data(req, "create", values, ret.error)
    return form_redirect(req, req.AppPath + "/bindings")


def find_binding(path):
    # Look up one binding by path from the bindings list
    ret = openrun.list_bindings()
    if ret.error:
        return None
    for entry in ret.value:
        if entry["path"] == path:
            return entry
    return None


def bindings_update_page_handler(req):
    # Binding update form page, prefilled with the staged grants
    path = utils.query_param(req, "path")
    binding = find_binding(path)
    if not binding:
        return binding_form_data(req, "update", {"path": path}, "binding %s not found" % path)

    # Updates apply to staging first, edit the staged grants
    staged_grants = binding["staged_metadata"]["grants"] or []
    values = {
        "path": path,
        "source": binding["source"],
        "grants_text": "\n".join(staged_grants),
    }
    data = binding_form_data(req, "update", values, "")
    data["Binding"] = binding
    return data


def bindings_update_submit_handler(req):
    # POST: apply the grant additions/removals from the textarea diff
    path = utils.query_param(req, "path")
    values = binding_form_values(req)
    values["path"] = path

    binding = find_binding(path)
    if not binding:
        return binding_form_data(req, "update", values, "binding %s not found" % path)

    current = binding["staged_metadata"]["grants"] or []
    wanted = utils.parse_lines(values["grants_text"])
    add_grants = [g for g in wanted if g not in current]
    delete_grants = [g for g in current if g not in wanted]

    if not add_grants and not delete_grants:
        return form_redirect(req, req.AppPath + "/bindings")

    ret = openrun_admin.update_binding(path, add_grants=add_grants,
                                     delete_grants=delete_grants, promote=True)
    if ret.error:
        data = binding_form_data(req, "update", values, ret.error)
        data["Binding"] = binding
        return data
    return form_redirect(req, req.AppPath + "/bindings")


def bindings_delete_handler(req):
    # POST: delete a binding from the bindings list
    path = utils.query_param(req, "path")
    ret = openrun_admin.delete_binding(path)
    error = ret.error
    return utils.flash_result(bindings_data(req), error, "Deleted binding %s" % path, "Delete failed")


def service_form_data(req, values, error):
    # Page context for the service create form
    return {
        "Title": "New service",
        "Nav": "bindings",
        "Error": error,
        "Values": values,
        "Validated": False,
        "Perms": utils.get_perms(),
    }


def services_create_page_handler(req):
    # Service create form page
    values = {"id": "", "config_rows": [], "is_default": False, "staging": ""}
    return service_form_data(req, values, "")


def services_create_submit_handler(req):
    # POST: validate (dry run) or create a service
    values = {
        "id": utils.query_param(req, "id").strip(),
        "config_rows": utils.raw_kv_rows(req, "config"),
        "is_default": utils.query_param(req, "is_default") == "on",
        "staging": utils.query_param(req, "staging").strip(),
    }
    action = utils.query_param(req, "action")

    config, err = utils.parse_kv_rows(req, "config")
    if err:
        return service_form_data(req, values, err)

    dry_run = action == "validate"
    ret = openrun_admin.create_service(values["id"], config=config,
                                       is_default=values["is_default"],
                                       staging=values["staging"], dry_run=dry_run)
    if ret.error:
        return service_form_data(req, values, ret.error)

    if dry_run:
        data = service_form_data(req, values, "")
        data["Validated"] = True
        return data
    return form_redirect(req, req.AppPath + "/bindings")


def services_delete_handler(req):
    # POST: delete a service from the bindings page
    id = utils.query_param(req, "id")
    ret = openrun_admin.delete_service(id)
    error = ret.error
    return utils.flash_result(bindings_data(req), error, "Deleted service %s" % id, "Service delete failed")


# ---------- Containers ----------


CONTAINER_TABS = ("apps", "sidecar", "litestream", "job", "all", "agent", "kaniko")


def container_kind(entry):
    # Classify a managed container: job run containers by their job labels
    # (a container / pod per run on both runtimes), the litestream
    # replication sidecars by their -ls name suffix (same heuristic as the
    # replication status API), app sidecars by the role label (separate
    # containers on docker/podman; on kubernetes they run inside the app
    # pod, which keeps kind app)
    if entry.get("job"):
        return "job"
    if entry["name"].endswith("-ls"):
        return "litestream"
    if entry.get("role") == "sidecar":
        return "sidecar"
    return "app"


def containers_data(req):
    # Containers page: tabs by container type - Apps (default), Sidecars,
    # Litestream, Jobs, All (everything OpenRun spawned, whatever the type)
    # plus the app builder's agent containers and (on kubernetes) the
    # kaniko image build pods as their own tabs. Every tab lists running
    # containers only unless "show stopped" is on - except Jobs, which
    # always lists the finished run containers too (a job container is
    # stopped for most of its life; it stays until the run is retired); the
    # search query and the show-stopped state carry across tab switches
    query = utils.query_param(req, "query").lower()
    filter = utils.query_param(req, "filter") or "apps"
    if filter in ("running", "exited"):
        # Old state chips / bookmarks: map onto the Apps tab
        filter = "apps"
    if filter not in CONTAINER_TABS:
        filter = "apps"
    show_stopped = utils.query_param(req, "stopped") in ("1", "true", "on")

    data = {
        "Title": "Containers",
        "Nav": "containers",
        "Query": query,
        "Filter": filter,
        "ShowStopped": show_stopped,
        "Total": 0,
        "Running": 0,
        "Runtime": "",
        "Containers": [],
        "Perms": utils.get_perms(),
    }

    # times=True: every tab is sorted by container start time (one batched
    # inspect per listing)
    ret = openrun.list_containers(times=True)
    if ret.error:
        data["FlashError"] = ret.error
        return data

    # Header counts and the runtime come from the app containers, whatever
    # the tab (the builder agents and kaniko pods are separate listings)
    managed = []
    for entry in ret.value:
        e = dict(entry.items())
        e["kind"] = container_kind(e)
        managed.append(e)
        if e["kind"] == "app":
            data["Total"] += 1
            data["Runtime"] = e["runtime"]
            if e["state"] == "running":
                data["Running"] += 1

    def matches(e):
        if not show_stopped and e["state"] != "running" and filter != "job":
            return False
        if query and query not in e["name"].lower() and query not in e["app_path"].lower() and \
           query not in e.get("image", "").lower() and query not in e["id"].lower():
            return False
        return True

    def special(ctype):
        # Best effort on the All tab (a missing builder feature or a
        # non-kubernetes runtime just has no such containers), surfaced as
        # an error on the dedicated tabs
        ret = openrun.list_containers(type=ctype, times=True)
        error = ret.error
        if error:
            return None, error
        entries = []
        for entry in ret.value:
            e = dict(entry.items())
            e["kind"] = ctype
            entries.append(e)
        return entries, ""

    containers = []
    if filter in ("agent", "kaniko"):
        entries, error = special(filter)
        if error:
            data["FlashError"] = error
            return data
        containers = [e for e in entries if matches(e)]
        data["Containers"] = sort_by_start(containers)
        return data

    for e in managed:
        if filter == "sidecar":
            # On kubernetes the tab lists the pods that have sidecars
            if e["kind"] != "sidecar" and not e.get("sidecars"):
                continue
        elif filter == "litestream":
            if e["kind"] != "litestream":
                continue
        elif filter == "job":
            if e["kind"] != "job":
                continue
        elif filter == "apps":
            if e["kind"] != "app":
                continue
        if matches(e):
            containers.append(e)
    if filter == "all":
        if data["Perms"].get("feature:builder"):
            entries, _ = special("agent")
            containers.extend([e for e in (entries or []) if matches(e)])
        if data["Runtime"] == "kubernetes":
            entries, _ = special("kaniko")
            containers.extend([e for e in (entries or []) if matches(e)])

    data["Containers"] = sort_by_start(containers)
    return data


def sort_by_start(containers):
    # Most recently started containers first (a stopped container keeps its
    # last start time; creation time is the fallback when the runtime does
    # not report a start). Stable two-pass sort: app path/name ascending
    # as the tie break
    containers = sorted(containers, key=lambda c: c["app_path"] + " " + c["name"])
    return sorted(containers, key=lambda c: c.get("started_at") or c.get("created_at") or "", reverse=True)


def container_lifecycle_action(req, data_fn):
    # Start or stop a container, re-rendering the given page
    id = utils.query_param(req, "id")
    action = utils.query_param(req, "action")
    if action == "start":
        ret = openrun_admin.start_container(id)
    else:
        ret = openrun_admin.stop_container(id)
    error = ret.error
    return utils.flash_result(data_fn(req), error, "Container %s requested" % action,
                        "Container %s failed" % action)


def containers_lifecycle_handler(req):
    # POST: container start/stop from the containers list
    return container_lifecycle_action(req, containers_data)


def containers_detail_lifecycle_handler(req):
    # POST: container start/stop from the detail page
    return container_lifecycle_action(req, containers_detail_data)


def containers_detail_data(req):
    # Fast path: basic info only. Stats, disk usage and logs are slow to
    # collect and are filled in asynchronously via the fragment routes
    id = utils.query_param(req, "id")
    data = {
        "Title": "Container detail",
        "Nav": "containers",
        "Id": id,
        "Error": "",
        "Container": None,
        "Perms": utils.get_perms(),
        "HelpUrl": utils.docs_link("/docs/container/overview/"),
    }

    ret = openrun.get_container(id, stats=False)
    if ret.error:
        data["Error"] = ret.error
        return data

    c = dict(ret.value.items())
    c["started_at"] = utils.nonzero_time(c.get("started_at"))
    # Type: app / sidecar / litestream / agent (builder sandbox); kaniko
    # build pods are not managed containers (no app id) and cannot be opened
    c["kind"] = "agent" if c.get("builder_session") else container_kind(c)
    # Sidecar links: a sidecar container links to its app container's detail
    # (docker/podman, the sidecar_of label holds the app container name,
    # which get_container resolves like an id); an app container lists its
    # sidecars. Kubernetes pods carry the sidecar names themselves (their
    # state is in the pod status fragment). A job run container links to
    # its run on the app's Jobs tab instead (template)
    if c.get("role") == "sidecar":
        c["app_container_id"] = c.get("sidecar_of") or ""
    elif c["kind"] == "app" and c.get("runtime") != "kubernetes" and c.get("app_id"):
        cont_ret = openrun.list_containers()
        cont_error = cont_ret.error
        sidecars = []
        if not cont_error:
            for entry in cont_ret.value:
                if entry.get("role") == "sidecar" and entry.get("sidecar_of") == c["name"]:
                    sidecars.append(entry)
        c["sidecar_containers"] = sorted(sidecars, key=lambda e: e["sidecar"])
    data["Container"] = c
    return data


def containers_detail_stats_handler(req):
    # Slow fragment: live resource stats and disk usage
    id = utils.query_param(req, "id")
    data = {"Id": id, "Container": None, "StatsError": "", "StatsLoaded": True}

    ret = openrun.get_container(id)
    if ret.error:
        data["StatsError"] = ret.error
        return data

    c = dict(ret.value.items())
    c["size_rw_human"] = utils.human_size(c.get("size_rw") or 0)
    c["size_root_human"] = utils.human_size(c.get("size_root_fs") or 0)
    if c.get("stats"):
        stats = dict(c["stats"].items())
        stats["cpu_num"] = utils.pct_num(stats.get("cpu_percent"))
        stats["mem_num"] = utils.pct_num(stats.get("mem_percent"))
        c["stats"] = stats
    data["Container"] = c
    return data


def containers_k8s_stats_handler(req):
    # Async fragment on the container list: pod stats of the kubernetes
    # namespaces (system and apps). Renders nothing for the other runtimes
    ret = openrun.kubernetes_stats()
    if ret.error:
        return {"K8s": None, "K8sError": ret.error}
    return {"K8s": ret.value if ret.value["enabled"] else None, "K8sError": ""}


def containers_detail_k8s_handler(req):
    # Async fragment on the container detail page: kubernetes specific pod
    # status (conditions, container states, recent events)
    id = utils.query_param(req, "id")
    data = {"Id": id, "K8s": None, "K8sError": ""}
    ret = openrun.container_kubernetes_status(id)
    if ret.error:
        data["K8sError"] = ret.error
        return data
    data["K8s"] = ret.value
    return data


def containers_logs_stream_handler(req):
    # Streaming TEXT route: the last tail lines of the container logs,
    # optionally following new output (follow=1) until the client
    # disconnects. Rendered by the <log-tail> element on the detail page
    id = utils.query_param(req, "id")
    tail = utils.query_param(req, "tail")
    tail_int = int(tail) if tail.isdigit() else 500
    if tail_int > 10000:
        tail_int = 10000
    follow = utils.query_param(req, "follow") == "1"

    ret = openrun.container_logs_stream(id, tail=tail_int, follow=follow)
    if ret.error:
        return "error: %s" % ret.error
    # Return the stream response object itself. Deferred producer errors are
    # handled by the framework: HTTP 500 before data, an aborted response
    # after data. Only errors returned by the plugin call use the text above
    return ret


# ---------- Audit logs ----------


AUDIT_FILTERS = ["app_glob", "event_type", "operation", "target", "user_id",
                 "status", "start_date", "end_date", "rid"]


def audit_data(req):
    # Audit logs page: filtered events with keyset-paged infinite scroll
    filters = {}
    for key in AUDIT_FILTERS:
        filters[key] = utils.query_param(req, key)
    before = utils.query_param(req, "before_timestamp")

    data = {
        "Title": "Audit Logs",
        "Nav": "audit",
        "Filters": filters,
        "Events": [],
        "Apps": [],
        "Operations": [],
        "NextPage": "",
        "Perms": utils.get_perms(),
    }

    ret = openrun.list_audit_events(app_glob=filters["app_glob"], user_id=filters["user_id"],
                                    event_type=filters["event_type"], operation=filters["operation"],
                                    target=filters["target"], status=filters["status"],
                                    start_date=filters["start_date"], end_date=filters["end_date"],
                                    rid=filters["rid"], before_timestamp=before)
    if ret.error:
        data["FlashError"] = ret.error
        return data

    data["IsMore"] = bool(before)

    events = []
    for entry in ret.value:
        e = dict(entry.items())
        status = e.get("status") or ""
        if status == "Success" or status.startswith("2") or status.startswith("3"):
            e["status_style"] = "ok"
        elif status:
            e["status_style"] = "error"
        else:
            e["status_style"] = ""
        events.append(e)
    data["Events"] = events

    # Keyset pagination for the infinite scroll: the next page starts before
    # the oldest event on this page
    if events:
        parts = []
        for key in AUDIT_FILTERS:
            parts.append(key + "=" + filters[key])
        parts.append("before_timestamp=" + events[-1]["create_time_epoch"])
        data["NextPage"] = req.AppPath + "/audit?" + "&".join(parts)

    # Filter dropdown contents are only needed for the full page render
    if not before:
        apps_ret = openrun.list_all_apps()
        if not apps_ret.error:
            data["Apps"] = sorted([entry["path"] for entry in apps_ret.value])
        ops_ret = openrun.list_operations()
        if not ops_ret.error:
            data["Operations"] = ops_ret.value
    return data


# ---------- Configuration ----------

# Config entry sections shown on the sub pages. Each descriptor drives one
# entry table and the generic entry form; the backend API is generic (any
# named-entry section of openrun.toml), so adding a section here requires no
# backend change. Field kinds: text (default), secret, bool, list (one value
# per line). Sections with "kv": True have free-form properties instead of
# fixed fields, edited as key/value rows. Secret fields round-trip as the
# "<redacted>" placeholder, which the backend swaps for the stored value.
# Fields with "secretable": True render as a secret-input component (the
# value can be encrypted into the secrets store with one click, using the
# section's "secret_prefix" for generated names); "file": True additionally
# offers storing a picked file's content
REDACTED_VALUE = "<redacted>"

CONFIG_SECTIONS = [
    {
        "section": "git_auth",
        "title": "Git auth",
        "desc": "SSH keys for private git repo access",
        "name_help": "the name used as git_auth in app create and sync setup",
        "secret_prefix": "gitauth",
        "fields": [
            {"name": "user_id", "label": "User id", "secretable": True, "help": "ssh user, defaults to git"},
            {"name": "private_key", "label": "Private key", "kind": "secret", "secretable": True, "file": True,
             "help": "the private key contents; pick the key file to store it encrypted in the secrets store"},
            {"name": "key_file_path", "label": "Key file path", "help": "path to the private key file on the server, when the key is not set inline"},
            {"name": "password", "label": "Key password", "kind": "secret", "secretable": True, "help": "password for the private key file, if any"},
        ],
    },
    {
        "section": "auth",
        "title": "OAuth / OIDC accounts",
        "desc": "login providers, usable as app auth",
        "name_help": "provider type, optionally with a _suffix: github, google_mycorp, oidc_okta, auth0, okta, gitlab, ...",
        "secret_prefix": "oauth",
        "fields": [
            {"name": "key", "label": "Client id", "secretable": True},
            {"name": "secret", "label": "Client secret", "kind": "secret", "secretable": True},
            {"name": "org_url", "label": "Org URL", "help": "required for okta"},
            {"name": "domain", "label": "Domain", "help": "required for auth0"},
            {"name": "discovery_url", "label": "Discovery URL", "help": "required for oidc"},
            {"name": "hosted_domain", "label": "Hosted domain", "help": "google workspace domain restriction"},
            {"name": "scopes", "label": "Scopes", "kind": "list", "help": "one oauth scope per line"},
        ],
    },
    {
        "section": "saml",
        "title": "SAML accounts",
        "desc": "SAML identity providers, used as saml_<name> app auth",
        "name_help": "used as saml_<name> in app auth settings",
        "fields": [
            {"name": "metadata_url", "label": "Metadata URL", "help": "the IdP metadata url"},
            {"name": "groups_attr", "label": "Groups attribute", "help": "SAML attribute carrying the group list"},
            {"name": "use_post", "label": "Use POST binding", "kind": "bool"},
            {"name": "force_authn", "label": "Force authn", "kind": "bool"},
            {"name": "sp_key_file", "label": "SP key file", "help": "path on the server"},
            {"name": "sp_cert_file", "label": "SP cert file", "help": "path on the server"},
        ],
    },
    {
        "section": "secret",
        "title": "Secrets managers",
        "desc": "secret providers, used by {{secret ...}} templates in config and params",
        "name_help": "provider type, optionally with a _suffix: asm, ssm, vault, env, prop, kubernetes (e.g. asm_prod)",
        "kv": True,
        "kv_label": "Properties",
        "kv_help": "provider specific properties (e.g. region for asm). " +
                   "Values are parsed as numbers/booleans when possible; use \"quotes\" to force a string",
        "secret_prefix": "secretmgr",
        "fields": [],
    },
    {
        "section": "builder_agent",
        "title": "Agent configs",
        "desc": "AI agent configurations for the app builder",
        "name_help": "agent type with optional _suffix: opencode, opencode_dev, claude, codex, pi, " +
                     "or custom_<name> (custom needs dockerfile + command). The type comes from the name",
        "fields": [
            {"name": "dockerfile", "label": "Dockerfile path",
             "help": "server path to a Dockerfile overriding the embedded one; required for custom_* agents"},
            {"name": "command", "label": "ACP command", "kind": "list",
             "help": "command speaking ACP on stdio, one argument per line; required for custom_* " +
                     "agents, overrides the type default otherwise"},
            {"name": "env", "label": "Container env", "kind": "kvtable",
             "help": "environment variables set in the agent sandbox (API keys go here; " +
                     "use the lock button to store a value as a {{secret ...}} reference)"},
            {"name": "config_files", "label": "Config file mounts", "kind": "list",
             "help": "host:container[:ro] mounts for agent config/auth files, one per line"},
            {"name": "model", "label": "Model",
             "help": "model passed to the agent at session start, in the agent's naming " +
                     "(e.g. anthropic/claude-fable-5); empty uses the agent's default"},
            {"name": "effort", "label": "Effort level",
             "help": "reasoning effort passed to the agent at session start (e.g. low, medium, high); " +
                     "empty uses the agent's default"},
        ],
    },
    {
        "section": "builder_profile",
        "title": "Builder profiles",
        "desc": "how builder apps are built and published: agent config, git target, publish " +
                "destination, default spec and prompt. With none configured, the built-in default " +
                "applies (opencode, local publish, publish anywhere); with several, the new-app " +
                "form asks which to use",
        "name_help": "short name shown in the new-app form (e.g. internal_tool, dashboard)",
        "fields": [
            {"name": "agent", "label": "Agent config", "kind": "select", "options": "builder_agent",
             "required": True,
             "help": "builder_agent entry sessions created with this profile run (required)"},
            {"name": "git_config", "label": "Git target", "kind": "select", "options": "builder_git",
             "empty_label": "none (publish locally)",
             "help": "builder_git entry apps publish to; empty publishes locally to $OPENRUN_HOME/app_src"},
            {"name": "publish_mode", "label": "Publish destination type", "kind": "select",
             "options_list": ["subdomain", "path", "glob"],
             "empty_label": "anywhere (no restriction)",
             "help": "shapes the publish dialog: subdomain - the user types a subdomain of the " +
                     "target domain; path - the user types an app name under the target path prefix; " +
                     "glob - the user types a full path matching the target glob"},
            {"name": "publish_target", "label": "Publish destination",
             "help": "subdomain: the parent domain, ending in . appends the server default_domain " +
                     "(\".\" alone = subdomain of the default domain); path: the path prefix (e.g. " +
                     "/teams); glob: the app path glob (e.g. /teams/* or example.com:/**)"},
            {"name": "spec", "label": "Default spec", "kind": "select", "options": "specs",
             "empty_label": "none (plain OpenRun app)",
             "help": "framework scaffold new sessions start from; empty builds a plain server-rendered app"},
            {"name": "services", "label": "Bindable services", "kind": "checklist",
             "options": "builder_services",
             "help": "services the new-app form offers for auto binding: pick specific services, " +
                     "\"defaults\" to offer the default service of each type, or none to offer " +
                     "no services. The app gets one auto binding per chosen service (one per type)"},
            {"name": "prompt", "label": "Prompt", "kind": "textarea",
             "help": "instructions for the agent; appended to the system prompt, or replacing it when Replace is set"},
            {"name": "replace", "label": "Replace the system prompt", "kind": "bool",
             "help": "when set this profile's prompt replaces the system prompt entirely instead of being appended"},
            {"name": "description", "label": "Description",
             "help": "shown next to the profile in the new-app form"},
        ],
    },
    {
        "section": "builder_git",
        "title": "Git targets",
        "desc": "named git repos builder apps publish to; a builder profile picks one via " +
                "Git target; no choice publishes locally",
        "name_help": "short name for this repo (e.g. tools, prod)",
        "fields": [
            {"name": "repo", "label": "Repo",
             "help": "git repo url for publish commits (e.g. github.com/org/apps)"},
            {"name": "branch", "label": "Branch",
             "help": "branch for publish commits; empty means main"},
            {"name": "auth", "label": "Git auth", "kind": "select", "options": "git_auths",
             "empty_label": "none (public repo)", "default_from": "git_auth_default",
             "help": "git_auth entry for this repo; empty for public/unauthenticated"},
            {"name": "apps_file", "label": "Apps file",
             "help": "declarative file relative to the repo root; empty means apps.star"},
            {"name": "source_dir", "label": "Source directory",
             "help": "repo directory for published app sources; empty means apps"},
        ],
    },
]


def config_section_meta(section):
    for meta in CONFIG_SECTIONS:
        if meta["section"] == section:
            return meta
    return None


# Config sub pages under /config. Each page groups entry sections (tables of
# named entries) and settings (individual fields of the struct sections,
# set through set_config_value). Setting kinds: text (default), select, bool,
# int. select options come from the named source resolved in
# config_setting_options. All changes on these pages are live immediately
CONFIG_PAGES = [
    {
        "page": "auth",
        "title": "Authentication",
        "desc": "login providers and the default app authentication",
        "entry_sections": ["auth", "saml"],
        "settings": [
            {"section": "security", "key": "app_default_auth_type",
             "label": "Default app auth", "kind": "select", "options": "auths",
             "help": "auth used for apps set to 'default': none/system, or any oauth, saml or client cert auth (cert auths are configured in openrun.toml)"},
        ],
    },
    {
        "page": "api",
        "title": "Remote Management API",
        "desc": "the management REST API over HTTPS used by the OpenRun CLI on other machines, and the " +
                "external url and token lifetimes shared by every API surface",
        "entry_sections": [],
        # One card, in the order the page shows: the external url, the REST
        # surface switch and its logins; the token lifetimes and the REST
        # operation overrides under Additional settings
        "settings": [
            {"section": "api", "key": "external_url", "label": "External URL",
             "help": "canonical https origin (https://host[:port]) for API tokens and the OAuth metadata; " +
                     "defaults to security.callback_url. Required by an enabled surface, along with an HTTPS " +
                     "listener (or security.trusted_proxies) and RBAC enforcement"},
            {"section": "api", "key": "rest.enable", "label": "Remote management (REST) enabled", "kind": "bool",
             "help": "serve the management API over HTTPS; needs an HTTPS listener (or trusted proxies), " +
                     "the external URL and RBAC enforcement"},
            {"section": "api", "key": "rest.auth", "label": "Login mechanisms", "kind": "checklist",
             "options": "api_auths",
             "help": "login mechanisms for openrun login (API keys work regardless); at least one is " +
                     "required, admin by default"},
            {"section": "api", "key": "access_token_ttl", "label": "Access token TTL", "advanced": True,
             "help": "OAuth access token lifetime (Go duration, default 1h)"},
            {"section": "api", "key": "refresh_token_ttl", "label": "Refresh token TTL", "advanced": True,
             "help": "OAuth refresh token lifetime per rotation (default 720h)"},
            {"section": "api", "key": "grant_max_ttl", "label": "Login max lifetime", "advanced": True,
             "help": "absolute OAuth grant lifetime, after which a new interactive login is required (default 2160h)"},
            {"section": "api", "key": "federated_identity_ttl", "label": "Federated identity TTL", "advanced": True,
             "help": "provider-derived group snapshot max age (default 720h)"},
            {"section": "api", "key": "pat_default_ttl", "label": "API key default expiry", "advanced": True,
             "help": "default API key lifetime (default 2160h); --expires=never is required for non-expiring keys"},
            {"section": "api", "key": "rest.enable_apis", "label": "Additional REST APIs to enable", "kind": "checklist",
             "options": "api_ops_rest_disabled", "advanced": True,
             "help": "operations disabled for the remote API by default, opted back in (none currently)"},
            {"section": "api", "key": "rest.disable_apis", "label": "REST APIs to disable", "kind": "checklist",
             "options": "api_ops", "advanced": True,
             "help": "operations turned off for remote CLI callers (the local unix socket CLI is never affected)"},
        ],
    },
    {
        "page": "mcp",
        "title": "MCP API",
        "desc": "Model Context Protocol access for AI clients: the app actions endpoint, the management endpoint and how to connect Claude Code, Codex or any MCP client",
        "entry_sections": [],
        "settings": [],
        # Per-endpoint settings cards, each rendered inside its endpoint's
        # section of the connect guide (config_mcp.go.html, by key): the
        # enable switch and the login mechanisms up front, the operation
        # overrides under Additional settings
        "groups": [
            {
                "key": "mcp",
                "title": "Management MCP",
                "desc": "the Model Context Protocol endpoint at <external url>/_openrun/mcp: apps, versions, " +
                        "services, secrets and config for AI clients (off by default)",
                "settings": [
                    {"section": "api", "key": "mcp.enable", "label": "Enabled", "kind": "bool",
                     "help": "serve the MCP endpoint; needs an HTTPS listener (or trusted proxies), the external " +
                             "URL and RBAC enforcement"},
                    {"section": "api", "key": "mcp.auth", "label": "Login mechanisms", "kind": "checklist",
                     "options": "api_auths",
                     "help": "browser login mechanisms for MCP OAuth clients (API keys minted with --resource mcp " +
                             "work regardless); at least one is required, admin by default"},
                    {"section": "api", "key": "mcp.enable_apis", "label": "Additional APIs to enable", "kind": "checklist",
                     "options": "api_ops_mcp_disabled", "advanced": True,
                     "help": "dangerous operations disabled for MCP by default, opted back in (logged at startup)"},
                    {"section": "api", "key": "mcp.disable_apis", "label": "APIs to disable", "kind": "checklist",
                     "options": "api_ops_mcp", "advanced": True,
                     "help": "operations turned off for MCP on top of the defaults; secret_create is on by default, " +
                             "disable it if secret values must never pass through an AI client"},
                    {"section": "api", "key": "mcp.skip_destructive_confirm", "label": "Skip destructive confirmation",
                     "kind": "bool", "advanced": True,
                     "help": "disable the dry-run preview + confirmation prompt on destructive MCP tools (headless automation)"},
                ],
            },
            {
                "key": "app_mcp",
                "title": "App actions MCP",
                "desc": "the actions of every app the user can run as MCP tools, at <external url>/_openrun/app_mcp " +
                        "(on by default; the login comes from the url's ?auth= param)",
                "settings": [
                    {"section": "api", "key": "app_mcp.enable", "label": "Enabled", "kind": "bool",
                     "help": "serve the app actions endpoint; it is never served while security.unsafe_disable_rbac is set"},
                    {"section": "api", "key": "app_mcp.allowed_auth", "label": "Allowed logins", "kind": "checklist",
                     "options": "app_mcp_auths", "advanced": True,
                     "help": "logins the url's ?auth= param may name; none selected allows every configured login"},
                    {"section": "api", "key": "app_mcp.max_tools", "label": "Max tools", "kind": "int", "advanced": True,
                     "help": "the tool list fails above this many tools, asking for a narrower ?apps= glob (default 300)"},
                    {"section": "api", "key": "app_mcp.list_ttl", "label": "Tool list TTL", "advanced": True,
                     "help": "freshness hint of the tool list for clients (Go duration, default 3m)"},
                ],
            },
        ],
    },
    {
        "page": "git",
        "title": "Git auth",
        "desc": "git credentials for private repos and the default entry",
        "entry_sections": ["git_auth"],
        "settings": [
            {"section": "security", "key": "default_git_auth",
             "label": "Default git auth", "kind": "select", "options": "git_auths",
             "help": "git auth entry used when an app or sync does not name one"},
        ],
    },
    {
        "page": "secrets",
        "title": "Secrets",
        "desc": "secret manager providers",
        "entry_sections": ["secret"],
        "settings": [],
    },
    {
        "page": "system",
        "title": "System Settings",
        "desc": "server level defaults, app config and node config overrides",
        "entry_sections": [],
        "settings": [
            {"section": "system", "key": "default_domain", "label": "Default domain",
             "help": "domain used for apps created without a domain"},
            {"section": "system", "key": "stage_at", "label": "Stage at",
             "help": "staging mode for new prod apps: domain, path, or a staging domain name"},
            {"section": "system", "key": "list_apps_title", "label": "List apps title",
             "help": "title of the app listing page"},
            {"section": "system", "key": "show_hosted_with", "label": "Show \"Hosted with OpenRun\"",
             "kind": "bool", "help": "footer on the app listing page"},
        ],
        "kv_sections": [
            {"section": "app_config",
             "title": "App config defaults",
             "help": "defaults applied to all apps on their next reload - dotted keys like " +
                     "cors.allow_origin or container.health_timeout_secs. Values are parsed as " +
                     "numbers/booleans when possible; use \"quotes\" to force a string",
             "placeholder": "cors.allow_origin"},
            {"section": "node_config",
             "title": "Node config",
             "help": "values apps read with the config() builtin, applied on their next " +
                     "reload - free form keys. Values are parsed as numbers/booleans when " +
                     "possible; use \"quotes\" to force a string",
             "placeholder": "key_name"},
        ],
    },
    {
        "page": "builder",
        "title": "App builder",
        "desc": "agent configs, git targets, builder profiles and builder settings",
        "entry_sections": ["builder_agent", "builder_git", "builder_profile"],
        "settings": [
            {"section": "app_builder", "key": "enabled", "label": "Enabled", "kind": "bool",
             "help": "the AI app builder (Builder tab); needs a docker/podman runtime, not supported on Kubernetes"},
            {"section": "app_builder", "key": "default_builder_profile", "label": "Default builder profile",
             "kind": "select", "options": "builder_profile",
             "help": "builder_profile entry used when the user does not pick one; empty auto-uses a " +
                     "single profile, or the built-in opencode default when none are configured"},
            {"section": "app_builder", "key": "preview_path", "label": "Preview path prefix", "advanced": True,
             "help": "where draft preview apps are mounted"},
            {"section": "app_builder", "key": "max_sessions", "label": "Max live sessions", "kind": "int",
             "advanced": True,
             "help": "concurrent agent sandboxes; further creates ask to stop an idle session"},
            {"section": "app_builder", "key": "session_idle_mins", "label": "Session idle minutes", "kind": "int",
             "advanced": True,
             "help": "stop the agent sandbox after this idle time (the draft and transcript remain)"},
            {"section": "app_builder", "key": "system_prompt", "label": "System prompt", "kind": "textarea",
             "advanced": True,
             "help": "replaces the embedded base prompt sent to the agent; leave empty for the built-in default"},
        ],
    },
]


def page_settings(meta):
    # Every settings row of a config page: the flat list plus the rows of
    # each per-surface/group card
    rows = list(meta["settings"])
    for group in meta.get("groups") or []:
        rows.extend(group["settings"])
    return rows


def config_page_meta(page):
    for meta in CONFIG_PAGES:
        if meta["page"] == page:
            return meta
    return None


def config_page_for_section(section):
    # The sub page owning an entry section, for redirects after entry edits
    for meta in CONFIG_PAGES:
        if section in meta["entry_sections"]:
            return meta["page"]
    return ""


def config_setting_options(source):
    # Resolve a select setting's option list by source name (shared by the
    # settings rows and the entry-form select fields)
    if source == "auths":
        ret = openrun.list_auths()
        # "default" is what the setting resolves, exclude it from the choices
        return [a for a in (ret.value if not ret.error else []) if a != "default"]
    if source == "git_auths":
        ret = openrun.list_git_auths()
        return list(ret.value["entries"]) if not ret.error else []
    if source == "specs":
        ret = openrun.list_specs()
        error = ret.error
        return sorted([s for s in (ret.value or []) if s != "dummy"]) if not error else []
    if source == "builder_services":
        # "defaults" plus the live service ids, for the profile services
        # checklist
        ret = openrun.list_services()
        error = ret.error
        options = ["defaults"]
        if not error:
            for entry in ret.value:
                options.append(entry["service_type"] + "/" + entry["name"])
        return options
    if source == "api_auths":
        # Login mechanisms for the remote API surfaces: the fixed builtin and
        # admin accounts plus every oauth/saml entry name (static + dynamic)
        options = ["admin", "builtin"]
        ret = openrun.get_config_entries(["auth", "saml"])
        error = ret.error
        if not error:
            for section in ("auth", "saml"):
                for entry in ret.value["sections"].get(section) or []:
                    if entry["name"] not in options:
                        options.append(entry["name"])
        return options
    if source == "app_mcp_auths":
        # Logins the app actions endpoint's ?auth= param can name: the app
        # auth types without default and the client cert entries (no browser
        # login produces those identities)
        ret = openrun.list_auths()
        return [a for a in (ret.value if not ret.error else []) if a != "default" and a != "cert" and not a.startswith("cert_")]
    if source in ("api_ops", "api_ops_mcp", "api_ops_mcp_disabled", "api_ops_rest_disabled"):
        # Registry operation names for the enable/disable checklists:
        # api_ops = every operation (rest disable), api_ops_mcp = the ones
        # with an MCP tool (mcp disable), api_ops_mcp_disabled = the ones
        # off for MCP by default (mcp enable_apis); nothing is off by
        # default for rest
        if source == "api_ops_rest_disabled":
            return []
        ret = openrun.list_api_operations()
        error = ret.error
        if error:
            return []
        options = []
        for op in ret.value:
            if source == "api_ops_mcp" and op["mcp_excluded"]:
                continue
            if source == "api_ops_mcp_disabled" and not op["mcp_disabled"]:
                continue
            options.append(op["name"])
        return options
    if source in ("builder_agent", "builder_git", "builder_profile"):
        # Config entry names of the section, dynamic and static merged
        ret = openrun.get_config_entries([source])
        error = ret.error
        if error:
            return []
        names = {}
        for entry in ret.value["sections"].get(source) or []:
            names[entry["name"]] = True
        return sorted(names.keys())
    return []


def parse_config_value(raw):
    # Free-form config values: booleans and numbers are typed, "quotes" force
    # a string, everything else stays a string
    raw = raw.strip()
    if len(raw) >= 2 and raw.startswith('"') and raw.endswith('"'):
        return raw[1:-1]
    if raw == "true":
        return True
    if raw == "false":
        return False
    if raw.lstrip("-").isdigit() and raw.lstrip("-"):
        return int(raw)
    intpart = raw.lstrip("-")
    if intpart.count(".") == 1:
        left, right = intpart.split(".")
        if left.isdigit() and right.isdigit():
            return float(raw)
    return raw


def rbac_permission_groups():
    # The canonical permission list, grouped by resource type, from the
    # server (types.RBACPermissionGroups). Ordered list of
    # {resource, permissions} entries
    ret = openrun.list_rbac_permissions()
    return ret.value if not ret.error else []


def rbac_section(rbac):
    # Massage one RBAC config (live or draft) for display
    groups = []
    for name in sorted((rbac.get("groups") or {}).keys()):
        groups.append({"name": name, "users": rbac["groups"][name] or []})
    roles = []
    for name in sorted((rbac.get("roles") or {}).keys()):
        roles.append({"name": name, "permissions": rbac["roles"][name] or []})
    grants = []
    for i, grant in enumerate(rbac.get("grants") or []):
        grants.append({
            "index": i,
            "description": grant.get("description") or "",
            "users": grant.get("users") or [],
            "roles": grant.get("roles") or [],
            "targets": grant.get("targets") or [],
        })
    return {
        "groups": groups,
        "roles": roles,
        "grants": grants,
    }


def rbac_diff(live, draft):
    # Names of entities that differ between live and draft, per category
    def dict_diff(a, b):
        changed = []
        for name in sorted(dict(a.items() + b.items()).keys()):
            if a.get(name) != b.get(name):
                changed.append(name)
        return changed

    diff = {
        "groups": dict_diff(live.get("groups") or {}, draft.get("groups") or {}),
        "roles": dict_diff(live.get("roles") or {}, draft.get("roles") or {}),
        "grants": len(draft.get("grants") or []) != len(live.get("grants") or []) or
                  (live.get("grants") or []) != (draft.get("grants") or []),
    }
    diff["any"] = bool(diff["groups"] or diff["roles"] or diff["grants"])
    return diff


def config_data(req):
    # Top-level configuration page: one card per config area (RBAC and the
    # sub pages) and the config history
    data = {
        "Title": "Configuration",
        "Nav": "config",
        "Error": "",
        "Perms": utils.get_perms(),
        "History": [],
        "Pages": [],
    }

    ret = openrun.get_rbac_config()
    if ret.error:
        data["Error"] = ret.error
        return data

    cfg = ret.value
    live = rbac_section(cfg["rbac"])
    data["VersionId"] = cfg["version_id"]
    data["RBAC"] = {
        "groups": len(live["groups"]),
        "roles": len(live["roles"]),
        "grants": len(live["grants"]),
        "has_staged": cfg["has_staged"],
        "staged_by": cfg["draft"]["updated_by"] if cfg["has_staged"] else "",
    }

    all_sections = []
    for meta in CONFIG_PAGES:
        all_sections.extend(meta["entry_sections"])
    entries = openrun.get_config_entries(all_sections)
    if entries.error:
        data["Error"] = entries.error
        return data
    values = openrun.get_config_values()
    if values.error:
        data["Error"] = values.error
        return data

    for meta in CONFIG_PAGES:
        entry_count = 0
        for section in meta["entry_sections"]:
            for entry in entries.value["sections"].get(section) or []:
                if not (entry["source"] == "static" and entry["overridden"]):
                    entry_count += 1
        # One count per managed setting key that has a dynamic override, plus
        # all keys of the page's free-form kv sections
        dynamic_count = 0
        for setting in page_settings(meta):
            section_values = values.value["sections"].get(setting["section"]) or {}
            if setting["key"] in (section_values.get("dynamic") or {}):
                dynamic_count += 1
        for kv in meta.get("kv_sections") or []:
            kv_values = values.value["sections"].get(kv["section"]) or {}
            dynamic_count += len(kv_values.get("dynamic") or {})
        data["Pages"].append({
            "page": meta["page"],
            "title": meta["title"],
            "desc": meta["desc"],
            "entry_count": entry_count,
            "has_entries": len(meta["entry_sections"]) > 0,
            "dynamic_count": dynamic_count,
        })

    hist = openrun.list_config_history()
    if not hist.error:
        data["History"] = hist.value
    return data


def _has_static(entries, name):
    for entry in entries:
        if entry["name"] == name and entry["source"] == "static":
            return True
    return False


def entry_summary(meta, values):
    # Compact "field: value" display line for an entry card row, secrets and
    # empty fields skipped. Free-form kv entries list all their properties
    # (values arrive redacted from the server)
    parts = []
    if meta.get("kv"):
        for key in sorted(values.keys()):
            parts.append("%s: %s" % (key, values[key]))
        return "  ·  ".join(parts)
    for field in meta["fields"]:
        kind = field.get("kind") or "text"
        value = values.get(field["name"])
        if kind == "secret" or value == None or value == "" or value == False or value == []:
            continue
        if kind == "list":
            value = ", ".join([str(v) for v in value])
        parts.append("%s: %s" % (field["name"], value))
    return "  ·  ".join(parts)


def config_action_handler(req):
    # Top-level page action: history restore, live immediately
    action = utils.query_param(req, "action")
    force = utils.query_param(req, "force") == "true"

    if action == "restore":
        ret = openrun_admin.restore_config(utils.query_param(req, "restore_version"), force=force)
        ok = "Configuration restored"
    else:
        data = config_data(req)
        data["FlashError"] = "unknown action %s" % action
        return data

    error = ret.error
    return utils.flash_result(config_data(req), error, ok)


def config_page_data(req, page):
    # A config sub page: the page's settings (with effective values and
    # dynamic badges), entry section tables and the app_config key/value table
    meta = config_page_meta(page)
    data = {
        "Title": meta["title"] + " configuration",
        "Nav": "config",
        "Error": "",
        "Perms": utils.get_perms(),
        "Page": meta["page"],
        "PageTitle": meta["title"],
        "PageDesc": meta["desc"],
        "Settings": [],
        "AdvancedSettings": [],
        "SettingGroups": [],
        "Sections": [],
        "KVs": [],
    }

    ret = openrun.get_rbac_config()
    if ret.error:
        data["Error"] = ret.error
        return data
    data["VersionId"] = ret.value["version_id"]

    sections = [s["section"] for s in page_settings(meta)]
    for kv in meta.get("kv_sections") or []:
        sections.append(kv["section"])
    values = {"sections": {}}
    if sections:
        ret = openrun.get_config_values(sections)
        if ret.error:
            data["Error"] = ret.error
            return data
        values = ret.value

    def setting_row(setting):
        section_values = values["sections"].get(setting["section"]) or {}
        dynamic = section_values.get("dynamic") or {}
        static = section_values.get("static") or {}
        is_dynamic = setting["key"] in dynamic
        value = dynamic[setting["key"]] if is_dynamic else static.get(setting["key"])
        row = dict(setting)
        row["kind"] = setting.get("kind") or "text"
        if row["kind"] == "checklist":
            row["value"] = list(value) if value else []
        else:
            row["value"] = value if value != None else ""
        row["is_dynamic"] = is_dynamic
        row["static_value"] = static.get(setting["key"])
        if row["kind"] == "checklist" and row["static_value"]:
            row["static_value"] = ", ".join([str(v) for v in row["static_value"]])
        if row["kind"] in ("select", "checklist"):
            row["option_list"] = config_setting_options(setting.get("options") or "")
        return row

    for setting in meta["settings"]:
        row = setting_row(setting)
        # advanced settings render in a collapsed section, hidden by default
        if setting.get("advanced"):
            data["AdvancedSettings"].append(row)
        else:
            data["Settings"].append(row)

    for group in meta.get("groups") or []:
        rows = []
        advanced = []
        for setting in group["settings"]:
            row = setting_row(setting)
            if setting.get("advanced"):
                advanced.append(row)
            else:
                rows.append(row)
        data["SettingGroups"].append({
            "key": group.get("key") or "",
            "title": group["title"],
            "desc": group.get("desc") or "",
            "settings": rows,
            "advanced": advanced,
        })

    for kv in meta.get("kv_sections") or []:
        dynamic = (values["sections"].get(kv["section"]) or {}).get("dynamic") or {}
        rows = []
        for key in sorted(dynamic.keys()):
            rows.append({"key": key, "value": dynamic[key]})
        data["KVs"].append({
            "section": kv["section"],
            "title": kv.get("title") or kv["section"],
            "help": kv.get("help") or "",
            "placeholder": kv.get("placeholder") or "",
            "rows": rows,
        })

    if meta["entry_sections"]:
        entries = openrun.get_config_entries(meta["entry_sections"])
        if entries.error:
            data["Error"] = entries.error
            return data
        for section in meta["entry_sections"]:
            section_meta = config_section_meta(section)
            section_entries = []
            for entry in entries.value["sections"].get(section) or []:
                # A static entry shadowed by a dynamic one is not listed; the
                # dynamic entry shows the "overrides static" badge
                if entry["source"] == "static" and entry["overridden"]:
                    continue
                overrides = entry["source"] == "dynamic" and _has_static(entries.value["sections"][section], entry["name"])
                section_entries.append({
                    "name": entry["name"],
                    "source": entry["source"],
                    "overrides": overrides,
                    "summary": entry_summary(section_meta, entry["values"]),
                })
            data["Sections"].append({
                "section": section,
                "title": section_meta["title"],
                "desc": section_meta["desc"],
                "entries": section_entries,
            })
    return data


def _page_kv_section(meta, section):
    # The kv section from the form must be one this page manages
    for kv in meta.get("kv_sections") or []:
        if kv["section"] == section:
            return section
    return ""


def config_page_action_handler(req, page, page_data=None):
    # Sub page actions: set/reset a settings field, delete a dynamic entry,
    # set/delete an app_config key. All take effect immediately. page_data
    # rebuilds the page context after the action (default the generic sub
    # page; the MCP page adds its connect guide)
    meta = config_page_meta(page)
    if not page_data:
        page_data = lambda: config_page_data(req, page)
    action = utils.query_param(req, "action")
    version_id = utils.query_param(req, "version_id")
    section = utils.query_param(req, "section")
    key = utils.query_param(req, "key")

    if action == "set_value":
        kind = utils.query_param(req, "kind")
        raw = utils.query_param(req, "value")
        if kind == "bool":
            ret = openrun_admin.set_config_value(section, key, raw == "on", version_id)
            ok = "Set %s %s - change is live" % (section, key)
        elif kind == "checklist":
            # The checked set is the value, an empty selection included
            # (the server rejects an empty auth list); the reset button
            # removes the dynamic value
            ret = openrun_admin.set_config_value(section, key, utils.query_param_list(req, "value"), version_id)
            ok = "Set %s %s - change is live" % (section, key)
        elif raw.strip() == "":
            # Clearing the field resets to the static config value
            if utils.query_param(req, "is_dynamic") == "true":
                ret = openrun_admin.delete_config_value(section, key, version_id)
                ok = "Reset %s %s to the static config value" % (section, key)
            else:
                data = page_data()
                data["FlashError"] = "no value provided for %s %s" % (section, key)
                return data
        elif kind == "int":
            if not raw.strip().lstrip("-").isdigit():
                data = page_data()
                data["FlashError"] = "%s %s must be a number" % (section, key)
                return data
            ret = openrun_admin.set_config_value(section, key, int(raw.strip()), version_id)
            ok = "Set %s %s - change is live" % (section, key)
        else:
            ret = openrun_admin.set_config_value(section, key, raw.strip(), version_id)
            ok = "Set %s %s - change is live" % (section, key)
    elif action == "delete_value":
        ret = openrun_admin.delete_config_value(section, key, version_id)
        ok = "Reset %s %s to the static config value" % (section, key)
    elif action == "delete_entry":
        name = utils.query_param(req, "name")
        ret = openrun_admin.delete_config_entry(section, name, version_id)
        ok = "Deleted %s entry %s - change is live" % (section, name)
    elif action == "kv_set":
        kv_section = _page_kv_section(meta, utils.query_param(req, "kv_section"))
        kv_key = utils.query_param(req, "key").strip()
        if not kv_section or not kv_key:
            data = page_data()
            data["FlashError"] = "key cannot be empty" if kv_section else "unknown kv section"
            return data
        value = parse_config_value(utils.query_param(req, "value"))
        ret = openrun_admin.set_config_value(kv_section, kv_key, value, version_id)
        ok = "Set %s %s - applies on the next app reload" % (kv_section, kv_key)
    elif action == "kv_delete":
        kv_section = _page_kv_section(meta, utils.query_param(req, "kv_section"))
        if not kv_section:
            data = page_data()
            data["FlashError"] = "unknown kv section"
            return data
        ret = openrun_admin.delete_config_value(kv_section, utils.query_param(req, "key"), version_id)
        ok = "Removed %s %s" % (kv_section, utils.query_param(req, "key"))
    else:
        data = page_data()
        data["FlashError"] = "unknown action %s" % action
        return data

    error = ret.error
    return utils.flash_result(page_data(), error, ok)


def config_auth_data(req):
    return config_page_data(req, "auth")


def config_auth_action_handler(req):
    return config_page_action_handler(req, "auth")


def config_git_data(req):
    return config_page_data(req, "git")


def config_git_action_handler(req):
    return config_page_action_handler(req, "git")


def config_secrets_data(req):
    return config_page_data(req, "secrets")


def config_secrets_action_handler(req):
    return config_page_action_handler(req, "secrets")


def config_system_data(req):
    return config_page_data(req, "system")


def config_system_action_handler(req):
    return config_page_action_handler(req, "system")


def config_api_data(req):
    return config_page_data(req, "api")


def config_api_action_handler(req):
    return config_page_action_handler(req, "api")


def config_mcp_data(req):
    # The MCP sub page: the connect guide (endpoint state, commands, the
    # management enable) above the MCP settings cards
    data = config_page_data(req, "mcp")
    if not data["Error"]:
        mcp_connect_data(req, data)
        # Each endpoint is ONE card: its settings rows render inside it,
        # without the enable row - the card header's Enable/Disable button
        # (POST enable) is that switch
        data["GroupByKey"] = {}
        for group in data["SettingGroups"]:
            data["GroupByKey"][group["key"]] = {
                "settings": [r for r in group["settings"] if r["key"] != group["key"] + ".enable"],
                "advanced": group["advanced"],
            }
    return data


def config_mcp_action_handler(req):
    return config_page_action_handler(req, "mcp", lambda: config_mcp_data(req))


def config_builder_data(req):
    return config_page_data(req, "builder")


def config_builder_action_handler(req):
    return config_page_action_handler(req, "builder")


def config_rbac_data(req):
    # RBAC sub page: live/staged groups/roles/grants tables with the staged
    # draft publish workflow
    data = {
        "Title": "RBAC configuration",
        "Nav": "config",
        "Error": "",
        "Perms": utils.get_perms(),
    }

    ret = openrun.get_rbac_config()
    if ret.error:
        data["Error"] = ret.error
        return data

    cfg = ret.value
    data["VersionId"] = cfg["version_id"]
    data["HasStaged"] = cfg["has_staged"]
    data["DraftVersion"] = ""
    data["Live"] = rbac_section(cfg["rbac"])
    # The tables show the draft when one exists; enforcement uses live.
    # ?view=live switches to a read-only view of the live config while a
    # draft is pending (no-op without a draft, the tables show live anyway)
    data["ViewLive"] = cfg["has_staged"] and utils.query_param(req, "view") == "live"
    if cfg["has_staged"] and not data["ViewLive"]:
        data["View"] = rbac_section(cfg["staged"])
    else:
        data["View"] = data["Live"]
    if cfg["has_staged"]:
        data["Diff"] = rbac_diff(cfg["rbac"], cfg["staged"])
        data["Draft"] = cfg["draft"]
        data["DraftVersion"] = cfg["draft"]["draft_version"]
    return data


def config_rbac_action_handler(req):
    # Publish / discard / delete actions on the RBAC page. All of these edit
    # the staged draft except publish/discard. RBAC enforcement is always on
    # (no dynamic enable flag; security.unsafe_disable_rbac is static only)
    action = utils.query_param(req, "action")
    draft_version = utils.query_param(req, "draft_version")
    force = utils.query_param(req, "force") == "true"

    if action == "publish":
        ret = openrun_admin.publish_rbac_config(draft_version, force=force)
        ok = "Published RBAC configuration"
    elif action == "discard":
        ret = openrun_admin.discard_rbac_draft(draft_version)
        ok = "Discarded staged changes"
    elif action == "delete_group":
        ret = openrun_admin.delete_rbac_group(utils.query_param(req, "name"), draft_version)
        ok = "Deleted group %s from the staged config" % utils.query_param(req, "name")
    elif action == "delete_role":
        ret = openrun_admin.delete_rbac_role(utils.query_param(req, "name"), draft_version)
        ok = "Deleted role %s from the staged config" % utils.query_param(req, "name")
    elif action == "delete_grant":
        ret = openrun_admin.delete_rbac_grant(int(utils.query_param(req, "index")), draft_version)
        ok = "Deleted grant from the staged config"
    else:
        data = config_rbac_data(req)
        data["FlashError"] = "unknown action %s" % action
        return data

    error = ret.error
    return utils.flash_result(config_rbac_data(req), error, ok)


# Documentation page per config section, for the entry form help link;
# sections without an entry link the configuration overview
CONFIG_SECTION_DOCS = {
    "auth": "/docs/configuration/authentication/",
    "saml": "/docs/configuration/authentication/",
    "git_auth": "/docs/configuration/security/",
    "secret": "/docs/configuration/secrets/",
}


def config_entry_form_data(req, meta, name, values, is_edit, source, error):
    # Page context for the generic config entry form
    ret = openrun.get_rbac_config()
    version_id = ret.value["version_id"] if not ret.error else ""
    field_options = {}
    for field in meta["fields"]:
        if field.get("kind") in ("select", "checklist"):
            # options_list is a fixed choice list; options names a dynamic
            # source resolved by config_setting_options
            if field.get("options_list"):
                field_options[field["name"]] = field["options_list"]
            else:
                field_options[field["name"]] = config_setting_options(field.get("options") or "")
    return {
        "Title": "Configuration",
        "Nav": "config",
        "Meta": meta,
        "Name": name,
        "Values": values,
        "IsEdit": is_edit,
        "Source": source,
        "Error": error,
        "FieldOptions": field_options,
        "VersionId": version_id,
        "Perms": utils.get_perms(),
        "ReturnPath": "/config/" + config_page_for_section(meta["section"]),
        "HelpUrl": utils.docs_link(CONFIG_SECTION_DOCS.get(meta["section"], "/docs/configuration/overview/")),
    }


def config_entry_page_handler(req):
    # Generic entry form page (any CONFIG_SECTIONS section). With a name, the
    # form edits the dynamic entry, or overrides the static entry of that name
    section = utils.query_param(req, "section")
    name = utils.query_param(req, "name")
    meta = config_section_meta(section)
    if not meta:
        return ace.redirect(req.AppPath + "/config")

    values = {}
    source = ""
    if name:
        ret = openrun.get_config_entries([section])
        if ret.error:
            return config_entry_form_data(req, meta, name, values, False, "", ret.error)
        # Prefer the dynamic entry; fall back to the static one so an
        # override form starts prefilled with the static values
        for entry in ret.value["sections"].get(section) or []:
            if entry["name"] == name and (entry["source"] == "dynamic" or not source):
                values = entry["values"]
                source = entry["source"]
    if meta.get("kv"):
        # Free-form entries edit as key/value rows; secret-ish values arrive
        # redacted and round-trip through the placeholder
        values = {"properties_rows": utils.kv_rows(values)}
    for field in meta["fields"]:
        # kvtable fields edit their dict value as key/value rows
        if field.get("kind") == "kvtable":
            values[field["name"] + "_rows"] = utils.kv_rows(values.get(field["name"]) or {})
        # New-entry forms preselect declared field defaults (the builder_git
        # auth select starts on the server's default git auth)
        if not name and field.get("default_from") == "git_auth_default":
            values[field["name"]] = git_auth_options()["default"]
    return config_entry_form_data(req, meta, name, values, source == "dynamic", source, "")


def config_entry_submit_handler(req):
    # POST: save one dynamic config entry. The change is validated and takes
    # effect immediately (config entries are not staged, unlike RBAC)
    section = utils.query_param(req, "section")
    meta = config_section_meta(section)
    if not meta:
        return form_redirect(req, req.AppPath + "/config")

    name = utils.query_param(req, "name").strip()
    is_edit = utils.query_param(req, "is_edit") == "true"
    values = {}
    if meta.get("kv"):
        rows = utils.raw_kv_rows(req, "properties")
        parsed, error = utils.parse_kv_rows(req, "properties")
        if error:
            return config_entry_form_data(req, meta, name, {"properties_rows": rows},
                                          is_edit, utils.query_param(req, "source"), error)
        for key in parsed:
            values[key] = parse_config_value(parsed[key])
        ret = openrun_admin.set_config_entry(section, name, values, utils.query_param(req, "version_id"))
        if ret.error:
            return config_entry_form_data(req, meta, name, {"properties_rows": rows},
                                          is_edit, utils.query_param(req, "source"), ret.error)
        return form_redirect(req, req.AppPath + "/config/" + config_page_for_section(section))
    for field in meta["fields"]:
        kind = field.get("kind") or "text"
        raw = utils.query_param(req, field["name"])
        if kind == "bool":
            if raw == "on":
                values[field["name"]] = True
        elif kind == "list":
            lines = utils.parse_lines(raw)
            if lines:
                values[field["name"]] = lines
        elif kind == "checklist":
            picked = [v.strip() for v in utils.query_param_list(req, field["name"]) if v.strip()]
            if picked:
                values[field["name"]] = picked
        elif kind == "textarea":
            # multi-line text, newlines preserved
            if raw.strip():
                values[field["name"]] = raw
        elif kind == "kvtable":
            # key/value rows (kv_table template); the _rows key is for form
            # re-render only and is stripped before the entry is saved
            values[field["name"] + "_rows"] = utils.raw_kv_rows(req, field["name"])
            parsed, error = utils.parse_kv_rows(req, field["name"])
            if error:
                return config_entry_form_data(req, meta, name, values, is_edit,
                                              utils.query_param(req, "source"),
                                              "%s: %s" % (field["label"], error))
            if parsed:
                values[field["name"]] = parsed
        elif kind == "secret":
            if raw:
                values[field["name"]] = raw
            elif is_edit and utils.query_param(req, field["name"] + "__keep") == "true":
                # Blank on edit keeps the stored secret via the placeholder
                values[field["name"]] = REDACTED_VALUE
        elif raw.strip():
            values[field["name"]] = raw.strip()

    submit_values = {k: values[k] for k in values.keys() if not k.endswith("_rows")}
    ret = openrun_admin.set_config_entry(section, name, submit_values, utils.query_param(req, "version_id"))
    if ret.error:
        return config_entry_form_data(req, meta, name, values, is_edit, utils.query_param(req, "source"), ret.error)
    return form_redirect(req, req.AppPath + "/config/" + config_page_for_section(section))


def load_rbac_config():
    # The form pages edit the staged config when a draft exists, else live
    ret = openrun.get_rbac_config()
    if ret.error:
        return {"error": ret.error, "rbac": {}, "draft_version": ""}
    cfg = ret.value
    return {
        "error": "",
        "rbac": cfg["staged"] if cfg["has_staged"] else cfg["rbac"],
        "draft_version": cfg["draft"]["draft_version"] if cfg["has_staged"] else "",
        "builtin_roles": cfg.get("builtin_roles") or [],
    }


def config_form_data(req, kind, values, error, cfg=None):
    # Page context shared by the group/role/grant forms
    cfg = cfg or load_rbac_config()
    rbac = cfg["rbac"]
    return {
        "Title": "Configuration",
        "Nav": "config",
        "Kind": kind,
        "Error": error,
        "Values": values,
        "Perms": utils.get_perms(),
        "PermGroups": rbac_permission_groups(),
        "DraftVersion": cfg["draft_version"],
        # Built-in roles (admin + openrun-*) are selectable in grants too
        "RoleNames": sorted((rbac.get("roles") or {}).keys()) + cfg.get("builtin_roles", []),
        "GroupNames": sorted((rbac.get("groups") or {}).keys()),
    }


def config_group_page_handler(req):
    # RBAC group form page, prefilled when editing
    name = utils.query_param(req, "name")
    cfg = load_rbac_config()
    values = {"name": name, "users_text": "", "is_edit": bool(name)}
    if name:
        values["users_text"] = "\n".join((cfg["rbac"].get("groups") or {}).get(name) or [])
    return config_form_data(req, "group", values, "", cfg)


def config_group_submit_handler(req):
    # POST: save a group to the staged config
    values = {
        "name": utils.query_param(req, "name").strip(),
        "users_text": utils.query_param(req, "users_text"),
        "is_edit": utils.query_param(req, "is_edit") == "true",
    }
    users = utils.parse_lines(values["users_text"])
    ret = openrun_admin.set_rbac_group(values["name"], users, utils.query_param(req, "draft_version"))
    if ret.error:
        return config_form_data(req, "group", values, ret.error)
    return form_redirect(req, req.AppPath + "/config/rbac")


def config_role_page_handler(req):
    # RBAC role form page, prefilled when editing
    name = utils.query_param(req, "name")
    cfg = load_rbac_config()
    values = {"name": name, "selected": {}, "custom_text": "", "is_edit": bool(name)}
    if name:
        # Split the role's entries into the known permission checkboxes and
        # free-form custom entries (globs, role references)
        known = {}
        for group in rbac_permission_groups():
            for p in group["permissions"]:
                known[p] = True
        custom = []
        for perm in (cfg["rbac"].get("roles") or {}).get(name) or []:
            if known.get(perm):
                values["selected"][perm] = True
            else:
                custom.append(perm)
        values["custom_text"] = "\n".join(custom)
    return config_form_data(req, "role", values, "", cfg)


def config_role_submit_handler(req):
    # POST: save a role (checkboxes + custom entries) to the staged config
    name = utils.query_param(req, "name").strip()
    perms = req.Form.get("permissions") or []
    custom_text = utils.query_param(req, "custom_text")
    values = {"name": name, "selected": {}, "custom_text": custom_text,
              "is_edit": utils.query_param(req, "is_edit") == "true"}
    for p in perms:
        values["selected"][p] = True
    all_perms = list(perms) + utils.parse_lines(custom_text)
    ret = openrun_admin.set_rbac_role(name, all_perms, utils.query_param(req, "draft_version"))
    if ret.error:
        return config_form_data(req, "role", values, ret.error)
    return form_redirect(req, req.AppPath + "/config/rbac")


def config_grant_page_handler(req):
    # RBAC grant form page, prefilled when editing by index
    index = utils.query_param(req, "index")
    cfg = load_rbac_config()
    values = {"index": index, "description": "", "users_text": "",
              "roles": {}, "targets_text": "", "is_edit": index != ""}
    grants = cfg["rbac"].get("grants") or []
    if index != "" and int(index) >= 0 and int(index) < len(grants):
        grant = grants[int(index)]
        values["description"] = grant.get("description") or ""
        values["users_text"] = "\n".join(grant.get("users") or [])
        values["targets_text"] = "\n".join(grant.get("targets") or [])
        for role in grant.get("roles") or []:
            values["roles"][role] = True
    return config_form_data(req, "grant", values, "", cfg)


def config_grant_submit_handler(req):
    # POST: add or update a grant in the staged config
    index = utils.query_param(req, "index")
    roles = req.Form.get("roles") or []
    values = {
        "index": index,
        "description": utils.query_param(req, "description").strip(),
        "users_text": utils.query_param(req, "users_text"),
        "roles": {},
        "targets_text": utils.query_param(req, "targets_text"),
        "is_edit": index != "",
    }
    for r in roles:
        values["roles"][r] = True

    users = utils.parse_lines(values["users_text"])
    targets = utils.parse_lines(values["targets_text"])
    draft_version = utils.query_param(req, "draft_version")
    if index != "":
        ret = openrun_admin.update_rbac_grant(int(index), values["description"], users,
                                              list(roles), targets, draft_version)
    else:
        ret = openrun_admin.add_rbac_grant(values["description"], users, list(roles),
                                           targets, draft_version)
    if ret.error:
        return config_form_data(req, "grant", values, ret.error)
    return form_redirect(req, req.AppPath + "/config/rbac")


def config_version_handler(req):
    # Config history page: one snapshot rendered as formatted json
    version = utils.query_param(req, "version")
    data = {
        "Title": "Config version",
        "Nav": "config",
        "Version": version,
        "Error": "",
        "Json": "",
        "Perms": utils.get_perms(),
    }
    ret = openrun.get_config_version(version)
    if ret.error:
        data["Error"] = ret.error
    else:
        data["Json"] = ret.value["json"]
    return data


# ---------- Syncs ----------


def syncs_data(req):
    # Syncs page: all sync entries with state and last run info
    query = utils.query_param(req, "query").lower()

    data = {
        "Title": "Syncs",
        "Nav": "syncs",
        "Query": query,
        "Total": 0,
        "Perms": utils.get_perms(),
        "Syncs": [],
    }

    ret = openrun.list_sync()
    if ret.error:
        data["FlashError"] = ret.error
        return data

    syncs = []
    for entry in ret.value:
        data["Total"] += 1
        # Search matches the sync file path and the creator
        if query and query not in entry["path"].lower() and \
           query not in (entry["user_id"] or "").lower():
            continue
        status = entry["status"]
        metadata = entry["metadata"]

        syncs.append({
            "id": entry["id"],
            "repo": entry["path"],
            "user": entry["user_id"] or "",
            "branch": metadata["git_branch"],
            "is_scheduled": entry["is_scheduled"],
            "schedule_frequency": metadata["schedule_frequency"],
            "flags": utils.sync_flags(metadata),
            "state": status["state"],  # Enabled / Disabled / Failing
            "clobber": metadata["clobber"],
            "commit": utils.short_sha(status["commit_id"]),
            "last_exec": utils.nonzero_time(status["last_execution_time"]),
            "error": status["error"],
            "failure_count": status["failure_count"],
        })

    data["Syncs"] = sorted(syncs, key=lambda s: s["repo"])
    return data


def syncs_create_page_handler(req):
    # Sync create form page. Promote (without verification) defaults on when
    # the user holds app:promote somewhere; approve is opt-in and the form
    # renders the options disabled without the matching permission. Git auth
    # preselects the server's default entry
    values = sync_form_values(req)
    perms = utils.get_perms()
    if perms.get("app:promote"):
        values["promote_mode"] = "noverify"
    if not values["git_auth"]:
        values["git_auth"] = git_auth_options()["default"]
    return sync_form_data(req, values, "")


def sync_form_values(req):
    # The form fields for the sync create subpage. promote_mode is the
    # promotion choice: none, verify (promote with verification) or noverify
    return {
        "path": utils.query_param(req, "path"),
        "git_branch": utils.query_param(req, "git_branch"),
        "git_auth": utils.query_param(req, "git_auth"),
        "minutes": utils.query_param(req, "minutes"),
        "promote_mode": utils.query_param(req, "promote_mode") or "none",
        "approve": utils.query_param(req, "approve"),
    }


def sync_form_data(req, values, error):
    # Page context for the sync create form
    return {
        "Title": "Add sync source",
        "Nav": "syncs",
        "Mode": "create",
        "Error": error,
        "Values": values,
        "GitAuthOptions": git_auth_options()["entries"],
        "Perms": utils.get_perms(),
    }


def syncs_create_submit_handler(req):
    # POST: validate (dry run) or create a sync entry
    values = sync_form_values(req)
    action = utils.query_param(req, "action")

    if not values["path"]:
        return sync_form_data(req, values, "Source path is required")

    minutes = 0
    if values["minutes"]:
        if not values["minutes"].isdigit():
            return sync_form_data(req, values, "Schedule minutes must be a number")
        minutes = int(values["minutes"])

    dry_run = action == "validate"
    mode = values["promote_mode"]
    ret = openrun_admin.create_sync(values["path"], git_branch=values["git_branch"],
                                  git_auth=values["git_auth"], minutes=minutes,
                                  promote=mode == "verify" or mode == "noverify",
                                  verify=mode == "verify",
                                  approve=bool(values["approve"]), dry_run=dry_run)
    if ret.error:
        return sync_form_data(req, values, ret.error)

    if dry_run:
        data = sync_form_data(req, values, "")
        data["Validated"] = True
        return data
    return form_redirect(req, req.AppPath + "/syncs")


def syncs_delete_handler(req):
    # POST: delete a sync entry from the syncs list
    sync_id = utils.query_param(req, "sync_id")
    ret = openrun_admin.delete_sync(sync_id)
    error = ret.error
    return utils.flash_result(syncs_data(req), error, "Sync source removed", "Delete failed")


def syncs_run_handler(req):
    # POST: run a sync from the syncs list
    return run_sync_action(req, syncs_data)


def syncs_detail_data(req):
    # Sync detail page: settings, status and the last invocation results
    id = utils.query_param(req, "id")
    data = {
        "Title": "Sync detail",
        "Nav": "syncs",
        "Id": id,
        "Error": "",
        "Sync": None,
        "Perms": utils.get_perms(),
        "HelpUrl": utils.docs_link("/docs/applications/overview/"),
    }

    ret = openrun.list_sync()
    if ret.error:
        data["Error"] = ret.error
        return data

    entry = None
    for candidate in ret.value:
        if candidate["id"] == id:
            entry = candidate
            break
    if not entry:
        data["Error"] = "sync entry %s not found" % id
        return data

    status = entry["status"]
    metadata = entry["metadata"]
    last_exec = utils.nonzero_time(status["last_execution_time"])

    data["Sync"] = {
        "id": entry["id"],
        "repo": entry["path"],
        "branch": metadata["git_branch"],
        "git_auth": metadata["git_auth"],
        "reload": metadata["reload"],
        "is_scheduled": entry["is_scheduled"],
        "schedule_frequency": metadata["schedule_frequency"],
        "webhook_url": metadata["webhook_url"],
        "flags": utils.sync_flags(metadata),
        "state": status["state"],
        "commit": status["commit_id"],
        "commit_short": utils.short_sha(status["commit_id"]),
        "last_exec": last_exec,
        "error": status["error"],
        "failure_count": status["failure_count"],
        "user": entry["user_id"],
        "create_time": utils.nonzero_time(entry["create_time"]),
    }
    if last_exec:
        # Details of what the last invocation applied
        data["LastResult"] = utils.sync_result_summary(status)

    # Apps last applied by this sync, filtered server side by sync_id
    apps_ret = openrun.list_apps(sync_id=id, include_internal=True)
    if apps_ret.error:
        data["AppsError"] = apps_ret.error
        data["Apps"] = []
    else:
        data["Apps"] = sorted(build_app_rows(apps_ret.value), key=lambda app: app["path"])
    return data


def syncs_detail_run_handler(req):
    # POST: run the sync from its detail page
    sync_id = utils.query_param(req, "id")
    ret = openrun_admin.run_sync(sync_id)
    error = ret.error
    data = syncs_detail_data(req)
    if error:
        data["FlashError"] = "Sync failed: %s" % error
    elif ret.value.get("error"):
        data["FlashError"] = "Sync failed: %s" % ret.value["error"]
    else:
        # The Last invocation card re-renders with the fresh stored status
        data["Flash"] = "Sync completed"
    return data


def syncs_detail_delete_handler(req):
    # POST: delete the sync and return to the syncs list
    sync_id = utils.query_param(req, "id")
    ret = openrun_admin.delete_sync(sync_id)
    if ret.error:
        data = syncs_detail_data(req)
        data["FlashError"] = "Delete failed: %s" % ret.error
        return data
    return ace.response(syncs_detail_data(req), block="sync_content",
                        redirect=req.AppPath + "/syncs")


# ---------- Secrets ----------


def secret_input_data(req):
    # Common re-render context for the secret-input component fragments: the
    # component echoes its rendering attributes (field, prefix, masked, ...)
    # so the response fragment can reproduce the element
    perms = utils.get_perms()
    return {
        "Name": utils.query_param(req, "field"),
        "AppPath": req.AppPath,
        "Prefix": utils.query_param(req, "prefix"),
        "InputId": utils.query_param(req, "input_id"),
        "Placeholder": utils.query_param(req, "placeholder"),
        "Masked": utils.query_param(req, "masked") == "true",
        "File": utils.query_param(req, "file") == "true",
        "Small": utils.query_param(req, "small") == "true",
        "Description": utils.query_param(req, "description"),
        "CanCreate": perms.get("secret:create", False),
        "CanDelete": perms.get("secret:delete", False),
    }


def secrets_store_handler(req):
    # POST from the secret-input component (console.js): encrypt the value
    # (or uploaded file content) into the db secrets provider and re-render
    # the component with the {{secret ...}} reference as its value. The
    # store dialog names the secret: store_key is an exact name (fails if it
    # already exists), else store_prefix (the dialog's edited prefix,
    # falling back to the field's default) generates the name
    data = secret_input_data(req)

    value = utils.query_param(req, "value").strip()
    value_b64 = utils.query_param(req, "value_b64")
    store_key = utils.query_param(req, "store_key").strip()
    store_prefix = utils.query_param(req, "store_prefix").strip() or data["Prefix"]
    if not store_key and not store_prefix:
        data["Error"] = "no secret name prefix is configured for this field"
        data["Value"] = value
        return data
    if not value and not value_b64:
        data["Error"] = "enter a value to store as a secret"
        return data

    if store_key:
        ret = openrun_admin.create_secret(
            value=value_b64 if value_b64 else value,
            name=store_key,
            encoding="base64" if value_b64 else "",
            description=data["Description"],
            source_file=utils.query_param(req, "source_file"))
    else:
        ret = openrun_admin.create_secret(
            value=value_b64 if value_b64 else value,
            prefix=store_prefix,
            encoding="base64" if value_b64 else "",
            description=data["Description"],
            source_file=utils.query_param(req, "source_file"))
    if ret.error:
        # The field goes back to the plain (unencrypted) value with the
        # error shown inline, e.g. when the exact name already exists
        data["Error"] = ret.error
        data["Value"] = value
        return data
    data["Value"] = ret.value["secret_ref"]
    return data


def secrets_delete_handler(req):
    # POST from the secret-input component when the user unlocks a stored
    # field and chooses to also delete the secret from the database. The
    # component parses the {{secret ...}} reference into name/provider and
    # echoes the original ref so a failure re-renders the locked state
    data = secret_input_data(req)

    name = utils.query_param(req, "name").strip()
    if not name:
        data["Error"] = "no secret name to delete"
        data["Value"] = utils.query_param(req, "ref")
        return data

    ret = openrun_admin.delete_secret(name=name, provider=utils.query_param(req, "provider"))
    error = ret.error
    if error:
        data["Error"] = error
        data["Value"] = utils.query_param(req, "ref")
        return data
    # Deleted: the field goes back to accepting a plain value
    data["Value"] = ""
    return data


# ---------- Builder ----------


def form_redirect(req, target):
    # Success navigation for the narrow-target op-forms: the htmx submit
    # (partial) needs HX-Redirect - a plain 303 would be followed by htmx
    # and swapped INTO the form slot. The no-JS fallback posts (and the API
    # tests) get a real 303
    if req.IsPartial:
        return ace.response({}, block="op_redirect", redirect=target)
    return ace.redirect(target)


def slugify(name):
    # App-name slug from a free-form session name: lowercase alnum runs
    # joined by single dashes ("Tip Calculator" -> "tip-calculator")
    out = ""
    pending_dash = False
    for ch in name.strip().lower().elems():
        if ch.isalnum():
            if pending_dash and out:
                out += "-"
            out += ch
            pending_dash = False
        else:
            pending_dash = True
    return out


def builder_publish_prefill(session_name, dest_mode, dest_target):
    # The publish dialog input prefill, derived from the session name and
    # shaped by the profile's destination mode
    slug = slugify(session_name)
    if dest_mode in ("subdomain", "path"):
        return slug
    if dest_mode == "glob":
        # prefill the fixed part of the glob plus the app name
        if "*" in dest_target:
            return dest_target.split("*")[0].rstrip("/") + "/" + slug
        return dest_target
    return "/" + slug


def builder_publish_target_path(req, config):
    # Compose the publish target from the dialog input based on the session
    # profile's destination mode. Republishes target the existing path
    if utils.query_param(req, "publish_choice").strip() == "__same__":
        return utils.query_param(req, "current_path").strip()
    value = utils.query_param(req, "publish_input").strip()
    if not value:
        return ""
    mode = config["publish_mode"]
    target = config["publish_target"]
    if mode == "subdomain":
        # The target stays RAW: a trailing "." keeps the app declaration
        # relative (portable across instances; the server resolves it to
        # this instance's default domain when operating on the app)
        if target == ".":
            return value.strip(".") + ".:/"
        return value.strip(".") + "." + target + ":/"
    if mode == "path":
        return target.rstrip("/") + "/" + value.strip("/")
    return value  # glob and unrestricted: the full path is typed


def builder_services_offer(config, profile_name):
    # The new-app Services checklist offer for a profile choice: the
    # profile's services list ("defaults" offers the default service of
    # every type, empty offers none, bare types offer the type's default
    # service), intersected with live services. profile_name "" is the
    # implicit default profile = defaults
    profile = None
    for entry in config["profiles"]:
        if entry["name"] == profile_name:
            profile = entry
    allow_defaults = profile == None
    allowed = {}
    if profile != None:
        lst = profile.get("services") or []
        if not lst:
            return []
        for e in lst:
            if e == "defaults":
                allow_defaults = True
            allowed[e] = True
    offer = []
    for svc in config["all_services"]:
        if allowed.get(svc["id"]) or (svc["is_default"] and (allow_defaults or allowed.get(svc["type"]))):
            offer.append(svc)
    return offer


def builder_effective_profile(config, chosen):
    # The profile whose services offer the create form shows initially:
    # the submitted choice, else the configured default, else the single
    # profile, else the first option the browser would select, else the
    # implicit default ("")
    if chosen:
        return chosen
    if config["default_builder_profile"]:
        return config["default_builder_profile"]
    profiles = config["profiles"]
    if len(profiles) >= 1:
        return profiles[0]["name"]
    return ""


def builder_publish_config(session_id=""):
    # Publish setup (mode, publish restriction, builder profiles). Returns
    # (config, error); config is None when the builder is not enabled server
    # side. With session_id the mode/git/publish fields reflect that
    # session's resolved builder profile
    ret = build.get_publish_config(session_id=session_id)
    error = ret.error
    if error:
        return None, error
    return ret.value, None


def builder_data(req):
    # Builder sessions list (/builder), filtered by the search query. Other
    # users' sessions are included when the caller holds the admin permission
    # (the backend enforces this; the perms map only picks the request shape)
    perms = utils.get_perms()
    query = utils.query_param(req, "query").strip().lower()
    data = {"Title": "Builder", "Nav": "builder", "Perms": perms, "Query": utils.query_param(req, "query"),
            "Sessions": [], "Enabled": False, "Flash": "", "FlashError": ""}

    if perms.get("feature:system_blocked"):
        # The build plugin rejects anonymous callers with a hard error (not
        # a ret.error), which would crash the handler into the error page -
        # losing the sidebar (and its sign-in notice), which then jumps
        # between pages. Render the page shell instead; the sidebar notice
        # explains the blocked state
        data["FlashError"] = "The builder is unavailable: management operations are disabled for anonymous users"
        return data

    config, error = builder_publish_config()
    if error:
        data["FlashError"] = error
        return data
    data["Enabled"] = config["enabled"]
    data["PublishMode"] = config["mode"]
    if not config["enabled"]:
        return data

    if perms.get("admin"):
        ret = build.list_sessions(all_users=True)
    else:
        ret = build.list_sessions()
    error = ret.error
    if error:
        data["FlashError"] = error
        return data
    sessions = ret.value
    if query:
        sessions = [s for s in sessions
                    if query in s["name"].lower() or query in s["status"].lower() or
                    query in s["preview_path"].lower() or query in s["publish_path"].lower() or
                    query in s["agent"].lower() or query in s["user_id"].lower()]
    data["Sessions"] = sessions
    return data


def builder_rows_action(req, action):
    # Row actions on the sessions list re-render the list with a flash
    id = utils.query_param(req, "id").strip()
    if not id:
        return builder_data(req)
    if action == "stop":
        ret = build.stop_session(id)
    elif action == "resume":
        ret = build.resume_session(id)
    else:
        ret = build.delete_session(id)
    error = ret.error
    data = builder_data(req)
    if error:
        data["FlashError"] = error
    else:
        data["Flash"] = "Session %s %s" % (id, "deleted" if action == "delete" else action + (
            "ped" if action == "stop" else "d"))
    return data


def builder_create_page_handler(req):
    # New app form (/builder/create): name + what the app should do, plus a
    # Builder Profile choice when two or more profiles are configured (one
    # profile applies silently; none uses the built-in default). With
    # ?edit=<path> the session modifies an existing builder-published app
    # (source only, no declaration change)
    data = {"Title": "Builder", "Nav": "builder", "Perms": utils.get_perms(),
            "Profiles": [], "Values": {},
            "EditApp": utils.query_param(req, "edit").strip()}
    config, error = builder_publish_config()
    if error:
        data["Error"] = error
        return data
    data["Enabled"] = config["enabled"]
    data["Profiles"] = config["profiles"]
    data["DefaultProfile"] = config["default_builder_profile"]
    data["ServicesOffer"] = builder_services_offer(
        config, builder_effective_profile(config, data["Values"].get("profile") or ""))
    data["ServicesChecked"] = {}

    if data["EditApp"]:
        # The workspace is seeded from the app. Apps published by the
        # builder are edited in place; other apps fork - publish creates a
        # new app with the original's settings copied
        ret = openrun.get_app(data["EditApp"])
        error = ret.error
        if error:
            data["Error"] = error
        else:
            data["EditPublished"] = ret.value.get("builder_published")
    return data


def builder_create_services_handler(req):
    # Fragment: the Services checklist re-rendered for the selected profile
    # (the create form refreshes it when the profile select changes)
    data = {"ServicesOffer": [], "ServicesChecked": {}}
    config, error = builder_publish_config()
    if not error and config["enabled"]:
        profile = utils.query_param(req, "profile").strip()
        data["ServicesOffer"] = builder_services_offer(
            config, builder_effective_profile(config, profile))
    return ace.response(data, block="builder_services_checklist")


def builder_create_submit_handler(req):
    # Create the session and go to its workspace; generation continues
    # asynchronously and streams into the chat
    data = builder_create_page_handler(req)
    name = utils.query_param(req, "name").strip()
    prompt = utils.query_param(req, "prompt").strip()
    profile = utils.query_param(req, "profile").strip()
    edit_app = utils.query_param(req, "edit_app").strip()
    services = [v.strip() for v in utils.query_param_list(req, "services") if v.strip()]
    data["EditApp"] = edit_app
    data["Values"] = {"name": name, "prompt": prompt, "profile": profile}
    checked = {}
    for svc in services:
        checked[svc] = True
    data["ServicesChecked"] = checked
    if not name or not prompt:
        data["Error"] = "Name and app description are required"
        return data

    ret = build.create_session(name=name, prompt=prompt, profile=profile, edit_app=edit_app,
                               services=services)
    error = ret.error
    if error:
        data["Error"] = error
        return data
    # Plain form post: a real redirect, not HX-Redirect
    return form_redirect(req, "%s/builder/detail?id=%s" % (req.AppPath, ret.value["id"]))


def builder_detail_data(req):
    # Session workspace (/builder/detail?id=...): transcript, preview and
    # publish state. The chat pane live-updates over the event stream; this
    # page render is the durable transcript
    id = utils.query_param(req, "id").strip()
    data = {"Title": "Builder", "Nav": "builder", "Perms": utils.get_perms(), "Id": id,
            "Flash": "", "FlashError": "", "PublishResult": None,
            # Publish dialog re-render state (set by builder_publish_form_error)
            "PublishError": "", "PublishInput": "", "PublishCommitMsg": "",
            # No dedicated builder docs page yet, link the docs root
            "HelpUrl": utils.docs_link("/docs/")}
    if not id:
        data["Error"] = "session id is required"
        return data
    if data["Perms"].get("feature:system_blocked"):
        # See builder_data: the gated build plugin would crash the handler
        data["Error"] = "The builder is unavailable: management operations are disabled for anonymous users"
        return data

    config, error = builder_publish_config(session_id=id)
    if error:
        data["Error"] = error
        return data
    data["PublishMode"] = config["mode"]
    # The session profile's publish restriction shapes the dialog input:
    # subdomain label, app name under a prefix, or a full path (glob /
    # unrestricted)
    data["PublishDestMode"] = config["publish_mode"]
    data["PublishDestTarget"] = config["publish_target"]
    data["PublishDestResolved"] = config["publish_target_resolved"]
    data["PublishDestDesc"] = config["publish_desc"]
    data["GitRepo"] = config["git_repo"]

    ret = build.get_session(id)
    error = ret.error
    if error:
        data["Error"] = error
        return data
    data["Session"] = ret.value
    data["PublishPrefill"] = builder_publish_prefill(
        ret.value["name"], config["publish_mode"], config["publish_target"])

    ret = build.get_messages(id)
    error = ret.error
    if error:
        data["Error"] = error
        return data
    # Fold runs of consecutive tool calls into one line of chips (read ×2,
    # write, edit ...) and consecutive lifecycle rows into one muted line,
    # so tool bursts and restart churn do not pad the transcript
    merged = []
    for msg in ret.value["messages"]:
        if msg["kind"] in ("tool_call", "lifecycle") and merged and merged[-1]["kind"] == msg["kind"]:
            parts = merged[-1]["parts"]
            if parts[-1]["text"] == msg["content"]:
                parts[-1]["count"] += 1
            else:
                parts.append({"text": msg["content"], "count": 1})
        else:
            entry = dict(msg)
            entry["count"] = 1
            if msg["kind"] in ("tool_call", "lifecycle"):
                entry["parts"] = [{"text": msg["content"], "count": 1}]
            merged.append(entry)
    data["Messages"] = merged
    data["IsLive"] = ret.value["is_live"]
    data["TurnActive"] = ret.value["turn_active"]
    data["Partial"] = ret.value["partial"]

    ret = build.list_files(id)
    error = ret.error
    data["Files"] = [] if error else ret.value
    data["FileTree"] = build_file_tree(data["Files"])

    # Link the chat header to the sandbox's container detail page (the
    # builder container list carries the session id in app_path). Needs the
    # containers screens enabled and a live sandbox
    data["SandboxContainerId"] = ""
    if data["IsLive"] and data["Perms"].get("feature:container"):
        ret = openrun.list_containers(type="agent")
        error = ret.error
        if not error:
            for entry in ret.value:
                if entry["app_path"] == id:
                    data["SandboxContainerId"] = entry["id"]

    # Explain a missing preview: no app.star means OpenRun cannot load the
    # workspace; a failed creation attempt is in the activity log
    data["HasAppStar"] = "app.star" in data["Files"]
    data["PreviewError"] = ""
    if not data["Session"]["preview_path"]:
        for msg in data["Messages"]:
            if msg["kind"] == "error" and "preview app" in msg["content"]:
                data["PreviewError"] = msg["content"]
    return data


def builder_detail_action(req, action):
    # Session workspace actions re-render the workspace with a flash
    id = utils.query_param(req, "id").strip()
    flash = ""
    error = None
    if action == "message":
        message = utils.query_param(req, "message").strip()
        if message:
            ret = build.send_message(id, message=message)
            if ret.error:
                # The composer posts with hx-swap=none (the transcript is
                # driven by the JSON-lines stream), so a discarded error
                # looks like a hang. Retarget it into the chat's error slot
                return ace.response({"SendError": "Message not sent: " + ret.error}, block="bc_send_error",
                                    retarget="#bc-send-error", reswap="innerHTML")
    elif action == "cancel":
        # The Stop button posts with hx-swap=none too: an error dropped with
        # the body reads as a dead button (it is reachable while the sandbox
        # is still launching, when cancel_turn rejects with guidance).
        # builderchat.js places the retargeted error and shows a "Stopping"
        # status when the response carries no HX-Retarget
        ret = build.cancel_turn(id)
        if ret.error:
            return ace.response({"SendError": "Not stopped: " + ret.error}, block="bc_send_error",
                                retarget="#bc-send-error", reswap="innerHTML")
        return ace.response({"SendError": ""}, block="bc_send_error")
    elif action == "stop":
        ret = build.stop_session(id)
        error = ret.error
        flash = "Sandbox stopped"
    elif action == "resume":
        ret = build.resume_session(id)
        error = ret.error
        flash = "Sandbox resuming"
    elif action == "approve":
        ret = build.get_session(id)
        error = ret.error
        if not error and ret.value["preview_path"]:
            approve_ret = openrun_admin.approve_apps(ret.value["preview_path"])
            error = approve_ret.error
            flash = "Preview app permissions approved"
        elif not error:
            error = "no preview app yet"

    data = builder_detail_data(req)
    if error:
        data["FlashError"] = error
    elif flash:
        data["Flash"] = flash
    return data


def builder_delete_handler(req):
    # Delete the draft (workspace, preview app, sandbox) and go back to the
    # list. Published entries stay until unpublished; the dialog says so
    id = utils.query_param(req, "id").strip()
    ret = build.delete_session(id)
    error = ret.error
    if error:
        data = builder_detail_data(req)
        data["FlashError"] = error
        return data
    data = {"Title": "Builder", "Nav": "builder", "Perms": utils.get_perms()}
    return ace.response(data, "builder_session.go.html",
                        redirect=req.AppPath + "/builder")


def builder_publish_form_error(req, error):
    # A failed publish re-renders JUST the dialog form (retargeted,
    # outerHTML swap): the dropdown stays open with the submitted values and
    # the error inline, instead of a page flash after the dialog closed
    data = builder_detail_data(req)
    data["PublishError"] = error
    data["PublishInput"] = utils.query_param(req, "publish_input").strip()
    data["PublishCommitMsg"] = utils.query_param(req, "commit_msg").strip()
    return ace.response(data, block="publish_form",
                        retarget="#publish-form", reswap="outerHTML")


def builder_publish_handler(req):
    # Publish: the dialog's single input composed per the profile's
    # destination mode (subdomain label / app name / full path)
    id = utils.query_param(req, "id").strip()
    commit_msg = utils.query_param(req, "commit_msg").strip()

    config, error = builder_publish_config(session_id=id)
    if error:
        return builder_publish_form_error(req, error)
    path = builder_publish_target_path(req, config)
    if not path:
        return builder_publish_form_error(req, "Enter a publish destination")
    ret = build.publish_app(id, path=path, commit_msg=commit_msg)
    error = ret.error
    if error:
        return builder_publish_form_error(req, error)
    data = builder_detail_data(req)
    # PublishResult renders its own success alert (with mode + commit); a
    # Flash here would be a redundant second success message
    data["PublishResult"] = ret.value
    # Local publishes land on staging (except a first publish, whose initial
    # version is live on create): offer promotion as the next step. App
    # operations use the RESOLVED path (a relative-domain declaration
    # expands to this instance's default domain)
    if ret.value.get("mode") == "local":
        app_ret = openrun.get_app(ret.value.get("resolved_path"))
        if not app_ret.error and app_ret.value.get("staged_changes"):
            data["AskPromotePath"] = ret.value.get("resolved_path")
    return data


def builder_publish_check_handler(req):
    # Validate the publish destination without publishing: the same
    # normalization, profile restriction and app RBAC checks as the real
    # publish, rendered into the dialog's #publish-check-result slot
    id = utils.query_param(req, "id").strip()
    result = {"CheckError": "", "CheckPath": "", "CheckExists": False}
    config, error = builder_publish_config(session_id=id)
    if error:
        result["CheckError"] = error
        return ace.response(result, block="publish_check_result")
    path = builder_publish_target_path(req, config)
    if not path:
        result["CheckError"] = "Enter a publish destination"
        return ace.response(result, block="publish_check_result")
    ret = build.check_publish_path(id, path=path)
    error = ret.error
    if error:
        result["CheckError"] = error
        return ace.response(result, block="publish_check_result")
    result["CheckPath"] = ret.value["path"]
    result["CheckResolved"] = ret.value["resolved"]
    result["CheckExists"] = ret.value["exists"]
    return ace.response(result, block="publish_check_result")


def builder_promote_handler(req):
    # POST: promote the just-published staging app to prod
    path, error_data = require_app_path(req, builder_detail_data)
    if error_data:
        return error_data
    return promote_app_result(req, builder_detail_data, path)


def builder_unpublish_handler(req):
    id = utils.query_param(req, "id").strip()
    ret = build.unpublish_app(id)
    error = ret.error
    data = builder_detail_data(req)
    if error:
        data["FlashError"] = error
        return data
    data["Flash"] = "Unpublished " + ret.value["publish_path"]
    return data


def build_file_tree(files):
    # Flatten the sorted file list into explorer rows: directory header rows
    # for each new directory prefix, then file rows, both carrying the
    # nesting depth for indentation
    rows = []
    seen_dirs = {}
    for path in sorted(files):
        parts = path.split("/")
        for i in range(1, len(parts)):
            dir_path = "/".join(parts[:i])
            if dir_path not in seen_dirs:
                seen_dirs[dir_path] = True
                rows.append({"name": parts[i - 1], "path": dir_path, "depth": i - 1, "is_dir": True})
        rows.append({"name": parts[-1], "path": path, "depth": len(parts) - 1, "is_dir": False})
    return rows


def builder_file_handler(req):
    # Streaming TEXT route: raw content of one workspace file, rendered by
    # the <builder-files> viewer (client side syntax highlighting)
    id = utils.query_param(req, "id").strip()
    path = utils.query_param(req, "path").strip()
    ret = build.read_file(id, path)
    if ret.error:
        return "error: %s" % ret.error
    return ret.value


def builder_download_handler(req):
    # Bundle the workspace source into a zip and stream it back to the client
    # as an attachment (chunked, no disk/db staging). Errors returned by
    # get_source_zip render the session page with a flash; later producer
    # errors are handled by the framework while writing the download
    ret = build.get_source_zip(utils.query_param(req, "id").strip())
    error = ret.error
    if error:
        data = builder_detail_data(req)
        data["FlashError"] = "Source download failed: " + error
        return data
    return ace.response(ret.value["content"], download=ret.value["name"],
                        content_type="application/zip")


def builder_events_handler(req):
    # Streaming TEXT route: session events as JSON lines, consumed by the
    # <builder-chat> element until the sandbox stops or the client leaves
    id = utils.query_param(req, "id").strip()
    ret = build.session_events(id)
    if ret.error:
        return "error: %s" % ret.error
    return ret


# ---------- Actions ----------
# The Actions page (/actions): the actions of the apps the user can use,
# and the recent runs of their async actions across apps; the action detail,
# run detail and viewer pages. Everything is read as the console user: the
# openrun.in action APIs apply the app's login provider match, app:access
# and the action's permit list to the caller (arch/docs/actions-cli-mcp.md)

ACTION_RUN_LIMIT = 500  # runs per page of the runs lists (the newest first; infinite scroll loads the next page)
ACTION_RUN_STATUSES = ["running", "succeeded", "failed", "timed_out", "canceled", "lost"]
ACTIONS_HELP = "/docs/actions/overview/"
STAGE_SUFFIX = "_cl_stage"


def action_glob(query):
    # The apps page search convention: a search containing ":", starting
    # with "/", or equal to "all" is an app path glob evaluated server-side;
    # anything else is a text search applied to the listed rows. A search
    # with a name=value term is a runs filter (run_filter_match), never a
    # glob
    stripped = query.strip()
    if "=" in stripped:
        return ""
    if ":" in stripped or stripped.startswith("/") or stripped.lower() == "all":
        return stripped
    return ""


def run_filter_match(row, query):
    # The action app's run history filter (matchesRunFilter in
    # internal/app/action/runs_ui.go): every whitespace separated term must
    # match; a name=value term an argument exactly (case-insensitive), a
    # bare term a substring of any argument value, of the result status or
    # of the failure message - and here also of the app path, the action
    # name/path and the actor, the columns of the cross-app list
    for term in query.split():
        name, sep, value = term.partition("=")
        if sep and name:
            if (row["args"].get(name) or "").lower() != value.lower():
                return False
            continue
        needle = term.lower()
        texts = list(row["args"].values()) + [row["result_status"], row["message"], row["app_path"],
                                              row["action_name"], row["action_path"], row["actor"]]
        if not any([needle in (text or "").lower() for text in texts]):
            return False
    return True


def run_page_limit(req):
    # The page size of the runs lists: ACTION_RUN_LIMIT, or a smaller
    # ?limit= (the tests page with a few runs)
    raw = utils.query_param(req, "limit")
    if raw.isdigit() and int(raw) > 0 and int(raw) < ACTION_RUN_LIMIT:
        return int(raw)
    return ACTION_RUN_LIMIT


def run_cursor(row):
    # The before cursor which continues a listing after this row (the API's
    # keyset paging position: "<unix nanos>_<id>")
    started = row["started_at"]
    if type(started) != "time.time":
        return ""
    return "%d_%s" % (started.unix_nano, row["id"])


def args_summary(args):
    # The arguments of a run as "name=value" chips for the runs table: the
    # first few (sorted by name, values shortened) and the count of the rest
    names = sorted(args.keys())
    chips = []
    for name in names[:3]:
        value = str(args[name])
        if len(value) > 24:
            value = value[:24] + "…"
        chips.append(name + "=" + value)
    return {"chips": chips, "more": len(names) - len(chips),
            "all": ", ".join([name + "=" + str(args[name]) for name in names])}


def action_search_match(needle, fields):
    if not needle:
        return True
    for field in fields:
        if needle in (field or "").lower():
            return True
    return False


def action_hint_labels(hints):
    # The declared side-effect hints of an action as badge labels
    labels = []
    if not hints:
        return labels
    for key, label in [("read_only", "read-only"), ("destructive", "destructive"),
                       ("idempotent", "idempotent"), ("open_world", "open-world")]:
        if hints.get(key):
            labels.append(label)
    return labels


def action_row(info):
    # One action of the list / the detail header (an ActionInfo)
    hints = info.get("hints") or {}
    return {
        "app_path": info["app_path"],
        "name": info["name"],
        "tool": info["tool"],
        "path": info["path"],
        "description": info.get("description") or "",
        "suggest": info.get("suggest") or False,
        "async": info.get("async") or False,
        "hints": action_hint_labels(hints),
        "destructive": bool(hints.get("destructive")),
    }


def action_param_row(p):
    # One param of the detail page: the default is shown as text, masked for
    # password params
    default = p.get("default")
    display_type = p.get("display_type") or ""
    if default == None:
        default_text = ""
    elif display_type == "PASSWORD":
        default_text = "••••••"
    else:
        default_text = str(default)
    return {
        "name": p["name"],
        "type": p.get("type") or "",
        "description": p.get("description") or "",
        "default": default_text,
        "required": p.get("required") or False,
        "display_type": display_type,
        "options": list(p.get("options") or []),
    }


def actions_list_data(query):
    # The Actions tab: the actions the user can run, grouped by app (app
    # path order; the actions of an app in their declared order). A glob
    # search is passed to list_actions (server-side app matching), a
    # substring search filters name/tool/path/description/app path here.
    # Warnings name the apps whose actions could not be read (a broken app)
    glob = action_glob(query)
    needle = "" if glob else query.strip().lower()
    listing = {"Groups": [], "Total": 0, "Shown": 0, "Warnings": [], "Error": ""}
    ret = openrun.list_actions(path=glob or "all")
    error = ret.error
    if error:
        listing["Error"] = error
        return listing
    groups = {}
    for info in ret.value.get("actions") or []:
        listing["Total"] += 1
        row = action_row(info)
        if not action_search_match(needle, [row["app_path"], row["name"], row["tool"], row["path"], row["description"]]):
            continue
        listing["Shown"] += 1
        if row["app_path"] not in groups:
            groups[row["app_path"]] = {"path": row["app_path"], "actions": []}
            listing["Groups"].append(groups[row["app_path"]])
        groups[row["app_path"]]["actions"].append(row)
    listing["Warnings"] = list(ret.value.get("warnings") or [])
    return listing


def action_main_path(app_path):
    # The main app path of a run's app instance path (a staging instance's
    # own path carries the stage suffix), for the app detail links
    if app_path.endswith(STAGE_SUFFIX):
        return app_path[:-len(STAGE_SUFFIX)]
    return app_path


def action_run_row(run):
    # One async run of the runs tables and the run detail header (an
    # ActionRun basic view). started_at stays a time.time value (the
    # templates render it relative); the unset ended_at of an active run
    # maps to "" and leaves the duration empty
    started = utils.nonzero_time(run.get("started_at"))
    ended = utils.nonzero_time(run.get("ended_at"))
    exit_code = run.get("exit_code")
    status = run.get("status") or ""
    app_path = run.get("app_path") or ""
    return {
        "id": run["id"],
        "app_path": app_path,
        # The API names the main app of a staging instance's run; older
        # servers do not, the suffix form of the staging path is derived
        "main_path": run.get("main_app_path") or action_main_path(app_path),
        "action_path": run.get("action_path") or "",
        "action_name": run.get("action_name") or "",
        "source": run.get("source") or "",
        "actor": run.get("actor") or "",
        "request_id": run.get("request_id") or "",
        "stage": job_stage_of(run.get("app_id") or ""),
        # The links to the run's action and its page name the main app path
        # plus this flag: the staging instance is another app (its own id
        # and url), a run of it is not found through the prod action
        "stage_flag": "1" if job_stage_of(run.get("app_id") or "") == "stage" else "",
        "status": status,
        "active": status == "running",
        "started_at": started,
        "ended_at": ended,
        "duration": job_duration(started, ended),
        "exit_code": exit_code,
        "has_exit_code": exit_code != None,
        "message": run.get("message") or "",
        "args": run.get("args") or {},
        "args_summary": args_summary(run.get("args") or {}),
        "is_stream": run.get("is_stream") or False,
        "result_status": run.get("result_status") or "",
        "result_rows": run.get("result_rows") or 0,
        "output_bytes": run.get("output_bytes") or 0,
        "version": run.get("version") or 0,
        "node_id": run.get("node_id") or "",
    }


def action_runs_data(req, user_id, path="", action="", stage=False):
    # The runs table, newest first, one page (ACTION_RUN_LIMIT) at a time:
    # across the apps the user can use (the Runs tab; the search box glob
    # narrows the apps) or of one action (the action detail page: path +
    # action). ?status= filters server-side; ?mine=1 keeps the runs the
    # current user started and the search text is the run history filter
    # (run_filter_match), both applied to the page's rows. stage reads the
    # staging instance(s): the detail page of a staging run's action.
    # Infinite scroll: ?before= (the cursor of a page's last run) continues
    # the listing; a full page sets NextPage, the url the last row loads
    # when revealed (the /actions/runs_page fragment renders rows only)
    status = utils.query_param(req, "status")
    mine = utils.query_param(req, "mine") == "1"
    query = utils.query_param(req, "query")
    before = utils.query_param(req, "before")
    limit = run_page_limit(req)
    glob = "" if path else action_glob(query)
    text = "" if glob else query.strip()
    runs = {"Status": status, "Mine": mine, "Query": text, "Statuses": ACTION_RUN_STATUSES, "Rows": [],
            "Error": "", "Active": False, "Total": 0, "Warnings": [], "UserId": user_id, "NextPage": "",
            "IsMore": bool(before)}
    # Across apps the prod AND the staging instances are listed (one call
    # each, the API reads one instance kind per call), merged newest first;
    # one action's runs are those of the instance asked for
    instances = [stage] if path else [False, True]
    listed = []
    for stage_instance in instances:
        ret = openrun.list_action_runs(path=path or glob or "all", action=action, status=status,
                                       stage=stage_instance, limit=limit, before=before)
        error = ret.error
        if error:
            runs["Error"] = error
            return runs
        listed += [action_run_row(run) for run in ret.value.get("runs") or []]
        runs["Warnings"] += list(ret.value.get("warnings") or [])
    if len(instances) > 1:
        listed = utils.sort_recent(listed, "started_at", "id")[:limit]
    if len(listed) == limit and run_cursor(listed[-1]):
        params = ["path=" + path, "action=" + action, "stage=" + ("1" if stage else ""), "status=" + status,
                  "mine=" + ("1" if mine else ""), "query=" + query, "before=" + run_cursor(listed[-1])]
        if limit != ACTION_RUN_LIMIT:
            params.append("limit=%d" % limit)
        runs["NextPage"] = req.AppPath + "/actions/runs_page?" + "&".join(params)
    for row in listed:
        runs["Total"] += 1
        if mine and row["actor"] != user_id:
            continue
        if not run_filter_match(row, text):
            continue
        if row["active"]:
            runs["Active"] = True
        runs["Rows"].append(row)
    return runs


def action_runs_page_handler(req):
    # The next page of a runs list (infinite scroll): rows only, appended
    # after the row which requested them. path+action name one action's
    # list (the detail page), else the cross-app list
    path = utils.query_param(req, "path")
    selector = utils.query_param(req, "action")
    stage = utils.query_param(req, "stage") == "1"
    return {"Path": path, "Selector": selector, "Stage": stage, "UserId": req.UserId,
            "Runs": action_runs_data(req, req.UserId, path=path, action=selector, stage=stage)}


def actions_data(req):
    # Actions page: the Actions tab (grouped by app) or the Runs tab (the
    # recent async runs across apps). Only the active tab's data is read
    query = utils.query_param(req, "query")
    tab = utils.query_param(req, "tab")
    data = {
        "Title": "Actions",
        "Nav": "actions",
        "Query": query,
        "Tab": "runs" if tab == "runs" else "",
        "Perms": utils.get_perms(),
        "UserId": req.UserId,
        "HelpUrl": utils.docs_link(ACTIONS_HELP),
        "Actions": None,
        "Runs": None,
    }
    if data["Tab"] == "runs":
        data["Runs"] = action_runs_data(req, req.UserId)
    else:
        data["Actions"] = actions_list_data(query)
    return data


def action_detail_data(req):
    # Action detail page (/actions/detail?path=&action=): the definition
    # (params, hints, the form UI url) and, for an async action, its runs.
    # action is the tool name or the action path; per-app permissions gate
    # the app detail link
    path = utils.query_param(req, "path")
    selector = utils.query_param(req, "action")
    stage = utils.query_param(req, "stage") == "1"
    data = {
        "Title": "Action detail",
        "Nav": "actions",
        "Path": path,
        "Selector": selector,
        "Stage": stage,
        "Error": "",
        "Action": None,
        "Params": [],
        "Url": "",
        "Runs": None,
        "Perms": utils.get_perms(path) if path else utils.get_perms(),
        "UserId": req.UserId,
        "HelpUrl": utils.docs_link(ACTIONS_HELP),
    }
    if not path:
        data["Error"] = "app path is required"
        return data
    ret = openrun.get_action(path, action=selector, stage=stage)
    error = ret.error
    if error:
        data["Error"] = error
        return data
    detail = ret.value
    data["Action"] = action_row(detail)
    data["Selector"] = data["Action"]["tool"]
    data["Title"] = data["Action"]["name"]
    data["Params"] = [action_param_row(p) for p in detail.get("params") or []]
    data["Url"] = detail.get("url") or ""
    if data["Action"]["async"]:
        data["Runs"] = action_runs_data(req, req.UserId, path=path, action=data["Action"]["tool"], stage=stage)
    return data


def action_detail_runs_handler(req):
    # Runs panel fragment of the detail page: the status/mine filters
    # re-render #action-runs through here
    path = utils.query_param(req, "path")
    selector = utils.query_param(req, "action")
    stage = utils.query_param(req, "stage") == "1"
    return {"Path": path, "Selector": selector, "Stage": stage, "Perms": utils.get_perms(path), "UserId": req.UserId,
            "Runs": action_runs_data(req, req.UserId, path=path, action=selector, stage=stage)}


def run_document_data(doc):
    # The bounded run document of get_action_run for the run detail page:
    # a values run's result (table rows with their column union, or text
    # lines), a stream run's output tail and exit status, param errors
    values = list(doc.get("values") or [])
    columns = []
    rows = []
    lines = []
    for v in values:
        if type(v) == "dict":
            for k in v.keys():
                if k not in columns:
                    columns.append(k)
            rows.append(v)
        else:
            lines.append(str(v))
    exit_status = doc.get("exit_status")
    return {
        "status": doc.get("status") or "",
        "report": doc.get("report") or "",
        "columns": columns,
        "rows": rows,
        "lines": lines,
        "output": doc.get("output") or "",
        "truncated": doc.get("truncated") or False,
        "exit_status": exit_status,
        "has_exit_status": exit_status != None,
        "param_errors": doc.get("param_errors") or {},
        "message": doc.get("message") or "",
    }


def action_run_detail_data(req):
    # Run detail page (/actions/runs/detail?run=): the run record, its args
    # and its result or output tail; polls while the run is active
    run_id = utils.query_param(req, "run")
    data = {
        "Title": "Action run",
        "Nav": "actions",
        "RunId": run_id,
        "Error": "",
        "Run": None,
        "Doc": None,
        "Url": "",
        "Tool": "",
        "Perms": utils.get_perms(),
        "UserId": req.UserId,
        "HelpUrl": utils.docs_link(ACTIONS_HELP),
    }
    if not run_id:
        data["Error"] = "run id is required"
        return data
    ret = openrun.get_action_run(run_id)
    error = ret.error
    if error:
        data["Error"] = error
        return data
    view = ret.value
    data["Run"] = action_run_row(view["run"])
    data["Doc"] = run_document_data(view.get("document") or {})
    data["Url"] = view.get("url") or ""
    data["Tool"] = view.get("tool") or ""
    data["Perms"] = utils.get_perms(data["Run"]["main_path"])
    return data


def action_run_cancel_handler(req):
    # POST: cancel an active run (the run's action checks apply to the
    # caller); re-renders the run detail with the outcome
    run_id = utils.query_param(req, "run")
    ret = openrun_admin.cancel_action_run(run_id)
    error = ret.error
    data = action_run_detail_data(req)
    return utils.flash_result(data, error, "Cancel requested for run %s" % run_id, "Cancel failed")


# ---------- MCP config sub page: the connect guide ----------

MCP_HELP = "/docs/configuration/remoteaccess/#connect-an-mcp-client"

# Logins the app actions endpoint can be connected with when the server's
# login list cannot be read (server_info needs config:basic_read)
MCP_FIXED_LOGINS = ["none", "system", "builtin"]


def mcp_endpoint_url(req, api, key, suffix):
    # The url of a server MCP endpoint: the server's canonical url (built
    # from the issuer origin) when one applies, else the console's own
    # origin - a plaintext loopback dev server has no issuer but serves the
    # app actions endpoint and the MCP apps on loopback all the same
    server_url = (api or {}).get(key, {}).get("url") or ""
    if server_url:
        return server_url, True
    return url_origin(req.AppUrl) + "/_openrun/" + suffix, False


MCP_APPS_LIMIT = 50


def mcp_apps_with_actions(perms):
    # The apps whose actions the console user can run, with their action
    # counts and their own MCP endpoint url: what the app actions endpoint
    # lists as tools for this user. list_actions (the plugin, like the CLI)
    # includes apps with MCP disabled - only the MCP surfaces drop them - so
    # each app is resolved through get_app for its mcp state and url, and
    # the disabled ones are left out. One get_app per app with actions,
    # capped. None when the list could not be read
    ret = openrun.list_actions(path="all")
    error = ret.error
    if error:
        return None, error
    groups = {}
    apps = []
    for info in ret.value.get("actions") or []:
        app_path = info["app_path"]
        if app_path not in groups:
            groups[app_path] = {"path": app_path, "count": 0, "url": "", "can_read": False}
            apps.append(groups[app_path])
        groups[app_path]["count"] += 1
    listed = []
    for app in apps[:MCP_APPS_LIMIT]:
        app_ret = openrun.get_app(app["path"])
        app_error = app_ret.error
        if app_error:
            # No app:read on the app: listed without its endpoint
            listed.append(app)
            continue
        if app_ret.value.get("mcp_disabled"):
            continue
        app["url"] = app_ret.value.get("mcp_url") or ""
        app["can_read"] = True
        listed.append(app)
    return listed, ""


def mcp_connect_data(req, data):
    # The connect guide of the MCP config sub page (/config/mcp): the three
    # kinds of MCP endpoint the server has - the app actions endpoint (on by
    # default, the actions of every app the user can run), the management
    # endpoint (off by default, enabled here when its prerequisites hold)
    # and the per app endpoints - each with its state and the commands which
    # connect Claude Code, Codex or any MCP client. The endpoint state comes
    # from server_info; the urls fall back to the console's origin when the
    # server has no external url
    perms = data["Perms"]
    data.update({
        "HelpUrl": utils.docs_link(MCP_HELP),
        "ActionsDocs": utils.docs_link("/docs/actions/"),
        "Api": None,
        "ApiError": "",
        "Apps": [],
        "AppsError": "",
        "Logins": MCP_FIXED_LOGINS,
        "DefaultAuth": "none",
        "Checks": [],
        "Ready": False,
        # The management enable toggle is a config:update write (get_perms
        # masks the permission without the config + updates features)
        "CanEnable": bool(perms.get("config:update")),
    })
    ret = openrun.server_info()
    error = ret.error
    if error:
        data["ApiError"] = error
    else:
        api = ret.value.get("api") or {}
        data["Api"] = api
        data["DefaultAuth"] = api.get("app_default_auth") or "none"
        logins = list(api.get("logins") or MCP_FIXED_LOGINS)
        allowed = list(api.get("app_mcp", {}).get("allowed_auth") or [])
        if allowed:
            # [api.app_mcp] allowed_auth restricts what ?auth= may name
            logins = [l for l in logins if l in allowed]
        data["Logins"] = logins
        # The management prerequisites, as validateApiSurfaceConfig checks
        # them when the surface is enabled
        transport_ok = bool(api.get("https_listener") or api.get("trusted_proxies"))
        data["Checks"] = [
            {"key": "rbac", "ok": bool(api.get("rbac_enforced")),
             "label": "RBAC enforcement",
             "detail": "always on" if api.get("rbac_enforced") else
                       "security.unsafe_disable_rbac is set on this server: unset it (static config, restart)"},
            {"key": "transport", "ok": transport_ok,
             "label": "HTTPS listener or a trusted TLS proxy",
             "detail": ("HTTPS listener on" if api.get("https_listener") else "security.trusted_proxies set") if transport_ok else
                       "https.port is -1 and security.trusted_proxies is empty (static config, restart)"},
            {"key": "external_url", "ok": bool(api.get("external_url")),
             "label": "External URL",
             "detail": api.get("external_url") or
                       "set api.external_url (or security.callback_url) to the https origin clients use, " +
                       "or turn on the HTTPS listener so https://<default domain>:<port> applies"},
        ]
        data["Ready"] = all([c["ok"] for c in data["Checks"]])
    data["AppUrl"], data["AppUrlCanonical"] = mcp_endpoint_url(req, data["Api"], "app_mcp", "app_mcp")
    data["MgmtUrl"], data["MgmtUrlCanonical"] = mcp_endpoint_url(req, data["Api"], "mcp", "mcp")
    apps, error = mcp_apps_with_actions(perms)
    if apps == None:
        data["AppsError"] = error
    else:
        data["Apps"] = apps
    return data


def mcp_enable_handler(req):
    # POST /config/mcp/enable: turn an MCP endpoint on or off - surface is
    # mcp (the management endpoint, api.mcp.enable) or app_mcp (the app
    # actions endpoint, api.app_mcp.enable), action enable or disable. A
    # dynamic config value, live immediately; the server validates the
    # management prerequisites on enable and its error names what is missing
    surface = utils.query_param(req, "surface")
    if surface not in ("mcp", "app_mcp"):
        data = config_mcp_data(req)
        data["FlashError"] = "unknown MCP endpoint %s" % surface
        return data
    enable = utils.query_param(req, "action") != "disable"
    version_id = ""
    values = openrun.get_config_values(["api"])
    if not values.error:
        version_id = values.value.get("version_id") or ""
    ret = openrun_admin.set_config_value("api", surface + ".enable", enable, version_id)
    error = ret.error
    name = "Management MCP" if surface == "mcp" else "App actions MCP"
    ok = "%s enabled - connect a client with the commands below" % name if enable else \
         "%s disabled - connected clients are refused from now on" % name
    return utils.flash_result(config_mcp_data(req), error, ok)


def url_split(url):
    # (origin, rest) of a url: the origin is scheme://host[:port] lower-cased
    # with a default port dropped (https :443, http :80) so that two
    # spellings of one browser origin compare equal - GetAppUrl always spells
    # the port, the request url usually omits a default one; rest is the
    # path (and query) after the origin
    parts = url.split("/", 3)
    if len(parts) < 3:
        return url.lower(), ""
    scheme = parts[0].lower()
    host = parts[2].lower()
    if (scheme == "https:" and host.endswith(":443")) or (scheme == "http:" and host.endswith(":80")):
        host = host.rsplit(":", 1)[0]
    rest = "/" + parts[3] if len(parts) > 3 else ""
    return scheme + "//" + host, rest


def url_origin(url):
    return url_split(url)[0]


def action_open_data(req):
    # Viewer page (/actions/open?path=&action=[&run=]): the action's form
    # UI (or one run's page) framed inside the console. A prod app on the
    # default domain is served on every host, so it is framed through the
    # console's own origin: the frame is same-origin whatever host the user
    # typed, and the browser sends the app's session cookie. An app on its
    # own domain, and a staging instance (?stage=1; served on the stage
    # subdomain), is framed only when that is the console's origin: the
    # app's X-Frame-Options: SAMEORIGIN blocks the frame otherwise and its
    # Lax session cookie would not be sent, so the page offers a new tab
    path = utils.query_param(req, "path")
    selector = utils.query_param(req, "action")
    run_id = utils.query_param(req, "run")
    stage = utils.query_param(req, "stage") == "1"
    data = {
        "Title": "Open action",
        "Nav": "actions",
        "Path": path,
        "Selector": selector,
        "RunId": run_id,
        "Stage": stage,
        "Error": "",
        "Action": None,
        "Url": "",
        "SameOrigin": False,
        "Perms": utils.get_perms(path) if path else utils.get_perms(),
        "HelpUrl": utils.docs_link(ACTIONS_HELP),
    }
    if not path:
        data["Error"] = "app path is required"
        return data
    ret = openrun.get_action(path, action=selector, stage=stage)
    error = ret.error
    if error:
        data["Error"] = error
        return data
    data["Action"] = action_row(ret.value)
    data["Selector"] = data["Action"]["tool"]
    data["Title"] = data["Action"]["name"]
    url = ret.value.get("url") or ""
    if run_id:
        url = url + "/runs/" + run_id
    console_origin = url_origin(req.AppUrl)
    app_origin, app_rest = url_split(url)
    if ":" not in path and not stage:
        # A default-domain prod app: served on every host, framed through
        # the console's own origin
        url = console_origin + app_rest
        data["SameOrigin"] = True
    else:
        # Its own domain, or the staging instance (served on the stage
        # subdomain of its domain): framed only when that is the console's
        # origin
        data["SameOrigin"] = app_origin == console_origin
    data["Url"] = url
    return data