diff --git a/.env.example b/.env.example index 62c2d42..d9a83d2 100644 --- a/.env.example +++ b/.env.example @@ -22,3 +22,23 @@ PYMTHOUSE_M2M_CLIENT_ID= PYMTHOUSE_M2M_CLIENT_SECRET= # Set to 1 for local http issuer only (not needed for https://localhost with mkcert) PYMTHOUSE_ALLOW_INSECURE_HTTP= + +# Discovery Service — full raw endpoint (as-is for gateway tokens). +# Explore uses the URL origin for `/v1/discovery/capabilities` etc. +# Aliases: DISCOVERY_URL, LIVEPEER_DISCOVERY_SERVICE_URL +DISCOVERY_SERVICE_URL=https://discovery-service-production-8955.up.railway.app/v1/discovery/raw + +# Live-runner discovery (orchestrator /discovery endpoint). +# Local example-apps stack: http://localhost:8935/discovery +RUNNER_DISCOVERY_URL=http://localhost:8935/discovery +# Accept self-signed orchestrator TLS (local/dev only). Do not enable in production. +# RUNNER_GATEWAY_ALLOW_INSECURE_TLS=1 + +# Agent MCP SSO + mint (Scenario A). Unset secret/allowlist → mint route 404. +# MCP_INTERNAL_MINT_SECRET= +# MCP_INTERNAL_MINT_ALLOWLIST=https://agent.livepeer.org +# Optional explicit callback URLs; else {allowlist origin}/api/mcp/oauth/callback +# MCP_OAUTH_REDIRECT_ALLOWLIST=https://agent.livepeer.org/api/mcp/oauth/callback +# MCP_OAUTH_BRIDGE_SECRET= +# Non-prod mint requires PYMTHOUSE_PUBLIC_CLIENT_ID=app_98575870d7ae33589a3f0660 + diff --git a/app/(app)/apps/[id]/page.tsx b/app/(app)/apps/[...id]/page.tsx similarity index 86% rename from app/(app)/apps/[id]/page.tsx rename to app/(app)/apps/[...id]/page.tsx index 8d75aaf..1f99797 100644 --- a/app/(app)/apps/[id]/page.tsx +++ b/app/(app)/apps/[...id]/page.tsx @@ -22,7 +22,6 @@ import KeyBadge from "@/components/console/KeyBadge"; import CallsTable from "@/components/console/CallsTable"; import StatusDot from "@/components/console/StatusDot"; import { - getAppById, effectiveVisibility, setPipelineVisibility, organizationSlug, @@ -30,6 +29,8 @@ import { SETTINGS_API_KEYS, MOCK_RECENT_REQUESTS, } from "@/lib/console/mock-data"; +import { useDiscoveryModel } from "@/lib/console/useDiscoveryModel"; +import ConsolePageSkeleton from "@/components/console/ConsolePageSkeleton"; import { getAppIcon } from "@/lib/console/utils"; import PlaygroundForm from "@/components/console/playground/PlaygroundForm"; import JsonInput from "@/components/console/playground/JsonInput"; @@ -37,6 +38,15 @@ import PlaygroundOutput from "@/components/console/playground/PlaygroundOutput"; import TranscodingOutput from "@/components/console/playground/TranscodingOutput"; import CodeSnippets from "@/components/console/playground/CodeSnippets"; import WebcamPlayground from "@/components/console/playground/WebcamPlayground"; +import { + RunnerGatewayProvider, + useRunnerGatewayContext, +} from "@/components/console/playground/RunnerGatewayContext"; +import { + buildLiveRunnerPayload, + extractRunnerResultText, + runnerGatewayPostUrl, +} from "@/lib/console/runner-gateway-client"; import AppAnalytics from "@/components/console/stats/AppAnalytics"; import { OverviewTab, SettingsTab } from "@/components/console/AppDetailView"; import type { App, Pipeline, PipelineVisibility } from "@/lib/console/types"; @@ -103,6 +113,19 @@ function modelMatchesRow(catalogId: string, runModel: string): boolean { // ─── Playground Tab ─── function PlaygroundTab({ model }: { model: App }) { + return ( + + + + ); +} + +function PlaygroundTabInner({ model }: { model: App }) { + const { + canRunLive, + state: runnerGatewayState, + signerJwt, + } = useRunnerGatewayContext(); const [inputMode, setInputMode] = useState< "form" | "json" | "python" | "node" | "http" >("form"); @@ -113,12 +136,14 @@ function PlaygroundTab({ model }: { model: App }) { string, unknown > | null>(null); + const [runError, setRunError] = useState(null); - const handleRun = useCallback( + const runMock = useCallback( (values: Record) => { setLastRunValues(values); setIsRunning(true); setResult(null); + setRunError(null); const time = 0.3 + Math.random() * 1.5; setTimeout(() => { setIsRunning(false); @@ -158,6 +183,87 @@ function PlaygroundTab({ model }: { model: App }) { [model] ); + const runLive = useCallback( + async (values: Record) => { + if (runnerGatewayState.status !== "ready") { + runMock(values); + return; + } + + setLastRunValues(values); + setIsRunning(true); + setResult(null); + setRunError(null); + const started = performance.now(); + + try { + const runnerPath = + model.playgroundConfig?.runnerPath?.trim() || "chat/completions"; + const payload = buildLiveRunnerPayload(model, values); + const url = runnerGatewayPostUrl( + runnerGatewayState.gatewayBaseUrl, + runnerGatewayState.runnerAppId, + runnerPath + ); + const response = await fetch(url, { + method: "POST", + credentials: "include", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify(payload), + }); + + const contentType = response.headers.get("content-type") ?? ""; + if (!response.ok) { + let message = `Gateway error (${response.status})`; + try { + const errBody = (await response.json()) as { error?: string }; + if (errBody.error) message = errBody.error; + } catch { + // ignore + } + throw new Error(message); + } + + if (contentType.includes("text/event-stream") && response.body) { + const reader = response.body.getReader(); + const decoder = new TextDecoder(); + let streamed = ""; + while (true) { + const { done, value } = await reader.read(); + if (done) break; + streamed += decoder.decode(value, { stream: true }); + setResult(streamed); + } + } else { + const data = await response.json(); + setResult(extractRunnerResultText(data)); + } + + setInferenceTime( + parseFloat(((performance.now() - started) / 1000).toFixed(1)) + ); + } catch (error) { + const message = error instanceof Error ? error.message : "Run failed"; + setRunError(message); + setResult(null); + } finally { + setIsRunning(false); + } + }, + [model, runMock, runnerGatewayState] + ); + + const handleRun = useCallback( + (values: Record) => { + if (canRunLive) { + void runLive(values); + return; + } + runMock(values); + }, + [canRunLive, runLive, runMock] + ); + // Ctrl+Enter shortcut useEffect(() => { const handler = (e: KeyboardEvent) => { @@ -233,6 +339,7 @@ function PlaygroundTab({ model }: { model: App }) { config={model.playgroundConfig} onRun={handleRun} isRunning={isRunning} + signerJwt={signerJwt} /> )} {inputMode === "json" && ( @@ -246,8 +353,20 @@ function PlaygroundTab({ model }: { model: App }) { inputMode === "node" || inputMode === "http") && (
+ {signerJwt ? ( + + ) : null}
- +
+
+ ); +} + function ExplorePageInner() { + const exploreState = useExploreModels(); + const { status, models, reload } = exploreState; const searchParams = useSearchParams(); const initialCategory = (() => { const qp = searchParams.get("category"); @@ -449,37 +466,13 @@ function ExplorePageInner() { const [priceMin, setPriceMin] = useState(0); const [priceMax, setPriceMax] = useState(100); - // The org's public deployed apps are listed in Explore alongside the - // third-party catalog models. Seeded SSR-safely, then refreshed from the - // localStorage-backed publish state after mount so toggling an app's - // visibility on its Settings tab is reflected here on next navigation. - const [pipelineModels, setPipelineModels] = useState( - SEED_PUBLIC_PIPELINE_APPS - ); - useEffect(() => { - setPipelineModels(publicPipelines()); - }, []); - - // APPS now carries the org's own apps too (public + private). Take the - // third-party catalog from APPS and re-attach only the *public* owned apps so - // private deployments never leak into Explore and nothing is duplicated. - const catalogModels = useMemo( - () => APPS.filter((m) => !PIPELINE_APP_IDS.has(m.id)), - [] - ); - - const allModels = useMemo( - () => [...catalogModels, ...pipelineModels], - [catalogModels, pipelineModels] - ); - const dataMaxPrice = useMemo( - () => Math.max(...allModels.map((m) => m.pricing.amount), 0.01), - [allModels] + () => Math.max(...models.map((m) => m.pricing.amount), 0.01), + [models] ); const filtered = useMemo(() => { - const result = allModels.filter((m) => { + const result = models.filter((m) => { if (availabilityFilter === "warm" && m.status !== "hot") return false; if (availabilityFilter === "cold" && m.status !== "cold") return false; if (favoritesOnly && !isStarred(m.id)) return false; @@ -509,7 +502,7 @@ function ExplorePageInner() { return result; }, [ - allModels, + models, search, category, availabilityFilter, @@ -520,6 +513,36 @@ function ExplorePageInner() { dataMaxPrice, ]); + if (status === "loading" && models.length === 0) { + return ( +
+ + +
+ ); + } + + if (status === "error") { + return ( +
+ + +
+ ); + } + const activeFilters = [ ...(category ? [{ label: category, onClear: () => setCategory(null) }] @@ -713,20 +736,9 @@ function ExplorePageInner() {
) : view === "grid" ? (
- {filtered.map((model) => { - const isPipeline = PIPELINE_APP_IDS.has(model.id); - // Pipeline cards open the consumer/playground face (/apps/[id]); - // owners reach the operator console from a "Manage app" affordance - // there. The catalog is a consume surface, so a card never drops a - // caller straight into someone's operator view. - return ( - - ); - })} + {filtered.map((model) => ( + + ))}
) : (
@@ -827,7 +839,7 @@ function ExplorePageInner() { setPriceMin(min); setPriceMax(max); }} - models={allModels} + models={models} />
diff --git a/components/console/LoginPage.tsx b/components/console/LoginPage.tsx index d0f9289..2cdd77d 100644 --- a/components/console/LoginPage.tsx +++ b/components/console/LoginPage.tsx @@ -38,20 +38,24 @@ interface LoginPageProps { * toggle navigates between the two routes so the URL always matches the * visible mode. */ initialMode?: "signin" | "signup"; + /** Auth0 returnTo. MCP login uses the complete route, not /home. */ + returnTo?: string; } export default function LoginPage({ initialMode = "signin", + returnTo = "/home", }: LoginPageProps = {}) { const mode = initialMode; + const encodedReturnTo = encodeURIComponent(returnTo); const loginHref = mode === "signup" - ? "/auth/login?screen_hint=signup&returnTo=/home" - : "/auth/login?returnTo=/home"; + ? `/auth/login?screen_hint=signup&returnTo=${encodedReturnTo}` + : `/auth/login?returnTo=${encodedReturnTo}`; const googleHref = mode === "signup" - ? "/auth/login?screen_hint=signup&connection=google-oauth2&returnTo=/home" - : "/auth/login?connection=google-oauth2&returnTo=/home"; + ? `/auth/login?screen_hint=signup&connection=google-oauth2&returnTo=${encodedReturnTo}` + : `/auth/login?connection=google-oauth2&returnTo=${encodedReturnTo}`; const oauthButtonClass = "inline-flex h-10 w-full items-center justify-center gap-2.5 rounded-full border border-hairline bg-transparent px-4 text-[13px] font-medium text-fg-strong transition-colors hover:border-subtle hover:bg-hover hover:text-fg focus-visible:outline-none focus-visible:ring-1 focus-visible:ring-strong"; diff --git a/components/console/PlansPanel.tsx b/components/console/PlansPanel.tsx index 7d39539..f912b1f 100644 --- a/components/console/PlansPanel.tsx +++ b/components/console/PlansPanel.tsx @@ -10,11 +10,14 @@ import type { } from "@/lib/console/pymthouse-billing"; import { defaultCancelTimingChoice, + deriveBillingPlanAction, + deriveBillingSubscriptionUiState, formatBillingPlanPrice, resolveTimingPayload, toDateInputValue, type SubscriptionTimingChoice, } from "@/lib/console/billing-subscription-state"; +import type { MeBillingSurface } from "@/lib/console/pymthouse-me-billing-bff"; import { ScheduledChangeConflictError, useBillingPlans, @@ -79,12 +82,27 @@ function clearCheckoutQueryParam(): void { ); } -export default function PlansPanel() { +export default function PlansPanel({ + sessionBilling, +}: { + sessionBilling?: { + surface: MeBillingSurface | null; + loading: boolean; + reload: () => void; + }; +} = {}) { const { isConnected } = useAuth(); const { state, reload, subscribe, changePlan } = useBillingPlans(isConnected); - const wallet = useWalletBillingState(isConnected); - const included = - wallet.state.status === "ready" + const merchant = + sessionBilling?.surface?.mode === "merchant" + ? sessionBilling.surface + : null; + const wallet = useWalletBillingState( + isConnected && !merchant && !sessionBilling?.loading + ); + const included = merchant + ? includedUsageSummary(merchant.state) + : wallet.state.status === "ready" ? includedUsageSummary(wallet.state.wallet.billingState) : null; const [busyPlanId, setBusyPlanId] = useState(null); @@ -123,6 +141,7 @@ export default function PlansPanel() { return; } await reload(); + sessionBilling?.reload(); setError("Plan updated."); } catch (err) { setError( @@ -175,6 +194,7 @@ export default function PlansPanel() { return; } await reload(); + sessionBilling?.reload(); const plans = state.status === "ready" ? state.plans : []; const targetPlan = plans.find((p) => p.id === planId); setError( @@ -194,18 +214,15 @@ export default function PlansPanel() { try { const plans = state.status === "ready" ? state.plans : []; const targetPlan = plans.find((p) => p.id === planId); - const activePlanId = - state.status === "ready" ? state.subscription?.planId : null; - const activeStatus = - state.status === "ready" - ? (state.subscription?.status?.toLowerCase() ?? "") - : ""; - const hasActiveSubscription = - Boolean(activePlanId) && - (activeStatus === "active" || - activeStatus === "pending" || - activeStatus === "trialing" || - activeStatus === "scheduled"); + const liveSub = state.status === "ready" ? state.subscription : null; + const subscriptionUi = deriveBillingSubscriptionUiState(liveSub); + const action = deriveBillingPlanAction(subscriptionUi, planId); + const hasActiveSubscription = subscriptionUi.kind !== "none"; + + if (action === "current") { + setBusyPlanId(null); + return; + } if (hasActiveSubscription && targetPlan?.isStarterDefault === true) { setBusyPlanId(null); @@ -229,6 +246,7 @@ export default function PlansPanel() { return; } await reload(); + sessionBilling?.reload(); setError( targetPlan && isUsagePlan(targetPlan) ? "Plan updated. Add a payment method in Settings → Billing for pay-per-use auto-debit." @@ -306,23 +324,26 @@ export default function PlansPanel() { return null; } - const activePlanId = state.subscription?.planId ?? null; - const activeStatus = state.subscription?.status?.toLowerCase() ?? ""; - const hasActiveSubscription = - Boolean(activePlanId) && - (activeStatus === "active" || - activeStatus === "pending" || - activeStatus === "trialing" || - activeStatus === "scheduled"); + const subscriptionUiState = deriveBillingSubscriptionUiState( + state.subscription + ); + const hasActiveSubscription = subscriptionUiState.kind !== "none"; + const planSub = + included && !included.sharedWithApp + ? includedUsageRemainingLabel(included) + : state.subscription?.planName?.trim() || + (subscriptionUiState.kind === "pending" + ? "Payment needs to be completed" + : subscriptionUiState.kind === "active" + ? "Current subscription" + : "Subscribe via PymtHouse → Stripe Checkout"); return (

Plans

- {included - ? includedUsageRemainingLabel(included) - : "Subscribe via PymtHouse → Stripe Checkout"} + {planSub}

{flash === "success" ? (

@@ -339,7 +360,8 @@ export default function PlansPanel() {

    {state.plans.map((plan) => { - const isCurrent = hasActiveSubscription && plan.id === activePlanId; + const action = deriveBillingPlanAction(subscriptionUiState, plan.id); + const isCurrent = action === "current"; return (
  • - {isCurrent && included && included.planId === plan.id ? ( + {isCurrent && included && !included.sharedWithApp && included.planId === plan.id ? (

    ${included.remainingUsd} of ${included.totalUsd} included left diff --git a/components/console/SidebarUsageCard.tsx b/components/console/SidebarUsageCard.tsx index 2983af5..3c2e5ec 100644 --- a/components/console/SidebarUsageCard.tsx +++ b/components/console/SidebarUsageCard.tsx @@ -4,30 +4,55 @@ import type { ReactNode } from "react"; import Link from "next/link"; import { useAuth } from "@/components/console/AuthContext"; import { useAccountUsage } from "@/lib/console/useAccountUsage"; +import { useMeBillingSurface } from "@/lib/console/useMeBillingSurface"; import { useWalletBillingState } from "@/lib/console/useOwnerWallet"; import { formatPeriodResetLabel, microsToUsd, } from "@/lib/console/usage-capability-display"; -import { includedUsageSummary } from "@/lib/console/wallet-settlement-display"; +import { includedUsageSummary, sharedPoolUsageMeter } from "@/lib/console/wallet-settlement-display"; /** - * Sidebar usage meter. Remaining included usage comes from the wallet - * billing state when the live plan has an allowance; otherwise period spend. + * Sidebar usage meter. Merchant apps use this session's `/me/billing/state`. + * Owner-rollup apps use this user's spend vs the owner's remaining included. */ export default function SidebarUsageCard() { const { isConnected } = useAuth(); const usage = useAccountUsage(isConnected, 30); + const meBilling = useMeBillingSurface(isConnected); const wallet = useWalletBillingState(isConnected); - const included = - wallet.state.status === "ready" - ? includedUsageSummary(wallet.state.wallet.billingState) + const merchantIncluded = + meBilling.state.status === "ready" && + meBilling.state.surface.mode === "merchant" && + meBilling.state.surface.state + ? includedUsageSummary(meBilling.state.surface.state) : null; + const ownerRollup = + meBilling.state.status === "ready" && + meBilling.state.surface.mode === "owner_rollup"; + const actorUsdMicros = + usage.status === "ready" + ? usage.data.current.endUserBillableUsdMicros || + usage.data.current.networkFeeUsdMicros || + "0" + : "0"; + const poolMeter = + ownerRollup && wallet.state.status === "ready" + ? sharedPoolUsageMeter({ + state: wallet.state.wallet.billingState, + actorUsdMicros, + }) + : null; + const included = merchantIncluded; if ( usage.status === "loading" || usage.status === "idle" || (isConnected && + (meBilling.state.status === "loading" || + meBilling.state.status === "idle")) || + (isConnected && + ownerRollup && (wallet.state.status === "loading" || wallet.state.status === "idle")) ) { return ( @@ -57,7 +82,8 @@ export default function SidebarUsageCard() { } const { data } = usage; - const showUsdAllowance = Boolean(included); + const showUsdAllowance = Boolean(included) && !poolMeter; + const showPoolMeter = Boolean(poolMeter); const resetsAt = included?.resetsAt ? new Date(included.resetsAt).toLocaleDateString(undefined, { @@ -65,15 +91,31 @@ export default function SidebarUsageCard() { day: "numeric", }) : formatPeriodResetLabel(data.period.end); - const planLabel = - included?.planName?.trim() || (showUsdAllowance ? "Included usage" : "Usage"); + const planLabel = poolMeter + ? "Your usage" + : included?.planName?.trim() || (showUsdAllowance ? "Included usage" : "Usage"); let primaryUsed: number; let primaryLimit: number | null; let primaryDisplay: ReactNode; let footerLeft: string; - if (showUsdAllowance && included) { + if (showPoolMeter && poolMeter) { + const available = BigInt(poolMeter.availableUsdMicros || "0"); + const actor = BigInt(poolMeter.actorUsdMicros || "0"); + primaryUsed = + available > BigInt(0) + ? Number((actor * BigInt(10000)) / available) / 100 + : 0; + primaryLimit = 100; + primaryDisplay = ( + <> + ${poolMeter.actorUsd} + / ${poolMeter.availableUsd} + + ); + footerLeft = "available"; + } else if (showUsdAllowance && included) { const granted = BigInt(included.totalUsdMicros || "1"); const consumed = BigInt(included.consumedUsdMicros || "0"); primaryUsed = Number((consumed * BigInt(10000)) / granted) / 100; diff --git a/components/console/UsageView.tsx b/components/console/UsageView.tsx index d74cfb1..e0c0093 100644 --- a/components/console/UsageView.tsx +++ b/components/console/UsageView.tsx @@ -17,11 +17,14 @@ import { import ConsolePageSkeleton from "@/components/console/ConsolePageSkeleton"; import PlansPanel from "@/components/console/PlansPanel"; import WalletPanel from "@/components/console/WalletPanel"; -import { useWalletBillingState } from "@/lib/console/useOwnerWallet"; import { includedUsageSummary, + sharedPoolUsageMeter, type IncludedUsageSummary, + type SharedPoolUsageMeter, } from "@/lib/console/wallet-settlement-display"; +import { useMeBillingSurface } from "@/lib/console/useMeBillingSurface"; +import { useWalletBillingState } from "@/lib/console/useOwnerWallet"; const PERIOD_DAYS = 30; @@ -57,6 +60,9 @@ function AllowanceStrip({ requestCount, requestLimit, included, + poolMeter, + posture, + ownSpendUsdMicros, hasAccess, forecast, willExceed, @@ -68,6 +74,11 @@ function AllowanceStrip({ requestCount: number; requestLimit: number | null; included: IncludedUsageSummary | null; + /** Owner-rollup: this user's spend vs remaining pool (not grant total). */ + poolMeter: SharedPoolUsageMeter | null; + /** Set on owner_rollup. */ + posture: { label: string; canSpend: boolean } | null; + ownSpendUsdMicros: string; hasAccess: boolean; forecast: number; willExceed: boolean; @@ -76,19 +87,29 @@ function AllowanceStrip({ periodDelta: number; resetsAt: string; }) { - const showUsdAllowance = Boolean(included); + const showUsdAllowance = Boolean(included) && !poolMeter; const usedUsd = included?.consumedUsdMicros ?? "0"; const granted = included?.totalUsdMicros ?? "0"; const remaining = included?.remainingUsdMicros ?? "0"; const allowanceLabel = included?.planName?.trim() ? `${included.planName} included` : "Included this period"; - - const usedForBar = showUsdAllowance - ? Number((BigInt(usedUsd) * BigInt(10000)) / BigInt(granted || "1")) - : requestLimit - ? (requestCount / requestLimit) * 100 - : 0; + const actorLimit = poolMeter + ? Number( + (BigInt(poolMeter.availableUsdMicros || "0") > BigInt(0) + ? (BigInt(poolMeter.actorUsdMicros || "0") * BigInt(10000)) / + BigInt(poolMeter.availableUsdMicros || "1") + : BigInt(0)) + ) / 100 + : 0; + + const usedForBar = poolMeter + ? actorLimit + : showUsdAllowance + ? Number((BigInt(usedUsd) * BigInt(10000)) / BigInt(granted || "1")) + : requestLimit + ? (requestCount / requestLimit) * 100 + : 0; const pct = Math.min(100, usedForBar); const forecastPct = requestLimit ? Math.min(100, (forecast / requestLimit) * 100) @@ -98,16 +119,41 @@ function AllowanceStrip({

    - {showUsdAllowance ? allowanceLabel : "Usage this period"} + {poolMeter + ? "Your usage this period" + : showUsdAllowance + ? allowanceLabel + : "Your usage this period"}

    {showUsdAllowance && !hasAccess && ( Exhausted )} + {posture && ( + + {posture.label} + + )}
    - {showUsdAllowance ? ( + {poolMeter ? ( + + + ${poolMeter.actorUsd} + + + {" "} + / ${poolMeter.availableUsd} available + + + ) : showUsdAllowance ? ( ${microsToUsdDisplay(remaining)} @@ -134,7 +180,7 @@ function AllowanceStrip({
    - {(showUsdAllowance || requestLimit) && ( + {(showUsdAllowance || poolMeter || requestLimit) && (
    {fmt(requestCount)} signed requests this period - {showUsdAllowance && ( + {showUsdAllowance ? ( <> {" "} ·{" "} @@ -178,6 +224,15 @@ function AllowanceStrip({ {" "} consumed + ) : ( + <> + {" "} + ·{" "} + + ${microsToUsdDisplay(ownSpendUsdMicros)} + {" "} + your spend + )} )} @@ -194,6 +249,7 @@ export default function UsageView() { const { isConnected, user } = useAuth(); const usageState = useAccountUsage(isConnected, PERIOD_DAYS); const walletState = useWalletBillingState(isConnected); + const meBilling = useMeBillingSurface(isConnected); const [priceMin, setPriceMin] = useState(0); const [priceMax, setPriceMax] = useState(100); @@ -299,10 +355,29 @@ export default function UsageView() { const { data } = usageState; const grandReq = filteredRows.reduce((a, c) => a + c.requestCount, 0); const grandSpend = filteredRows.reduce((a, c) => a + c.spendUsd, 0); - const included: IncludedUsageSummary | null = - walletState.state.status === "ready" - ? includedUsageSummary(walletState.state.wallet.billingState) - : null; + + const actorUsdMicros = + data.current.endUserBillableUsdMicros || + data.current.networkFeeUsdMicros || + "0"; + let included: IncludedUsageSummary | null = null; + let billedToOwner = false; + let poolMeter: SharedPoolUsageMeter | null = null; + if (meBilling.state.status === "ready") { + const surface = meBilling.state.surface; + if (surface.mode === "owner_rollup") { + billedToOwner = true; + if (walletState.state.status === "ready") { + poolMeter = sharedPoolUsageMeter({ + state: walletState.state.wallet.billingState, + actorUsdMicros, + }); + } + } else if (surface.state) { + included = includedUsageSummary(surface.state); + } + } + const resetsAt = included?.resetsAt ? new Date(included.resetsAt).toLocaleDateString(undefined, { month: "short", @@ -310,13 +385,32 @@ export default function UsageView() { }) : formatPeriodResetLabel(data.period.end); + const sessionBilling = + meBilling.state.status === "ready" + ? { + surface: meBilling.state.surface, + loading: false, + reload: () => { + void meBilling.reload(); + }, + } + : { + surface: null, + loading: + meBilling.state.status === "idle" || + meBilling.state.status === "loading", + reload: () => { + void meBilling.reload(); + }, + }; + return (

    Account{user?.id ? ` · ${user.id}` : ""}

    - + = { export default function WalletPanel({ periodBillableUsdMicros = null, + sessionBilling, }: { /** Period end-user billable USD micros from the Usage page (metered usage, not credits). */ periodBillableUsdMicros?: string | null; + sessionBilling?: { + surface: MeBillingSurface | null; + loading: boolean; + reload: () => void; + }; }) { const { isConnected } = useAuth(); + const merchant = + sessionBilling?.surface?.mode === "merchant" + ? sessionBilling.surface + : null; + const waitingSession = + sessionBilling != null && + (sessionBilling.loading || sessionBilling.surface === null); const { state, - reload, - startTopUp, - startPaymentMethodCheckout, - ensureDefaultPaymentMethod, - } = useOwnerWallet(isConnected); + reload: reloadOwner, + startTopUp: ownerStartTopUp, + startPaymentMethodCheckout: ownerStartPm, + ensureDefaultPaymentMethod: ownerEnsureDefault, + } = useOwnerWallet(isConnected && !merchant && !waitingSession); + const checkout = useWalletCheckoutActions(); + const reload = merchant ? () => sessionBilling?.reload() : reloadOwner; + const startTopUp = merchant ? checkout.startTopUp : ownerStartTopUp; + const startPaymentMethodCheckout = merchant + ? checkout.startPaymentMethodCheckout + : ownerStartPm; + const ensureDefaultPaymentMethod = merchant + ? checkout.ensureDefaultPaymentMethod + : ownerEnsureDefault; const [showTopUp, setShowTopUp] = useState(false); const [amountUsd, setAmountUsd] = useState("25.00"); const [busy, setBusy] = useState<"topup" | "pm" | null>(null); @@ -127,7 +154,10 @@ export default function WalletPanel({ } } - if (state.status === "loading" || state.status === "idle") { + if ( + waitingSession || + (!merchant && (state.status === "loading" || state.status === "idle")) + ) { return (
    @@ -136,7 +166,7 @@ export default function WalletPanel({ ); } - if (state.status === "error") { + if (!merchant && state.status === "error") { return (

    Could not load wallet.

    @@ -153,12 +183,47 @@ export default function WalletPanel({ ); } - const { wallet, paymentMethods, invoices } = state; + if (merchant && !merchant.wallet) { + return ( +
    +

    Could not load session wallet.

    + +
    + ); + } + + const { wallet, paymentMethods, invoices } = merchant + ? { + wallet: merchant.wallet, + paymentMethods: merchant.paymentMethods, + invoices: merchant.invoices, + } + : state.status === "ready" + ? state + : { wallet: null, paymentMethods: [], invoices: [] }; + + if (!wallet) { + return null; + } const usageUsd = formatWalletUsd(periodBillableUsdMicros); const billingState = wallet.billingState; const posture = spendPostureBadge(billingState.status); const runway = availableRunway(billingState); const included = includedUsageSummary(billingState); + const poolMeter = + included?.sharedWithApp && periodBillableUsdMicros + ? sharedPoolUsageMeter({ + state: billingState, + actorUsdMicros: periodBillableUsdMicros, + }) + : null; const limitNote = overageLimitNote(billingState); const defaultPm = paymentMethods.find((pm) => pm.isDefault) ?? paymentMethods[0] ?? null; @@ -209,7 +274,20 @@ export default function WalletPanel({

    - {included ? ( + {poolMeter ? ( +

    + {poolMeter.label} + {included?.resetsAt + ? ` · resets ${new Date(included.resetsAt).toLocaleDateString( + "en-US", + { + month: "short", + day: "numeric", + } + )}` + : ""} +

    + ) : included ? (

    {includedUsageRemainingLabel(included)} {included.resetsAt diff --git a/components/console/playground/CodeSnippets.tsx b/components/console/playground/CodeSnippets.tsx index b21cc68..c98d6ea 100644 --- a/components/console/playground/CodeSnippets.tsx +++ b/components/console/playground/CodeSnippets.tsx @@ -30,13 +30,40 @@ function generateSnippets( token: string, runValues?: Record ): Record { - const baseUrl = model.apiEndpoint ?? "https://gateway.livepeer.org/v1"; + const isRunnerLlm = + model.category === "Language" && Boolean(model.runnerAppId?.trim()); + const origin = + typeof window !== "undefined" + ? window.location.origin + : "http://localhost:3000"; + + const baseUrl = isRunnerLlm + ? `${origin}/api/runner-gateway/v1` + : (model.apiEndpoint ?? "https://gateway.livepeer.org/v1"); + + const appQuery = isRunnerLlm + ? `?app=${encodeURIComponent(model.runnerAppId!.trim())}` + : ""; + const endpoint = model.category === "Language" - ? `${baseUrl}/chat/completions` + ? `${baseUrl}/chat/completions${appQuery}` : `${baseUrl}/${model.id}`; const isLLM = model.category === "Language"; + const openAiKey = isRunnerLlm ? "unused" : token; + const curlAuth = isRunnerLlm + ? ` --cookie "console session (signed in)" \\\n` + : ` -H "Authorization: Bearer ${token}" \\\n`; + const jsonAuthHeaders = isRunnerLlm + ? `"Content-Type": "application/json",` + : `"Authorization": "Bearer ${token}", + "Content-Type": "application/json",`; + const fetchAuthHeaders = isRunnerLlm + ? `"Content-Type": "application/json",` + : `Authorization: "Bearer ${token}", + "Content-Type": "application/json",`; + const credentialsOpt = isRunnerLlm ? `\n credentials: "include",` : ""; // If the user has supplied playground inputs, bake them into the request body // so "Copy code for this run" produces production code matching what they tested. @@ -66,8 +93,7 @@ function generateSnippets( return { curl: `curl -X POST "${endpoint}" \\ - -H "Authorization: Bearer ${token}" \\ - -H "Content-Type: application/json" \\ +${curlAuth} -H "Content-Type: application/json" \\ -d '${body}'`, python: useRunValues @@ -76,8 +102,7 @@ function generateSnippets( response = requests.post( "${endpoint}", headers={ - "Authorization": "Bearer ${token}", - "Content-Type": "application/json", + ${jsonAuthHeaders} }, json=${body.replace(/^/gm, " ").trimStart()}, ) @@ -87,7 +112,7 @@ print(response.json())` client = OpenAI( base_url="${baseUrl}", - api_key="${token}", + api_key="${openAiKey}", ) response = client.chat.completions.create( @@ -96,7 +121,7 @@ response = client.chat.completions.create( {"role": "user", "content": "Hello, how are you?"} ], temperature=0.7, - max_tokens=1024, + max_tokens=1024,${isRunnerLlm ? `\n extra_query={"app": "${model.runnerAppId}"},` : ""} ) print(response.choices[0].message.content)` : `import requests @@ -104,8 +129,7 @@ print(response.choices[0].message.content)` response = requests.post( "${endpoint}", headers={ - "Authorization": "Bearer ${token}", - "Content-Type": "application/json", + ${jsonAuthHeaders} }, json={ "prompt": "A scenic mountain landscape at sunset", @@ -116,10 +140,9 @@ print(response.json())`, node: useRunValues ? `const response = await fetch("${endpoint}", { - method: "POST", + method: "POST",${credentialsOpt} headers: { - Authorization: "Bearer ${token}", - "Content-Type": "application/json", + ${fetchAuthHeaders} }, body: JSON.stringify(${body.replace(/^/gm, " ").trimStart()}), }); @@ -130,7 +153,7 @@ console.log(result);` const client = new OpenAI({ baseURL: "${baseUrl}", - apiKey: "${token}", + apiKey: "${openAiKey}", }); const response = await client.chat.completions.create({ @@ -140,13 +163,12 @@ const response = await client.chat.completions.create({ ], temperature: 0.7, max_tokens: 1024, -}); +}${isRunnerLlm ? `, { query: { app: "${model.runnerAppId}" } }` : ""}); console.log(response.choices[0].message.content);` : `const response = await fetch("${endpoint}", { - method: "POST", + method: "POST",${credentialsOpt} headers: { - Authorization: "Bearer ${token}", - "Content-Type": "application/json", + ${fetchAuthHeaders} }, body: JSON.stringify({ prompt: "A scenic mountain landscape at sunset", @@ -157,9 +179,8 @@ const result = await response.json(); console.log(result);`, http: `POST ${endpoint} HTTP/1.1 -Host: ${new URL(baseUrl).host} -Authorization: Bearer ${token} -Content-Type: application/json +Host: ${new URL(baseUrl.startsWith("http") ? baseUrl : `https://${baseUrl}`).host} +${isRunnerLlm ? "Cookie: \n" : `Authorization: Bearer ${token}\n`}Content-Type: application/json ${body}`, }; diff --git a/components/console/playground/PlaygroundForm.tsx b/components/console/playground/PlaygroundForm.tsx index 6313880..86a4ac9 100644 --- a/components/console/playground/PlaygroundForm.tsx +++ b/components/console/playground/PlaygroundForm.tsx @@ -10,6 +10,8 @@ interface PlaygroundFormProps { config: PlaygroundConfig; onRun: (values: Record) => void; isRunning: boolean; + /** Short-lived signer JWT — rendered as a hidden input for live runner calls. */ + signerJwt?: string; } function TypeBadge({ type }: { type: string }) { @@ -175,6 +177,7 @@ export default function PlaygroundForm({ config, onRun, isRunning, + signerJwt, }: PlaygroundFormProps) { const getDefaults = useCallback(() => { const defaults: Record = {}; @@ -201,6 +204,9 @@ export default function PlaygroundForm({ return (

    + {signerJwt ? ( + + ) : null}
    {config.fields.map((field) => (
    diff --git a/components/console/playground/RunnerGatewayContext.tsx b/components/console/playground/RunnerGatewayContext.tsx new file mode 100644 index 0000000..ed36ce9 --- /dev/null +++ b/components/console/playground/RunnerGatewayContext.tsx @@ -0,0 +1,139 @@ +"use client"; + +import { + createContext, + useContext, + useEffect, + useMemo, + useState, + type ReactNode, +} from "react"; +import { useAuth } from "@/components/console/AuthContext"; +import type { App } from "@/lib/console/types"; + +export type RunnerGatewayState = + | { status: "idle" } + | { status: "loading" } + | { + status: "ready"; + gatewayBaseUrl: string; + runnerAppId: string; + expiresIn?: number; + /** Signer JWT minted for this session — kept in a hidden form field. */ + jwt?: string; + } + | { status: "unavailable"; reason: string } + | { status: "error"; message: string }; + +type RunnerGatewayContextValue = { + state: RunnerGatewayState; + canRunLive: boolean; + signerJwt: string | undefined; +}; + +const RunnerGatewayContext = createContext({ + state: { status: "idle" }, + canRunLive: false, + signerJwt: undefined, +}); + +export function useRunnerGatewayContext() { + return useContext(RunnerGatewayContext); +} + +function isLiveRunnerPlayground(model: App): boolean { + const runnerAppId = model.runnerAppId?.trim() ?? ""; + if (!runnerAppId) return false; + // Form-backed runners (hello-world) advertise a runnerPath; LLM runners use + // the Language category + OpenAI chat/completions path by default. + if (model.playgroundConfig?.runnerPath) return true; + return model.category === "Language"; +} + +export function RunnerGatewayProvider({ + model, + children, +}: { + model: App; + children: ReactNode; +}) { + const { isConnected, user } = useAuth(); + const [state, setState] = useState({ status: "idle" }); + + const runnerAppId = model.runnerAppId?.trim() ?? ""; + const isLiveRunner = isLiveRunnerPlayground(model); + + useEffect(() => { + if (!isLiveRunner) { + setState({ status: "idle" }); + return; + } + + if (!isConnected || !user?.id?.trim()) { + setState({ + status: "unavailable", + reason: "Sign in to run against the live runner gateway.", + }); + return; + } + + const controller = new AbortController(); + setState({ status: "loading" }); + + void (async () => { + try { + const response = await fetch("/api/pymthouse/signer-session", { + method: "POST", + credentials: "include", + headers: { "Content-Type": "application/json" }, + signal: controller.signal, + }); + const data = (await response.json()) as { + ready?: boolean; + expiresIn?: number; + jwt?: string; + error?: string; + error_description?: string; + }; + + if (!response.ok) { + const message = + data.error_description ?? + data.error ?? + "Signer session unavailable"; + setState({ status: "unavailable", reason: message }); + return; + } + + setState({ + status: "ready", + gatewayBaseUrl: "/api/runner-gateway/v1", + runnerAppId, + expiresIn: data.expiresIn, + jwt: data.jwt, + }); + } catch (error) { + if (controller.signal.aborted) return; + const message = + error instanceof Error + ? error.message + : "Failed to prepare signer session"; + setState({ status: "error", message }); + } + })(); + + return () => controller.abort(); + }, [isConnected, isLiveRunner, runnerAppId, user?.id]); + + const value = useMemo(() => { + const canRunLive = state.status === "ready"; + const signerJwt = state.status === "ready" ? state.jwt : undefined; + return { state, canRunLive, signerJwt }; + }, [state]); + + return ( + + {children} + + ); +} diff --git a/components/console/settings/BillingSection.tsx b/components/console/settings/BillingSection.tsx index 8edbc71..9cf7ffe 100644 --- a/components/console/settings/BillingSection.tsx +++ b/components/console/settings/BillingSection.tsx @@ -11,6 +11,7 @@ import { Trash2, } from "lucide-react"; import { useAuth } from "@/components/console/AuthContext"; +import EndUserMeBillingNote from "./EndUserMeBillingNote"; import Dialog from "@/components/design-system/Dialog"; import TimingChoicePanel from "@/components/console/TimingChoicePanel"; import { @@ -624,6 +625,7 @@ export default function BillingSection() { return (
    + {flash === "success" ? (

    Checkout completed — billing details refreshed. @@ -738,7 +740,8 @@ export default function BillingSection() { if ( isCurrent && included && - (included.planId === plan.id || !included.planId) + !included.sharedWithApp && + included.planId === plan.id ) { features.push( `$${included.remainingUsd} of $${included.totalUsd} included left` diff --git a/components/console/settings/EndUserMeBillingNote.tsx b/components/console/settings/EndUserMeBillingNote.tsx new file mode 100644 index 0000000..e7084a6 --- /dev/null +++ b/components/console/settings/EndUserMeBillingNote.tsx @@ -0,0 +1,35 @@ +"use client"; + +import { useEffect, useState } from "react"; +import type { MeBillingSurface } from "@/lib/console/pymthouse-me-billing-bff"; + +export default function EndUserMeBillingNote() { + const [note, setNote] = useState(null); + + useEffect(() => { + let cancelled = false; + void (async () => { + const response = await fetch("/api/pymthouse/me-billing", { + cache: "no-store", + }); + if (!response.ok || cancelled) return; + const body = (await response.json()) as MeBillingSurface; + if (cancelled) return; + if (body.mode === "owner_rollup") { + setNote( + "Your usage is billed to the app owner. This app is on owner_rollup, so end-user prepaid wallet and subscription reads are not available." + ); + } + })(); + return () => { + cancelled = true; + }; + }, []); + + if (!note) return null; + return ( +

    + {note} +

    + ); +} diff --git a/docs/SSO-MINT-OPERATOR.md b/docs/SSO-MINT-OPERATOR.md new file mode 100644 index 0000000..43250d6 --- /dev/null +++ b/docs/SSO-MINT-OPERATOR.md @@ -0,0 +1,38 @@ +# Console SSO + mint operator note + +Console is the Scenario A login + mint partner for Livepeer Agent MCP. Do not commit secret values. + +## Console env + +```bash +MCP_INTERNAL_MINT_SECRET= # shared with Storyboard +MCP_INTERNAL_MINT_ALLOWLIST=https://agent.livepeer.org +# MCP_OAUTH_REDIRECT_ALLOWLIST=https://agent.livepeer.org/api/mcp/oauth/callback +# MCP_OAUTH_BRIDGE_SECRET= # falls back to mint secret or AUTH0_SECRET +PYMTHOUSE_ISSUER_URL=https://pymthouse.com/api/v1/oidc +PYMTHOUSE_PUBLIC_CLIENT_ID=app_98575870d7ae33589a3f0660 # required in non-prod +PYMTHOUSE_M2M_CLIENT_ID=m2m_… +PYMTHOUSE_M2M_CLIENT_SECRET=pmth_cs_… +``` + +Mint is `POST /api/internal/mcp/mint`. Missing secret or empty allowlist → **404**. Wrong Bearer → **401**. Bad `Origin` / `X-Mcp-Caller-Origin` → **403**. Wrong billing app in non-prod → **503** `billing_app_mismatch`. + +Login: `GET /login?mcp_oauth=1&state=…&redirect_uri=https://agent.livepeer.org/api/mcp/oauth/callback` → Auth0 → callback with `state` + `external_user_id` (`eu_` of Auth0 `sub`, same as Console keys). + +## Storyboard / Agent + +```bash +MCP_OAUTH_ENABLED=1 +MCP_OAUTH_PROVIDER=sso_mint +SSO_MINT_ORIGIN=https:// +SSO_MINT_URL=https:///api/internal/mcp/mint +SSO_MINT_SECRET= +SSO_MINT_CALLER_ORIGIN=https://agent.livepeer.org +MCP_OAUTH_BILLING_APP_ID=app_98575870d7ae33589a3f0660 +``` + +Until `SSO_MINT_*` lands, equivalent names may be `NAAP_MCP_ORIGIN` / `NAAP_MCP_MINT_URL` / `MCP_INTERNAL_MINT_*`. + +After mint, Agent may call PymtHouse `GET /api/v1/apps/{app}/me/billing/*` with the composite Bearer. The minted JWT carries `billing_mode`. On **owner_rollup** (RS-2 default) skip money `/me/billing/*` — those routes return **403** `merchant_billing_required` because usage is billed to the app owner. Do not fall back to the M2M owner wallet (that discloses the shared pool). Merchant-mode JWTs may call the money routes. + +Do not retry those 403s via M2M `GET …/users/{id}/allowances` (that is the owner wallet). diff --git a/lib/console/billing-subscription-state.test.ts b/lib/console/billing-subscription-state.test.ts index b74a5b9..b08854f 100644 --- a/lib/console/billing-subscription-state.test.ts +++ b/lib/console/billing-subscription-state.test.ts @@ -10,6 +10,7 @@ import { formatIncludedUsdMicros, formatPendingCancelDate, isNothingToResumeError, + matchCatalogPlanId, paidCatalogPlanIds, resolveApplicablePendingCancel, resolveTimingPayload, @@ -374,3 +375,29 @@ test("formatBillingPlanPrice prefers Starter included usage over $0 fee", () => "Free included usage", ); }); + +test("matchCatalogPlanId prefers planId then sourcePlan then name", () => { + const catalog = [ + { id: "starter", name: "Starter" }, + { id: "ppu", name: "Pay per use" }, + ]; + assert.equal( + matchCatalogPlanId(catalog, { planId: "ppu", planName: "Pay per use" }), + "ppu" + ); + assert.equal( + matchCatalogPlanId(catalog, { planId: null, planName: null }, { + id: "ppu", + name: "Pay per use", + }), + "ppu" + ); + assert.equal( + matchCatalogPlanId(catalog, { planId: null, planName: "Pay per use" }), + "ppu" + ); + assert.equal( + matchCatalogPlanId(catalog, { planId: "missing", planName: null }), + null + ); +}); diff --git a/lib/console/billing-subscription-state.ts b/lib/console/billing-subscription-state.ts index 247acaf..f975216 100644 --- a/lib/console/billing-subscription-state.ts +++ b/lib/console/billing-subscription-state.ts @@ -108,6 +108,35 @@ export function deriveBillingSubscriptionUiState( return { kind: "none", planId: null }; } +/** + * Catalog row for the live subscription. Prefers `planId`, then included + * `sourcePlan.id`, then name — OpenMeter keys sometimes fail to resolve to + * Neon `plans.id`, which left every row as "Enable pay-per-use". + */ +export function matchCatalogPlanId( + plans: ReadonlyArray<{ id: string; name?: string | null }>, + subscription: { planId: string | null; planName?: string | null } | null, + sourcePlan?: { id: string | null; name: string | null } | null +): string | null { + const ids = [subscription?.planId, sourcePlan?.id] + .map((value) => value?.trim()) + .filter((value): value is string => Boolean(value)); + for (const id of ids) { + if (plans.some((plan) => plan.id === id)) return id; + } + + const names = [subscription?.planName, sourcePlan?.name] + .map((value) => value?.trim().toLowerCase()) + .filter((value): value is string => Boolean(value)); + for (const name of names) { + const hit = plans.find( + (plan) => (plan.name?.trim() || plan.id).toLowerCase() === name + ); + if (hit) return hit.id; + } + return null; +} + export function deriveBillingPlanAction( subscription: BillingSubscriptionUiState, planId: string diff --git a/lib/console/mcp-internal-mint.test.ts b/lib/console/mcp-internal-mint.test.ts new file mode 100644 index 0000000..9f13481 --- /dev/null +++ b/lib/console/mcp-internal-mint.test.ts @@ -0,0 +1,58 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import { + authorizeMcpMint, + billingAppMismatch, + mintRouteConfigured, + RS2_TEST_BILLING_APP_ID, +} from "./mcp-internal-mint.ts"; + +test("mintRouteConfigured requires secret and allowlist", () => { + delete process.env.MCP_INTERNAL_MINT_SECRET; + delete process.env.MCP_INTERNAL_MINT_ALLOWLIST; + assert.equal(mintRouteConfigured(), false); + process.env.MCP_INTERNAL_MINT_SECRET = "shared"; + process.env.MCP_INTERNAL_MINT_ALLOWLIST = "https://agent.livepeer.org"; + assert.equal(mintRouteConfigured(), true); +}); + +test("authorizeMcpMint fail-closed matrix", () => { + process.env.MCP_INTERNAL_MINT_SECRET = "shared"; + process.env.MCP_INTERNAL_MINT_ALLOWLIST = "https://agent.livepeer.org"; + assert.equal( + authorizeMcpMint({ + authorization: "Bearer wrong", + origin: null, + callerOrigin: "https://agent.livepeer.org", + }).ok, + false + ); + assert.deepEqual( + authorizeMcpMint({ + authorization: "Bearer shared", + origin: null, + callerOrigin: "https://evil.example", + }), + { ok: false, status: 403, error: "forbidden" } + ); + assert.deepEqual( + authorizeMcpMint({ + authorization: "Bearer shared", + origin: null, + callerOrigin: "https://agent.livepeer.org", + }), + { ok: true } + ); +}); + +test("billingAppMismatch pins RS-2 in non-prod", () => { + const prev = process.env.VERCEL_ENV; + process.env.VERCEL_ENV = "preview"; + process.env.PYMTHOUSE_PUBLIC_CLIENT_ID = "app_deadbeefdeadbeefdeadbeef"; + assert.equal(billingAppMismatch()?.error, "billing_app_mismatch"); + process.env.PYMTHOUSE_PUBLIC_CLIENT_ID = RS2_TEST_BILLING_APP_ID; + assert.equal(billingAppMismatch(), null); + if (prev === undefined) delete process.env.VERCEL_ENV; + else process.env.VERCEL_ENV = prev; +}); diff --git a/lib/console/mcp-internal-mint.ts b/lib/console/mcp-internal-mint.ts new file mode 100644 index 0000000..8978ea4 --- /dev/null +++ b/lib/console/mcp-internal-mint.ts @@ -0,0 +1,72 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; + +export const RS2_TEST_BILLING_APP_ID = "app_98575870d7ae33589a3f0660"; + +function parseMintAllowlist(raw: string | undefined): string[] { + return (raw ?? "") + .split(",") + .map((value) => value.trim()) + .filter(Boolean); +} + +function timingSafeEqualString(left: string, right: string): boolean { + const a = createHmac("sha256", "mcp-mint-compare").update(left).digest(); + const b = createHmac("sha256", "mcp-mint-compare").update(right).digest(); + return timingSafeEqual(a, b); +} + +export function mintRouteConfigured(): boolean { + const secret = process.env.MCP_INTERNAL_MINT_SECRET?.trim(); + const allowlist = parseMintAllowlist(process.env.MCP_INTERNAL_MINT_ALLOWLIST); + return Boolean(secret && allowlist.length > 0); +} + +export function authorizeMcpMint(input: { + authorization: string | null; + origin: string | null; + callerOrigin: string | null; +}): + | { ok: true } + | { ok: false; status: 401 | 403; error: string } { + const secret = process.env.MCP_INTERNAL_MINT_SECRET?.trim() ?? ""; + const presented = input.authorization?.startsWith("Bearer ") + ? input.authorization.slice("Bearer ".length).trim() + : ""; + if (!presented || !timingSafeEqualString(presented, secret)) { + return { ok: false, status: 401, error: "unauthorized" }; + } + const caller = (input.origin || input.callerOrigin || "").trim(); + const allowlist = parseMintAllowlist(process.env.MCP_INTERNAL_MINT_ALLOWLIST); + if (!caller || !allowlist.includes(caller)) { + return { ok: false, status: 403, error: "forbidden" }; + } + return { ok: true }; +} + +export function billingAppMismatch(): { error: string; error_description: string } | null { + if (process.env.VERCEL_ENV === "production") { + return null; + } + const publicClientId = process.env.PYMTHOUSE_PUBLIC_CLIENT_ID?.trim() ?? ""; + if (publicClientId === RS2_TEST_BILLING_APP_ID) { + return null; + } + return { + error: "billing_app_mismatch", + error_description: `Non-prod mint requires PYMTHOUSE_PUBLIC_CLIENT_ID=${RS2_TEST_BILLING_APP_ID}`, + }; +} + +export async function mintMcpCompositeKey(input: { + externalUserId: string; + email?: string; + label?: string; +}): Promise<{ apiKey: string }> { + const { createDashboardApiKey } = await import("./pymthouse-keys-bff"); + const created = await createDashboardApiKey({ + externalUserId: input.externalUserId, + email: input.email, + label: input.label?.trim() || "mcp-oauth", + }); + return { apiKey: created.apiKey }; +} diff --git a/lib/console/mcp-oauth-login-bridge.test.ts b/lib/console/mcp-oauth-login-bridge.test.ts new file mode 100644 index 0000000..9587cf5 --- /dev/null +++ b/lib/console/mcp-oauth-login-bridge.test.ts @@ -0,0 +1,74 @@ +import assert from "node:assert/strict"; +import { test } from "node:test"; + +import { + buildMcpOauthCallbackUrl, + decodeMcpOauthPendingCookie, + encodeMcpOauthPendingCookie, + isAllowedMcpRedirectUri, + parseMcpOauthLoginQuery, +} from "./mcp-oauth-login-bridge.ts"; + +const CALLBACK = "https://agent.livepeer.org/api/mcp/oauth/callback"; + +test("parseMcpOauthLoginQuery rejects evil redirect and missing state", () => { + process.env.MCP_OAUTH_REDIRECT_ALLOWLIST = CALLBACK; + assert.equal( + parseMcpOauthLoginQuery({ + mcpOauth: "1", + state: "abc", + redirectUri: "https://evil.example/cb", + }).ok, + false + ); + assert.equal( + parseMcpOauthLoginQuery({ + mcpOauth: "1", + state: "", + redirectUri: CALLBACK, + }).ok, + false + ); + assert.deepEqual( + parseMcpOauthLoginQuery({ + mcpOauth: "1", + state: "abc", + redirectUri: CALLBACK, + }), + { ok: true, pending: { state: "abc", redirectUri: CALLBACK } } + ); +}); + +test("isAllowedMcpRedirectUri derives callback from mint allowlist", () => { + delete process.env.MCP_OAUTH_REDIRECT_ALLOWLIST; + process.env.MCP_INTERNAL_MINT_ALLOWLIST = "https://agent.livepeer.org"; + assert.equal(isAllowedMcpRedirectUri(CALLBACK), true); + assert.equal(isAllowedMcpRedirectUri("https://evil.example/cb"), false); +}); + +test("pending cookie round-trips and rejects tampering", () => { + process.env.MCP_OAUTH_BRIDGE_SECRET = "test-bridge-secret"; + process.env.MCP_OAUTH_REDIRECT_ALLOWLIST = CALLBACK; + const encoded = encodeMcpOauthPendingCookie({ + state: "st-1", + redirectUri: CALLBACK, + }); + assert.deepEqual(decodeMcpOauthPendingCookie(encoded), { + state: "st-1", + redirectUri: CALLBACK, + }); + assert.equal(decodeMcpOauthPendingCookie(`${encoded}x`), null); +}); + +test("buildMcpOauthCallbackUrl echoes state and subject", () => { + const url = buildMcpOauthCallbackUrl({ + redirectUri: CALLBACK, + state: "st-1", + externalUserId: "eu_abc", + email: "user@example.com", + }); + const parsed = new URL(url); + assert.equal(parsed.searchParams.get("state"), "st-1"); + assert.equal(parsed.searchParams.get("external_user_id"), "eu_abc"); + assert.equal(parsed.searchParams.get("email"), "user@example.com"); +}); diff --git a/lib/console/mcp-oauth-login-bridge.ts b/lib/console/mcp-oauth-login-bridge.ts new file mode 100644 index 0000000..35c5d40 --- /dev/null +++ b/lib/console/mcp-oauth-login-bridge.ts @@ -0,0 +1,127 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; + +export const MCP_OAUTH_PENDING_COOKIE = "mcp_oauth_pending"; +const PENDING_TTL_MS = 10 * 60 * 1000; +const MAX_STATE_CHARS = 512; + +export type McpOauthPending = { + state: string; + redirectUri: string; +}; + +function bridgeSecret(): string { + return ( + process.env.MCP_OAUTH_BRIDGE_SECRET?.trim() || + process.env.MCP_INTERNAL_MINT_SECRET?.trim() || + process.env.AUTH0_SECRET?.trim() || + "" + ); +} + +export function parseMintAllowlist(raw: string | undefined): string[] { + return (raw ?? "") + .split(",") + .map((value) => value.trim()) + .filter(Boolean); +} + +export function mcpOauthRedirectAllowlist(): string[] { + const explicit = parseMintAllowlist(process.env.MCP_OAUTH_REDIRECT_ALLOWLIST); + if (explicit.length > 0) return explicit; + return parseMintAllowlist(process.env.MCP_INTERNAL_MINT_ALLOWLIST).map( + (origin) => `${origin.replace(/\/$/, "")}/api/mcp/oauth/callback` + ); +} + +export function isAllowedMcpRedirectUri(redirectUri: string): boolean { + return mcpOauthRedirectAllowlist().includes(redirectUri); +} + +export function parseMcpOauthLoginQuery(input: { + mcpOauth?: string; + state?: string; + redirectUri?: string; +}): { ok: true; pending: McpOauthPending } | { ok: false; error: string } { + if (input.mcpOauth !== "1") { + return { ok: false, error: "mcp_oauth_inactive" }; + } + const state = input.state?.trim() ?? ""; + const redirectUri = input.redirectUri?.trim() ?? ""; + if (!state || state.length > MAX_STATE_CHARS) { + return { ok: false, error: "invalid_state" }; + } + if (!redirectUri || !isAllowedMcpRedirectUri(redirectUri)) { + return { ok: false, error: "invalid_redirect_uri" }; + } + return { ok: true, pending: { state, redirectUri } }; +} + +export function encodeMcpOauthPendingCookie(pending: McpOauthPending): string { + const secret = bridgeSecret(); + if (!secret) { + throw new Error("MCP OAuth bridge secret is not configured"); + } + const payload = Buffer.from( + JSON.stringify({ + ...pending, + exp: Date.now() + PENDING_TTL_MS, + }), + "utf8" + ).toString("base64url"); + const sig = createHmac("sha256", secret).update(payload).digest("base64url"); + return `${payload}.${sig}`; +} + +export function decodeMcpOauthPendingCookie( + value: string | undefined +): McpOauthPending | null { + if (!value) return null; + const secret = bridgeSecret(); + if (!secret) return null; + const [payload, sig] = value.split("."); + if (!payload || !sig) return null; + const expected = createHmac("sha256", secret).update(payload).digest("base64url"); + const left = Buffer.from(sig); + const right = Buffer.from(expected); + if (left.length !== right.length || !timingSafeEqual(left, right)) { + return null; + } + try { + const parsed = JSON.parse(Buffer.from(payload, "base64url").toString("utf8")) as { + state?: unknown; + redirectUri?: unknown; + exp?: unknown; + }; + if ( + typeof parsed.state !== "string" || + typeof parsed.redirectUri !== "string" || + typeof parsed.exp !== "number" || + parsed.exp < Date.now() + ) { + return null; + } + if (!isAllowedMcpRedirectUri(parsed.redirectUri)) { + return null; + } + return { state: parsed.state, redirectUri: parsed.redirectUri }; + } catch { + return null; + } +} + +export function buildMcpOauthCallbackUrl(input: { + redirectUri: string; + state: string; + externalUserId: string; + email?: string; +}): string { + const url = new URL(input.redirectUri); + url.searchParams.set("state", input.state); + url.searchParams.set("external_user_id", input.externalUserId); + if (input.email) { + url.searchParams.set("email", input.email); + } + return url.toString(); +} + +export const MCP_OAUTH_COMPLETE_PATH = "/api/v1/auth/mcp/complete"; diff --git a/lib/console/mock-data.ts b/lib/console/mock-data.ts index b9b4cde..96ed29e 100644 --- a/lib/console/mock-data.ts +++ b/lib/console/mock-data.ts @@ -1056,6 +1056,7 @@ Typical end-to-end: 20-40ms per frame on dedicated orchestrators.`, name: "Qwen3 32B", provider: "Qwen", category: "Language", + runnerAppId: "vllm/qwen2.5-0.5b-instruct", coverImage: "/images/console/explore/qwen3-32b.webp", description: "High-performance 32B parameter language model with strong reasoning and multilingual capabilities.", diff --git a/lib/console/model-api-url.ts b/lib/console/model-api-url.ts new file mode 100644 index 0000000..8a8d34d --- /dev/null +++ b/lib/console/model-api-url.ts @@ -0,0 +1,31 @@ +import type { App } from "@/lib/console/types"; + +const DEFAULT_GATEWAY_BASE = "https://gateway.livepeer.org/v1"; + +function isHttpUrl(value: string): boolean { + return /^https?:\/\//i.test(value); +} + +/** Gateway base URL for snippets and docs (never a bare capability id). */ +export function getModelApiBaseUrl(model: App): string { + const candidate = model.apiEndpoint?.trim(); + if (candidate && isHttpUrl(candidate)) { + return candidate.replace(/\/$/, ""); + } + return DEFAULT_GATEWAY_BASE; +} + +/** POST target for the model's inference API. */ +export function getModelApiPostUrl(model: App): string { + const base = getModelApiBaseUrl(model); + if (model.category === "Language") { + return `${base}/chat/completions`; + } + const pipeline = encodeURIComponent(model.id); + return `${base}/${pipeline}`; +} + +/** Host header value for raw HTTP examples. */ +export function getModelApiHost(model: App): string { + return new URL(getModelApiBaseUrl(model)).host; +} diff --git a/lib/console/pymthouse-billing-bff.ts b/lib/console/pymthouse-billing-bff.ts index 8036c98..6c34eee 100644 --- a/lib/console/pymthouse-billing-bff.ts +++ b/lib/console/pymthouse-billing-bff.ts @@ -165,17 +165,21 @@ export async function changeDashboardBillingSubscription(input: { return readPymthouseResponse(response); } -export async function getDashboardUserSubscription( - externalUserId: string -): Promise { - const client = createPmtHouseClientForPublicApp(readPublicClientId()); - const result: UserSubscriptionResponse = - await client.getUserSubscription(externalUserId); +type UserSubscriptionWithLivePlan = UserSubscriptionResponse & { + livePlan?: { id?: string | null; name?: string | null } | null; +}; + +export function mapDashboardUserSubscription( + result: UserSubscriptionWithLivePlan +): DashboardUserSubscription { const sub = result.subscription; const pending = result.pendingCancel ?? null; + const livePlanId = result.livePlan?.id?.trim() || null; + const livePlanName = result.livePlan?.name?.trim() || null; return { - planId: sub?.planId?.trim() || pending?.planId?.trim() || null, - planName: sub?.planName?.trim() || pending?.planName?.trim() || null, + planId: sub?.planId?.trim() || livePlanId || pending?.planId?.trim() || null, + planName: + sub?.planName?.trim() || livePlanName || pending?.planName?.trim() || null, status: sub?.status?.trim() || (pending ? "canceled" : null), subscriptionId: sub?.id?.trim() || pending?.subscriptionId?.trim() || null, currentPeriodEnd: @@ -193,6 +197,16 @@ export async function getDashboardUserSubscription( }; } +export async function getDashboardUserSubscription( + externalUserId: string +): Promise { + const client = createPmtHouseClientForPublicApp(readPublicClientId()); + const result = (await client.getUserSubscription( + externalUserId + )) as UserSubscriptionWithLivePlan; + return mapDashboardUserSubscription(result); +} + export async function cancelDashboardUserSubscription( externalUserId: string, opts?: { timing?: string; effectiveAt?: string } diff --git a/lib/console/pymthouse-me-billing-bff.ts b/lib/console/pymthouse-me-billing-bff.ts new file mode 100644 index 0000000..b747f7f --- /dev/null +++ b/lib/console/pymthouse-me-billing-bff.ts @@ -0,0 +1,130 @@ +import "server-only"; + +import { + isMerchantBillingRequiredError, + readAccessTokenBillingMode, + type AppUserInvoice, + type BillingState, + type EndUserMeWallet, + type UserSubscriptionResponse, +} from "@pymthouse/builder-sdk"; + +import { mapDashboardUserSubscription } from "@/lib/console/pymthouse-billing-bff"; +import type { DashboardUserSubscription } from "@/lib/console/pymthouse-billing"; +import { + createPmtHouseClientForPublicApp, + mintEndUserAccessToken, +} from "@/lib/console/pymthouse-bff"; +import { readPublicClientId } from "@/lib/console/pymthouse-http"; +import type { + DashboardOwnerWallet, + DashboardWalletInvoice, + DashboardWalletPaymentMethod, +} from "@/lib/console/pymthouse-wallet"; + +export type MerchantMeBillingBundle = { + mode: "merchant"; + state: BillingState | null; + wallet: DashboardOwnerWallet | null; + subscription: DashboardUserSubscription | null; + paymentMethods: DashboardWalletPaymentMethod[]; + invoices: DashboardWalletInvoice[]; +}; + +export type MeBillingSurface = + | { + mode: "owner_rollup"; + code: "merchant_billing_required"; + } + | MerchantMeBillingBundle; + +type UserSubscriptionWithLivePlan = UserSubscriptionResponse & { + livePlan?: { id?: string | null; name?: string | null } | null; +}; + +function asOwnerWallet(wallet: EndUserMeWallet): DashboardOwnerWallet { + return { + clientId: wallet.clientId, + balance: wallet.balance, + paymentMethod: wallet.paymentMethod, + billingState: wallet.billingState, + payPerUsePlans: wallet.payPerUsePlans, + }; +} + +function mapInvoice(invoice: AppUserInvoice): DashboardWalletInvoice { + return { + id: invoice.id, + number: invoice.number, + status: invoice.status, + currency: invoice.currency, + totalAmount: invoice.totalAmount, + issuedAt: invoice.issuedAt, + periodStart: invoice.periodStart, + periodEnd: invoice.periodEnd, + invoiceType: invoice.invoiceType, + }; +} + +async function readMerchantPiece( + load: () => Promise +): Promise { + try { + return await load(); + } catch (error) { + if (isMerchantBillingRequiredError(error)) { + return "rollup"; + } + return null; + } +} + +export async function readSessionMeBilling(input: { + externalUserId: string; + email?: string; +}): Promise { + const accessToken = await mintEndUserAccessToken( + input.externalUserId, + input.email + ); + const mintedMode = readAccessTokenBillingMode(accessToken); + if (mintedMode === "owner_rollup") { + return { mode: "owner_rollup", code: "merchant_billing_required" }; + } + + const client = createPmtHouseClientForPublicApp(readPublicClientId()); + + const [stateResult, walletResult, subscriptionResult, pmResult, invoiceResult] = + await Promise.all([ + readMerchantPiece(() => client.getMeBillingState(accessToken)), + readMerchantPiece(() => client.getMeBillingWallet(accessToken)), + readMerchantPiece(() => client.getMeBillingSubscription(accessToken)), + readMerchantPiece(() => client.getMeBillingPaymentMethods(accessToken)), + readMerchantPiece(() => + client.getMeBillingInvoices(accessToken, { pageSize: 20 }) + ), + ]); + + if ( + stateResult === "rollup" || + walletResult === "rollup" || + subscriptionResult === "rollup" || + pmResult === "rollup" || + invoiceResult === "rollup" + ) { + return { mode: "owner_rollup", code: "merchant_billing_required" }; + } + + return { + mode: "merchant", + state: stateResult, + wallet: walletResult ? asOwnerWallet(walletResult) : null, + subscription: subscriptionResult + ? mapDashboardUserSubscription( + subscriptionResult as UserSubscriptionWithLivePlan + ) + : null, + paymentMethods: pmResult?.paymentMethods ?? [], + invoices: (invoiceResult?.items ?? []).map(mapInvoice), + }; +} diff --git a/lib/console/runner-gateway-client.ts b/lib/console/runner-gateway-client.ts new file mode 100644 index 0000000..a4deb81 --- /dev/null +++ b/lib/console/runner-gateway-client.ts @@ -0,0 +1,123 @@ +import type { App } from "@/lib/console/types"; + +export function buildOpenAIChatPayload( + model: App, + values: Record +): Record { + const prompt = + typeof values.prompt === "string" && values.prompt.trim() + ? values.prompt.trim() + : typeof values.messages !== "undefined" + ? values.messages + : "Hello, how are you?"; + + const messages = Array.isArray(prompt) + ? prompt + : [{ role: "user", content: String(prompt) }]; + + const payload: Record = { + model: model.id, + messages, + }; + + if (values.temperature !== undefined && values.temperature !== "") { + payload.temperature = Number(values.temperature); + } + if (values.max_tokens !== undefined && values.max_tokens !== "") { + payload.max_tokens = Number(values.max_tokens); + } + if (values.stream === true) { + payload.stream = true; + } + + return payload; +} + +/** Build the JSON body for a live-runner playground call. */ +export function buildLiveRunnerPayload( + model: App, + values: Record +): Record { + const runnerPath = model.playgroundConfig?.runnerPath?.trim(); + if (!runnerPath) { + return buildOpenAIChatPayload(model, values); + } + + // Hello-world and other form-backed runners: send field values as JSON. + const payload: Record = {}; + for (const [key, value] of Object.entries(values)) { + if (value === undefined || value === "") continue; + payload[key] = value; + } + if (runnerPath === "hello" && typeof payload.name !== "string") { + payload.name = "world"; + } + return payload; +} + +export function extractRunnerResultText(data: unknown): string { + if (typeof data === "string") return data; + if (!data || typeof data !== "object" || Array.isArray(data)) { + return JSON.stringify(data, null, 2); + } + const record = data as Record; + if (typeof record.message === "string") return record.message; + return extractAssistantText(data); +} + +export function extractAssistantText(data: unknown): string { + if (!data || typeof data !== "object" || Array.isArray(data)) { + return typeof data === "string" ? data : JSON.stringify(data, null, 2); + } + const record = data as Record; + const choices = record.choices; + if (!Array.isArray(choices) || choices.length === 0) { + return JSON.stringify(data, null, 2); + } + const first = choices[0]; + if (!first || typeof first !== "object") { + return JSON.stringify(data, null, 2); + } + const message = (first as Record).message; + if (message && typeof message === "object" && !Array.isArray(message)) { + const content = (message as Record).content; + if (typeof content === "string") return content; + } + const text = (first as Record).text; + if (typeof text === "string") return text; + return JSON.stringify(data, null, 2); +} + +export function parseSseAssistantText(chunk: string, prior: string): string { + const lines = chunk.split("\n"); + let output = prior; + for (const line of lines) { + if (!line.startsWith("data:")) continue; + const data = line.slice(5).trim(); + if (!data || data === "[DONE]") continue; + try { + const parsed = JSON.parse(data) as Record; + const choices = parsed.choices; + if (!Array.isArray(choices) || choices.length === 0) continue; + const delta = (choices[0] as Record).delta; + if (delta && typeof delta === "object" && !Array.isArray(delta)) { + const content = (delta as Record).content; + if (typeof content === "string") output += content; + } + } catch { + // ignore malformed SSE lines + } + } + return output; +} + +export function runnerGatewayPostUrl( + gatewayBaseUrl: string, + runnerAppId: string, + path = "chat/completions" +): string { + const base = gatewayBaseUrl.replace(/\/+$/, ""); + const tail = path.replace(/^\/+/, ""); + const params = new URLSearchParams({ app: runnerAppId }); + return `${base}/${tail}?${params.toString()}`; +} diff --git a/lib/console/signer-session-bff.ts b/lib/console/signer-session-bff.ts new file mode 100644 index 0000000..d16c1fc --- /dev/null +++ b/lib/console/signer-session-bff.ts @@ -0,0 +1,141 @@ +import "server-only"; + +import { + createSignerTokenManager, + mintUserSignerToken, + type CachedSignerToken, +} from "@pymthouse/builder-sdk/signer/server"; +import { PmtHouseError } from "@pymthouse/builder-sdk"; +import { createPmtHouseClientForPublicApp } from "@/lib/console/pymthouse-bff"; +import { + readPublicClientId, + readPymthouseM2mConfig, + requirePymthouseM2mConfig, +} from "@/lib/console/pymthouse-http"; + +export type SignerContext = { + jwt: string; + signerUrl: string | undefined; + balanceUsdMicros: string; + lifetimeGrantedUsdMicros: string; + expiresAt: number; +}; + +const SIGNER_ROUTING_TTL_MS = 5 * 60 * 1000; + +type CachedSignerRouting = { + signerUrl: string; + fetchedAt: number; +}; + +const signerRoutingByClient = new Map(); + +function isPymthouseConfigured(): boolean { + if (readPymthouseM2mConfig() === null) return false; + try { + readPublicClientId(); + return true; + } catch { + return false; + } +} + +/** + * Resolve the public signer DMZ URL from issuer app routing + * (`GET …/apps/{clientId}/signer/routing`) — not from dashboard env. + */ +async function resolveSignerUrl( + publicClientId: string +): Promise { + const now = Date.now(); + const cached = signerRoutingByClient.get(publicClientId); + if (cached && now - cached.fetchedAt < SIGNER_ROUTING_TTL_MS) { + return cached.signerUrl; + } + + const client = createPmtHouseClientForPublicApp(publicClientId); + const routing = await client.getSignerRouting(); + const signerUrl = + routing.routing?.signerApiUrl?.trim() || + routing.patterns?.directDmz?.signerApiUrl?.trim() || + ""; + if (!signerUrl) { + return undefined; + } + signerRoutingByClient.set(publicClientId, { signerUrl, fetchedAt: now }); + return signerUrl; +} + +const tokenManager = createSignerTokenManager({ + mint: async (publicClientId, externalUserId) => { + const config = requirePymthouseM2mConfig(); + createPmtHouseClientForPublicApp(publicClientId); + return mintUserSignerToken({ + issuerUrl: config.issuerUrl, + m2mClientId: config.m2mClientId, + m2mClientSecret: config.m2mClientSecret, + externalUserId, + allowInsecureHttp: config.allowInsecureHttp, + }); + }, +}); + +function toSignerContext( + token: CachedSignerToken, + signerUrl: string | undefined +): SignerContext { + return { + jwt: token.jwt, + signerUrl, + balanceUsdMicros: token.balanceUsdMicros, + lifetimeGrantedUsdMicros: token.lifetimeGrantedUsdMicros, + expiresAt: token.expiresAt, + }; +} + +export function isRunnerSignerConfigured(): boolean { + return isPymthouseConfigured(); +} + +export async function getSignerContext( + externalUserId: string, + options?: { forceRefresh?: boolean } +): Promise { + const trimmed = externalUserId.trim(); + if (!trimmed) { + throw new PmtHouseError("externalUserId is required", { + status: 400, + code: "invalid_external_user_id", + }); + } + const publicClientId = readPublicClientId(); + const [token, signerUrl] = await Promise.all([ + tokenManager.getToken(publicClientId, trimmed, { + forceRefresh: options?.forceRefresh, + }), + resolveSignerUrl(publicClientId), + ]); + return toSignerContext(token, signerUrl); +} + +export async function getSignerSessionStatus(externalUserId: string): Promise<{ + ready: boolean; + expiresIn: number; + balanceUsdMicros: string; + lifetimeGrantedUsdMicros: string; + /** Short-lived signer JWT — embedded hidden in the playground for live runs. */ + jwt: string; +}> { + const context = await getSignerContext(externalUserId); + const expiresIn = Math.max( + 1, + Math.floor((context.expiresAt - Date.now()) / 1000) + ); + return { + ready: true, + expiresIn, + balanceUsdMicros: context.balanceUsdMicros, + lifetimeGrantedUsdMicros: context.lifetimeGrantedUsdMicros, + jwt: context.jwt, + }; +} diff --git a/lib/console/streaming-playground.ts b/lib/console/streaming-playground.ts new file mode 100644 index 0000000..05e71a0 --- /dev/null +++ b/lib/console/streaming-playground.ts @@ -0,0 +1,108 @@ +import type { App, PlaygroundConfig } from "@/lib/console/types"; + +/** Discovery capability ids that get the LV2V webcam / gateway playground. */ +export function isLv2vPlaygroundCapability(capability: string): boolean { + const id = capability.toLowerCase(); + return ( + id.includes("streamdiffusion") || + id === "live-video-to-video" || + id.startsWith("live-video") + ); +} + +/** + * Resolve the orchestrator pipeline model name for a capability. + * Discovery page id may differ from the orchestrator pipeline model name. + */ +export function resolveGatewayModelId(capability: string): string { + const id = capability.trim(); + const lower = id.toLowerCase(); + if (lower === "streamdiffusion") { + return "streamdiffusion"; + } + if (lower === "live-video-to-video") { + return "streamdiffusion-sdxl"; + } + return id; +} + +export function buildLv2vPlaygroundConfig(_capability: string): PlaygroundConfig { + return { + fields: [ + { + name: "prompt", + label: "Prompt", + type: "textarea", + placeholder: "Describe the look or style for the stream…", + description: "Optional pipeline prompt (passed when starting the LV2V job).", + }, + { + name: "style", + label: "Style preset", + type: "select", + options: ["none", "cinematic", "anime", "watercolor", "neon", "sketch"], + defaultValue: "none", + description: "Local preview label only until full pipeline params are wired.", + }, + { + name: "strength", + label: "Strength", + type: "range", + min: 0, + max: 1, + step: 0.05, + defaultValue: 0.6, + }, + ], + outputType: "video", + playgroundVariant: "webcam", + mockOutputUrl: "https://picsum.photos/seed/streamdiffusion/640/360", + }; +} + +/** Live-runner demo apps with a simple request/response playground. */ +export function isHelloWorldCapability(capability: string): boolean { + const id = capability.toLowerCase(); + return id === "livepeer-example/hello-world" || id.endsWith("/hello-world"); +} + +export function buildHelloWorldPlaygroundConfig(): PlaygroundConfig { + return { + fields: [ + { + name: "name", + label: "Name", + type: "text", + required: true, + defaultValue: "livepeer", + placeholder: "Who should we greet?", + description: "Passed as JSON { name } to POST /hello on the runner.", + }, + ], + outputType: "text", + mockOutputText: "Hello, livepeer!", + runnerPath: "hello", + }; +} + +export function enrichDiscoveryModelForStreaming(model: App): App { + if (isHelloWorldCapability(model.id)) { + return { + ...model, + playgroundConfig: model.playgroundConfig ?? buildHelloWorldPlaygroundConfig(), + }; + } + + if (!isLv2vPlaygroundCapability(model.id)) { + return model; + } + + return { + ...model, + realtime: true, + category: + model.category === "Language" ? "Video Generation" : model.category, + gatewayModelId: resolveGatewayModelId(model.id), + playgroundConfig: model.playgroundConfig ?? buildLv2vPlaygroundConfig(model.id), + }; +} diff --git a/lib/console/types.ts b/lib/console/types.ts index 4d9694c..05dea7b 100644 --- a/lib/console/types.ts +++ b/lib/console/types.ts @@ -147,6 +147,10 @@ export interface PlaygroundConfig { mockOutputJson?: unknown; /** Selects the playground UI. "webcam" mocks live video-in/video-out with the user's camera. "transcoding" shapes the output like a Livepeer HLS stream (playbackId, rendition ladder, copyable URLs). Defaults to "form". */ playgroundVariant?: "form" | "webcam" | "transcoding"; + /** Live-runner HTTP path under the reserved session app URL (e.g. "hello"). + * When set, playground posts form values as JSON to this path instead of + * OpenAI-style chat/completions. */ + runnerPath?: string; } export interface UsageDataPoint { @@ -184,6 +188,10 @@ export interface App { featured?: boolean; /** Supports streaming (WebRTC) inference in addition to request/response. The differentiator on the network — flagged as a capability pill and filterable on Explore. */ realtime?: boolean; + /** LV2V model_id for gateway sessions when different from discovery capability `id`. */ + gatewayModelId?: string; + /** Live-runner app id for gateway.py-style HTTP apps, e.g. vllm/qwen2.5-0.5b-instruct */ + runnerAppId?: string; /** ISO-8601 date the model was published on the network. Drives the "NEW" badge and Recently-added sort. */ releasedAt?: string; tags?: string[]; diff --git a/lib/console/useDiscoveryModel.ts b/lib/console/useDiscoveryModel.ts new file mode 100644 index 0000000..e4bbb32 --- /dev/null +++ b/lib/console/useDiscoveryModel.ts @@ -0,0 +1,69 @@ +"use client"; + +import { useEffect, useState } from "react"; +import type { App } from "@/lib/console/types"; +import { DEFAULT_DISCOVERY_SERVICE_TYPE } from "@/lib/discovery/constants"; + +type ModelState = + | { status: "loading" } + | { status: "ready"; model: App } + | { status: "not_found" } + | { status: "error"; message: string }; + +export function useDiscoveryModel(capabilityId: string | undefined): ModelState { + const [state, setState] = useState({ status: "loading" }); + + useEffect(() => { + if (!capabilityId) { + setState({ status: "not_found" }); + return; + } + + let cancelled = false; + setState({ status: "loading" }); + + const params = new URLSearchParams({ serviceType: DEFAULT_DISCOVERY_SERVICE_TYPE }); + // Keep `/` as path separators so catch-all `[...id]` can rejoin slash-y + // capability ids (e.g. livepeer-example/hello-world). + const encodedId = capabilityId + .split("/") + .map((segment) => encodeURIComponent(segment)) + .join("/"); + const path = `/api/discovery/models/${encodedId}?${params}`; + + void (async () => { + try { + const response = await fetch(path); + const body = (await response.json()) as { model?: App; error?: string }; + + if (cancelled) return; + + if (response.status === 404) { + setState({ status: "not_found" }); + return; + } + if (!response.ok || !body.model) { + setState({ + status: "error", + message: body.error ?? `Failed to load capability (${response.status})`, + }); + return; + } + + setState({ status: "ready", model: body.model }); + } catch (error) { + if (cancelled) return; + setState({ + status: "error", + message: error instanceof Error ? error.message : "Failed to load capability", + }); + } + })(); + + return () => { + cancelled = true; + }; + }, [capabilityId]); + + return state; +} diff --git a/lib/console/useExploreModels.ts b/lib/console/useExploreModels.ts new file mode 100644 index 0000000..46e072c --- /dev/null +++ b/lib/console/useExploreModels.ts @@ -0,0 +1,86 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import type { ExploreApiResponse } from "@/lib/discovery/types"; +import type { App } from "@/lib/console/types"; +import { + DEFAULT_DISCOVERY_SERVICE_TYPE, + type DiscoveryServiceType, +} from "@/lib/discovery/constants"; + +export type { DiscoveryServiceType } from "@/lib/discovery/constants"; + +type ExploreState = + | { status: "loading"; models: App[] } + | { status: "ready"; models: App[]; capabilityCount: number; serviceType: string } + | { status: "error"; models: App[]; error: string }; + +let exploreCache: { + key: string; + payload: ExploreApiResponse; + fetchedAt: number; +} | null = null; + +const CACHE_TTL_MS = 60_000; + +export function useExploreModels( + serviceType: DiscoveryServiceType = DEFAULT_DISCOVERY_SERVICE_TYPE, +): ExploreState & { reload: () => void } { + const [state, setState] = useState({ status: "loading", models: [] }); + const cacheKey = serviceType; + + const load = useCallback(async () => { + const cached = + exploreCache && + exploreCache.key === cacheKey && + Date.now() - exploreCache.fetchedAt < CACHE_TTL_MS + ? exploreCache.payload + : null; + + if (cached) { + setState({ + status: "ready", + models: cached.models, + capabilityCount: cached.capabilityCount, + serviceType: cached.serviceType, + }); + return; + } + + setState((prev) => ({ ...prev, status: "loading" })); + + try { + const params = new URLSearchParams({ serviceType }); + const response = await fetch(`/api/discovery/explore?${params}`); + const body = (await response.json()) as ExploreApiResponse & { error?: string }; + + if (!response.ok) { + throw new Error(body.error ?? `Explore fetch failed (${response.status})`); + } + + exploreCache = { key: cacheKey, payload: body, fetchedAt: Date.now() }; + setState({ + status: "ready", + models: body.models, + capabilityCount: body.capabilityCount, + serviceType: body.serviceType, + }); + } catch (error) { + const message = error instanceof Error ? error.message : "Failed to load capabilities"; + setState({ status: "error", models: [], error: message }); + } + }, [cacheKey, serviceType]); + + useEffect(() => { + void load(); + }, [load]); + + const reload = useCallback(() => { + if (exploreCache?.key === cacheKey) { + exploreCache = null; + } + void load(); + }, [cacheKey, load]); + + return { ...state, reload }; +} diff --git a/lib/console/useMeBillingSurface.ts b/lib/console/useMeBillingSurface.ts new file mode 100644 index 0000000..61c5a91 --- /dev/null +++ b/lib/console/useMeBillingSurface.ts @@ -0,0 +1,49 @@ +"use client"; + +import { useCallback, useEffect, useState } from "react"; +import type { MeBillingSurface } from "@/lib/console/pymthouse-me-billing-bff"; +import { readResponseJson } from "@/lib/console/read-response-json"; + +type MeBillingState = + | { status: "idle" } + | { status: "loading" } + | { status: "ready"; surface: MeBillingSurface } + | { status: "error"; message: string }; + +/** End-user `/me/billing` surface from the minted JWT's `billing_mode`. */ +export function useMeBillingSurface(enabled: boolean) { + const [state, setState] = useState({ status: "idle" }); + + const load = useCallback(async () => { + if (!enabled) { + setState({ status: "idle" }); + return; + } + + setState({ status: "loading" }); + try { + const response = await fetch("/api/pymthouse/me-billing", { + cache: "no-store", + }); + const body = await readResponseJson( + response + ); + if (!response.ok) { + throw new Error(body.error ?? `Me billing failed (${response.status})`); + } + setState({ status: "ready", surface: body }); + } catch (error) { + setState({ + status: "error", + message: + error instanceof Error ? error.message : "Failed to load me billing", + }); + } + }, [enabled]); + + useEffect(() => { + void load(); + }, [load]); + + return { state, reload: load }; +} diff --git a/lib/console/useOwnerWallet.ts b/lib/console/useOwnerWallet.ts index bf5ffac..7a03ece 100644 --- a/lib/console/useOwnerWallet.ts +++ b/lib/console/useOwnerWallet.ts @@ -37,6 +37,71 @@ type WalletBillingState = | { status: "ready"; wallet: DashboardOwnerWallet } | { status: "error"; message: string }; +/** Checkout actions against `/api/pymthouse/wallet*` (session `externalUserId`). */ +export function useWalletCheckoutActions() { + const startTopUp = useCallback(async (input: { amountUsd: string }) => { + const response = await fetch("/api/pymthouse/wallet/top-up", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + amountUsd: input.amountUsd, + successUrl: `${window.location.origin}/usage?topup=succeeded`, + cancelUrl: `${window.location.origin}/usage?topup=canceled`, + }), + }); + const body = await readResponseJson<{ + checkoutUrl?: string; + error?: string; + }>(response); + if (!response.ok || !body.checkoutUrl) { + throw new Error(body.error ?? `Top-up failed (${response.status})`); + } + return { checkoutUrl: body.checkoutUrl }; + }, []); + + const startPaymentMethodCheckout = useCallback(async () => { + const response = await fetch("/api/pymthouse/wallet/payment-methods", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + successUrl: `${window.location.origin}/usage?topup=pm-saved`, + cancelUrl: `${window.location.origin}/usage?topup=canceled`, + }), + }); + const body = await readResponseJson<{ + checkoutUrl?: string; + error?: string; + }>(response); + if (!response.ok || !body.checkoutUrl) { + throw new Error( + body.error ?? `Payment method checkout failed (${response.status})` + ); + } + return { checkoutUrl: body.checkoutUrl }; + }, []); + + const ensureDefaultPaymentMethod = useCallback(async () => { + const response = await fetch("/api/pymthouse/wallet/payment-methods", { + method: "PATCH", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ ensureDefault: true }), + }); + const body = await readResponseJson<{ error?: string }>(response); + if (!response.ok) { + throw new Error( + body.error ?? + `Ensure default payment method failed (${response.status})` + ); + } + }, []); + + return { + startTopUp, + startPaymentMethodCheckout, + ensureDefaultPaymentMethod, + }; +} + /** Wallet GET only — remaining included usage + plan, without PM/invoice lists. */ export function useWalletBillingState(enabled: boolean) { const [state, setState] = useState({ status: "idle" }); @@ -138,68 +203,18 @@ export function useOwnerWallet(enabled: boolean) { void load(); }, [load]); - const startTopUp = useCallback(async (input: { amountUsd: string }) => { - const response = await fetch("/api/pymthouse/wallet/top-up", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - amountUsd: input.amountUsd, - successUrl: `${window.location.origin}/usage?topup=succeeded`, - cancelUrl: `${window.location.origin}/usage?topup=canceled`, - }), - }); - const body = await readResponseJson<{ - checkoutUrl?: string; - error?: string; - }>(response); - if (!response.ok || !body.checkoutUrl) { - throw new Error(body.error ?? `Top-up failed (${response.status})`); - } - return { checkoutUrl: body.checkoutUrl }; - }, []); - - const startPaymentMethodCheckout = useCallback(async () => { - const response = await fetch("/api/pymthouse/wallet/payment-methods", { - method: "POST", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ - successUrl: `${window.location.origin}/usage?topup=pm-saved`, - cancelUrl: `${window.location.origin}/usage?topup=canceled`, - }), - }); - const body = await readResponseJson<{ - checkoutUrl?: string; - error?: string; - }>(response); - if (!response.ok || !body.checkoutUrl) { - throw new Error( - body.error ?? `Payment method checkout failed (${response.status})` - ); - } - return { checkoutUrl: body.checkoutUrl }; - }, []); + const checkout = useWalletCheckoutActions(); const ensureDefaultPaymentMethod = useCallback(async () => { - const response = await fetch("/api/pymthouse/wallet/payment-methods", { - method: "PATCH", - headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ ensureDefault: true }), - }); - const body = await readResponseJson<{ error?: string }>(response); - if (!response.ok) { - throw new Error( - body.error ?? - `Ensure default payment method failed (${response.status})` - ); - } + await checkout.ensureDefaultPaymentMethod(); await load(); - }, [load]); + }, [checkout, load]); return { state, reload: load, - startTopUp, - startPaymentMethodCheckout, + startTopUp: checkout.startTopUp, + startPaymentMethodCheckout: checkout.startPaymentMethodCheckout, ensureDefaultPaymentMethod, }; } diff --git a/lib/console/wallet-settlement-display.test.ts b/lib/console/wallet-settlement-display.test.ts index 7e588b0..22815d0 100644 --- a/lib/console/wallet-settlement-display.test.ts +++ b/lib/console/wallet-settlement-display.test.ts @@ -9,6 +9,7 @@ import { includedUsageRemainingLabel, includedUsageSummary, overageLimitNote, + sharedPoolUsageMeter, spendPostureBadge, } from "./wallet-settlement-display"; @@ -29,6 +30,7 @@ function makeState(overrides: { remaining?: { usdMicros: string; usd: string } | null; utilizationBps?: number | null; leadThreshold?: { usdMicros: string; usd: string }; + subjectType?: BillingState["subject"]["type"]; }): BillingStateWithIncluded { const prepaid = { ...money("0", "0.00"), @@ -48,7 +50,7 @@ function makeState(overrides: { return { asOf: "2026-08-08T00:00:00.000Z", subject: { - type: "owner", + type: overrides.subjectType ?? "owner", externalUserId: null, billingMode: "owner_rollup", }, @@ -243,6 +245,7 @@ describe("includedUsageSummary", () => { includedTotal: { usdMicros: "5000000", usd: "5.00" }, includedConsumed: { usdMicros: "18000", usd: "0.02" }, sourcePlan: { id: "plan_1", name: "Starter", type: "free" }, + subjectType: "end_user", }), ); assert.ok(summary); @@ -255,6 +258,49 @@ describe("includedUsageSummary", () => { "Starter · $4.98 of $5.00 included left", ); }); + + it("flags an owner_rollup pool as shared so actor counts are not its source", () => { + const owner = includedUsageSummary( + makeState({ + includedRemaining: { usdMicros: "2430000", usd: "2.43" }, + includedTotal: { usdMicros: "5000000", usd: "5.00" }, + includedConsumed: { usdMicros: "2570000", usd: "2.57" }, + subjectType: "owner", + }), + ); + assert.equal(owner?.sharedWithApp, true); + assert.equal( + includedUsageRemainingLabel(owner!), + "Plan · $2.43 available", + ); + + const endUser = includedUsageSummary( + makeState({ + includedRemaining: { usdMicros: "2430000", usd: "2.43" }, + includedTotal: { usdMicros: "5000000", usd: "5.00" }, + subjectType: "end_user", + }), + ); + assert.equal(endUser?.sharedWithApp, false); + }); + + it("meters this user's spend against remaining pool, not the $5 grant", () => { + const meter = sharedPoolUsageMeter({ + state: makeState({ + includedRemaining: { usdMicros: "2430000", usd: "2.43" }, + includedTotal: { usdMicros: "5000000", usd: "5.00" }, + includedConsumed: { usdMicros: "2570000", usd: "2.57" }, + prepaid: { usdMicros: "140000", usd: "0.14" }, + spendable: { usdMicros: "2570000", usd: "2.57" }, + subjectType: "owner", + }), + actorUsdMicros: "3000", + }); + assert.equal(meter.actorUsd, "0.003"); + assert.equal(meter.availableUsd, "2.57"); + assert.equal(meter.label, "$0.003 of $2.57 available"); + assert.ok(!meter.label.includes("5.00")); + }); }); describe("overageLimitNote", () => { diff --git a/lib/console/wallet-settlement-display.ts b/lib/console/wallet-settlement-display.ts index db95d08..497b499 100644 --- a/lib/console/wallet-settlement-display.ts +++ b/lib/console/wallet-settlement-display.ts @@ -30,6 +30,13 @@ export function formatWalletUsd(micros: string | null | undefined): string { return microsToUsd(micros).toFixed(2); } +function formatMeterActorUsd(micros: string): string { + const usd = microsToUsd(micros); + if (usd >= 0.01) return usd.toFixed(2); + const trimmed = usd.toFixed(4).replace(/0+$/, "").replace(/\.$/, ""); + return trimmed || "0"; +} + function parseUsdMicros(raw: string | null | undefined): bigint { const trimmed = raw?.trim(); if (!trimmed || !/^-?\d+$/.test(trimmed)) return BigInt(0); @@ -156,6 +163,12 @@ export type IncludedUsageSummary = { planId: string | null; planName: string | null; resetsAt: string | null; + /** + * True when the allowance is the app owner's rollup pool, which every end + * user of the app draws from. Actor-scoped counts (request history, jobs by + * capability) must not be presented as the source of `consumedUsdMicros`. + */ + sharedWithApp: boolean; }; /** @@ -188,6 +201,7 @@ export function includedUsageSummary( planId, planName, resetsAt, + sharedWithApp: state.subject.type === "owner", }; } @@ -195,9 +209,46 @@ export function includedUsageRemainingLabel( summary: IncludedUsageSummary, ): string { const plan = summary.planName ?? "Plan"; + if (summary.sharedWithApp) { + return `${plan} · $${summary.remainingUsd} available`; + } return `${plan} · $${summary.remainingUsd} of $${summary.totalUsd} included left`; } +export type SharedPoolUsageMeter = { + actorUsdMicros: string; + actorUsd: string; + availableUsdMicros: string; + availableUsd: string; + /** `$0.003 of $2.57 available` — actor spend vs remaining pool, not grant total. */ + label: string; +}; + +/** + * Owner-rollup meter: this user's period spend against remaining spendable + * on the shared owner pool. Never uses grant total (the `$5.00`) or pool + * consumed (other users' usage). + */ +export function sharedPoolUsageMeter(input: { + state: BillingState; + actorUsdMicros: string; +}): SharedPoolUsageMeter { + const runway = availableRunway(input.state); + const availableMicros = + parseUsdMicros(runway.usdMicros) > BigInt(0) + ? parseUsdMicros(runway.usdMicros) + : BigInt(0); + const actorUsd = formatMeterActorUsd(input.actorUsdMicros || "0"); + const availableUsd = formatWalletUsd(availableMicros.toString()); + return { + actorUsdMicros: input.actorUsdMicros || "0", + actorUsd, + availableUsdMicros: availableMicros.toString(), + availableUsd, + label: `$${actorUsd} of $${availableUsd} available`, + }; +} + /** When the next invoice goes out, in the customer's terms. */ export function collectionSchedule(state: BillingState): string { const lead = state.collection.leadThreshold; diff --git a/lib/discovery/client.ts b/lib/discovery/client.ts new file mode 100644 index 0000000..96871a6 --- /dev/null +++ b/lib/discovery/client.ts @@ -0,0 +1,102 @@ +import { readDiscoveryServiceUrl } from "./config"; +import { DEFAULT_DISCOVERY_SERVICE_TYPE, type DiscoveryServiceType } from "./constants"; +import { mapCapabilityToModel } from "./map-to-model"; +import type { + DiscoveryCapabilitiesResponse, + DiscoveryFreshnessResponse, + DiscoveryQueryResponse, + ExploreApiResponse, +} from "./types"; + +export { DEFAULT_DISCOVERY_SERVICE_TYPE, type DiscoveryServiceType } from "./constants"; + +async function discoveryFetch(path: string, init?: RequestInit): Promise { + const baseUrl = readDiscoveryServiceUrl(); + const response = await fetch(`${baseUrl}${path}`, { + ...init, + headers: { + Accept: "application/json", + ...(init?.headers ?? {}), + }, + next: { revalidate: 60 }, + }); + + if (!response.ok) { + const body = await response.text(); + throw new Error(`Discovery Service ${response.status}: ${body || response.statusText}`); + } + + return response.json() as Promise; +} + +export async function fetchDiscoveryCapabilities( + serviceType: DiscoveryServiceType = DEFAULT_DISCOVERY_SERVICE_TYPE, +): Promise { + const params = new URLSearchParams({ serviceType }); + return discoveryFetch( + `/v1/discovery/capabilities?${params}`, + ); +} + +export async function fetchDiscoveryFreshness(): Promise { + return discoveryFetch("/v1/discovery/freshness"); +} + +export async function queryDiscoveryCapabilities( + capabilities: string[], + serviceType: DiscoveryServiceType = DEFAULT_DISCOVERY_SERVICE_TYPE, +): Promise { + if (capabilities.length === 0) { + return { results: {} }; + } + + return discoveryFetch("/v1/discovery/query", { + method: "POST", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ + capabilities, + serviceTypes: [serviceType], + topN: 50, + sortBy: "avail", + }), + }); +} + +export async function fetchExploreModels( + serviceType: DiscoveryServiceType = DEFAULT_DISCOVERY_SERVICE_TYPE, +): Promise { + const [capabilitiesResponse, freshness] = await Promise.all([ + fetchDiscoveryCapabilities(serviceType), + fetchDiscoveryFreshness().catch(() => undefined), + ]); + + const entries = capabilitiesResponse.entries ?? []; + const capabilityNames = + capabilitiesResponse.capabilities.length > 0 + ? capabilitiesResponse.capabilities + : entries.map((entry) => entry.capability); + + const entryByCapability = new Map(entries.map((entry) => [entry.capability, entry])); + + const queryResponse = await queryDiscoveryCapabilities(capabilityNames, serviceType); + + const models = capabilityNames.map((capability) => + mapCapabilityToModel( + capability, + entryByCapability.get(capability), + queryResponse.results[capability] ?? [], + ), + ); + + models.sort((a, b) => { + if (a.status !== b.status) return a.status === "hot" ? -1 : 1; + return b.orchestrators - a.orchestrators; + }); + + return { + models, + capabilityCount: capabilityNames.length, + serviceType, + freshness, + }; +} diff --git a/lib/discovery/config.ts b/lib/discovery/config.ts new file mode 100644 index 0000000..563a3c4 --- /dev/null +++ b/lib/discovery/config.ts @@ -0,0 +1,43 @@ +/** + * Livepeer discovery-service URL. + * + * Configure the full raw endpoint, e.g. + * `https://discovery-service-production-8955.up.railway.app/v1/discovery/raw` + * Tokens embed that value as-is. Explore uses the URL origin for sibling + * `/v1/discovery/…` routes. + */ + +const ENV_KEYS = [ + "DISCOVERY_URL", + "DISCOVERY_SERVICE_URL", + "LIVEPEER_DISCOVERY_SERVICE_URL", +] as const; + +function readConfiguredDiscoveryUrl( + env: NodeJS.ProcessEnv = process.env, +): string | undefined { + for (const key of ENV_KEYS) { + const value = env[key]?.trim(); + if (value) return value; + } + return undefined; +} + +/** Full raw endpoint for python-gateway `--token` (as configured). */ +export function readDiscoveryRawUrl(): string | undefined { + return readConfiguredDiscoveryUrl(); +} + +/** + * Origin for Explore catalog fetches (`/v1/discovery/capabilities`, etc.). + * Env must be an absolute URL to the raw discovery endpoint. + */ +export function readDiscoveryServiceUrl(): string { + const configured = readConfiguredDiscoveryUrl(); + if (!configured) { + throw new Error( + "DISCOVERY_SERVICE_URL (or DISCOVERY_URL) is not configured", + ); + } + return new URL(configured).origin; +} diff --git a/lib/discovery/constants.ts b/lib/discovery/constants.ts new file mode 100644 index 0000000..b0abb0e --- /dev/null +++ b/lib/discovery/constants.ts @@ -0,0 +1,3 @@ +export const DEFAULT_DISCOVERY_SERVICE_TYPE = "legacy" as const; + +export type DiscoveryServiceType = "legacy" | "registry"; diff --git a/lib/discovery/map-to-model.ts b/lib/discovery/map-to-model.ts new file mode 100644 index 0000000..5362cfb --- /dev/null +++ b/lib/discovery/map-to-model.ts @@ -0,0 +1,122 @@ +import type { App, AppCategory, AppStatus, PricingUnit } from "@/lib/console/types"; +import { enrichDiscoveryModelForStreaming } from "@/lib/console/streaming-playground"; +import type { DiscoveryCapabilityEntry, DiscoveryDatasetRow } from "./types"; + +function inferCategory(capability: string): AppCategory { + const c = capability.toLowerCase(); + + if (c.startsWith("video:transcode") || c === "video:live.rtmp") { + return "Live Transcoding"; + } + if ( + c.includes("streamdiffusion") || + c.includes("stable-video") || + c.includes("img2vid") || + c.startsWith("video:") + ) { + return "Video Generation"; + } + if ( + c.includes("whisper") || + c.startsWith("openai:audio") || + c.includes("tts") || + c.includes("parler") + ) { + return "Speech"; + } + if ( + c.startsWith("openai:images") || + c.includes("flux") || + c.includes("sdxl") || + c.includes("diffusion") || + c.includes("pix2pix") || + c.includes("upscaler") || + c.includes("realvis") || + c.includes("instruct-pix") + ) { + return "Image Generation"; + } + if (c.includes("sam2") || c.includes("vision")) { + return "Video Understanding"; + } + return "Language"; +} + +function inferPricingUnit(workUnit: string | undefined, capability: string): PricingUnit { + if (workUnit === "tokens") return "M Tokens"; + if (workUnit?.includes("second")) return "Second"; + if (capability.startsWith("video:")) return "Minute"; + return "Request"; +} + +function humanizeCapabilityName(capability: string): string { + const segment = capability.includes(":") + ? capability.split(":").slice(-1)[0]! + : capability; + return segment + .split(/[-_./]+/) + .filter(Boolean) + .map((part) => part.charAt(0).toUpperCase() + part.slice(1)) + .join(" "); +} + +function aggregateRows(rows: DiscoveryDatasetRow[]): { + orchestrators: number; + status: AppStatus; + latency: number; + price: number; + realtime: boolean; +} { + const orchUris = new Set(rows.map((row) => row.orchUri).filter(Boolean)); + const warm = rows.some((row) => row.avail > 0 || row.totalCap > 0); + const latencies = rows + .map((row) => row.avgLatMs ?? row.bestLatMs) + .filter((value): value is number => value != null && value > 0); + const prices = rows.map((row) => row.pricePerUnit).filter((value) => value > 0); + + return { + orchestrators: orchUris.size, + status: warm ? "hot" : "cold", + latency: + latencies.length > 0 + ? latencies.reduce((sum, value) => sum + value, 0) / latencies.length + : 0, + price: prices.length > 0 ? Math.min(...prices) : 0, + realtime: rows.some((row) => row.interactionMode?.includes("stream") ?? false), + }; +} + +export function mapCapabilityToModel( + capability: string, + entry: DiscoveryCapabilityEntry | undefined, + rows: DiscoveryDatasetRow[], +): App { + const stats = aggregateRows(rows); + const sample = rows[0]; + const provider = + entry?.offeringIds?.[0] ?? + (entry?.serviceType === "registry" ? "Registry" : "Livepeer network"); + + const runnerAppId = capability.includes("/") ? capability : undefined; + + return enrichDiscoveryModelForStreaming({ + id: capability, + runnerAppId, + name: humanizeCapabilityName(capability), + provider, + category: inferCategory(capability), + description: `${humanizeCapabilityName(capability)} on the Livepeer open GPU network (${stats.orchestrators} orchestrator${stats.orchestrators === 1 ? "" : "s"}).`, + status: stats.status, + pricing: { + amount: stats.price > 0 ? stats.price : 0.001, + unit: inferPricingUnit(sample?.workUnit, capability), + }, + latency: stats.latency, + orchestrators: stats.orchestrators, + runs7d: Math.max(stats.orchestrators * 8, stats.orchestrators > 0 ? 1 : 0), + uptime: stats.status === "hot" ? 99.2 : 0, + realtime: stats.realtime, + featured: stats.realtime && stats.status === "hot", + tags: entry?.serviceType ? [entry.serviceType] : undefined, + }); +} diff --git a/lib/discovery/types.ts b/lib/discovery/types.ts new file mode 100644 index 0000000..b4f8e13 --- /dev/null +++ b/lib/discovery/types.ts @@ -0,0 +1,54 @@ +/** Discovery Service API shapes (see discovery-service openapi). */ + +export interface DiscoveryCapabilityEntry { + serviceType: string; + capability: string; + offeringIds?: string[]; +} + +export interface DiscoveryCapabilitiesResponse { + capabilities: string[]; + entries?: DiscoveryCapabilityEntry[]; +} + +export interface DiscoveryDatasetRow { + serviceType?: string; + ethAddress?: string; + offeringId?: string; + interactionMode?: string; + workUnit?: string; + pricePerUnitWei?: string; + orchUri: string; + gpuName?: string; + gpuGb?: number; + avail: number; + totalCap: number; + pricePerUnit: number; + bestLatMs?: number | null; + avgLatMs?: number | null; + swapRatio?: number | null; + avgAvail?: number | null; + score?: number; + slaScore?: number | null; +} + +export interface DiscoveryQueryResponse { + results: Record; + datasetVersion?: number; + queryTimeMs?: number; +} + +export interface DiscoveryFreshnessResponse { + populated?: boolean; + refreshedAt?: number; + ageMs?: number; + capabilityCount?: number; + totalRows?: number; +} + +export interface ExploreApiResponse { + models: import("@/lib/console/types").App[]; + capabilityCount: number; + serviceType: string; + freshness?: DiscoveryFreshnessResponse; +} diff --git a/lib/runner-gateway/call-runner.ts b/lib/runner-gateway/call-runner.ts new file mode 100644 index 0000000..87decfa --- /dev/null +++ b/lib/runner-gateway/call-runner.ts @@ -0,0 +1,549 @@ +import { + DIRECT_SIGNER_PATHS, + signerEndpointUrl, +} from "@pymthouse/builder-sdk/signer/server"; +import { + NoRunnerAvailableError, + RunnerGatewayError, + RunnerHttpError, +} from "@/lib/runner-gateway/errors"; +import type { LiveRunnerInstance } from "@/lib/runner-gateway/discovery"; + +export type LiveRunnerSession = { + sessionId: string; + appUrl: string; + runnerUrl: string; + runner: LiveRunnerInstance | null; +}; + +export type LiveRunnerCallResult = { + data: Record; + runnerUrl: string; + runner: LiveRunnerInstance | null; + sessionId: string; +}; + +export type LiveRunnerStreamResult = { + response: Response; + runnerUrl: string; + runner: LiveRunnerInstance | null; + sessionId: string; +}; + +type RunnerPaymentChallenge = { + paymentParams: string; + orchestratorUrl: string; + manifestId: string; +}; + +type SignerAuth = { + signerUrl: string; + jwt: string; +}; + +const PAYER_ADDRESS_HEADER = "Livepeer-Payer-Address"; + +async function getSignerAddress(signer: SignerAuth): Promise { + const url = signerEndpointUrl( + signer.signerUrl, + DIRECT_SIGNER_PATHS.signOrchestratorInfo + ); + const response = await fetch(url, { + method: "POST", + headers: { + Authorization: `Bearer ${signer.jwt}`, + Accept: "application/json", + "Content-Type": "application/json", + }, + body: "{}", + cache: "no-store", + }); + const text = await response.text(); + if (!response.ok) { + throw new RunnerGatewayError( + `Signer info failed: HTTP ${response.status}${text ? ` — ${text.slice(0, 200)}` : ""}`, + { code: "signer_error", status: 502 } + ); + } + let data: Record; + try { + data = JSON.parse(text) as Record; + } catch { + throw new RunnerGatewayError("Signer info response was not valid JSON", { + code: "signer_error", + status: 502, + }); + } + const address = typeof data.address === "string" ? data.address.trim() : ""; + if (!address) { + throw new RunnerGatewayError("Signer info response missing address", { + code: "signer_error", + status: 502, + }); + } + return address; +} + +function parseRunnerPaymentChallenge( + error: RunnerHttpError +): RunnerPaymentChallenge { + let data: unknown; + try { + data = JSON.parse(error.body); + } catch { + throw new RunnerGatewayError( + "Live runner payment challenge response was not valid JSON", + { + code: "payment_challenge_invalid", + status: 502, + } + ); + } + if (!data || typeof data !== "object" || Array.isArray(data)) { + throw new RunnerGatewayError( + "Live runner payment challenge response must be a JSON object", + { + code: "payment_challenge_invalid", + status: 502, + } + ); + } + const record = data as Record; + const paymentParams = + typeof record.payment_params === "string" ? record.payment_params : ""; + const orchestratorUrl = + typeof record.orchestrator === "string" ? record.orchestrator : ""; + const manifestId = + typeof record.manifest_id === "string" ? record.manifest_id : ""; + if (!paymentParams || !orchestratorUrl || !manifestId) { + throw new RunnerGatewayError( + "Live runner payment challenge missing required fields", + { + code: "payment_challenge_invalid", + status: 502, + } + ); + } + return { paymentParams, orchestratorUrl, manifestId }; +} + +async function getRunnerPayment( + challenge: RunnerPaymentChallenge, + signer: SignerAuth, + attribution?: { capability?: string; modelId?: string } +): Promise<{ payment: string; segCreds: string }> { + const url = signerEndpointUrl( + signer.signerUrl, + DIRECT_SIGNER_PATHS.generateLivePayment + ); + const capability = attribution?.capability?.trim() || ""; + const modelId = attribution?.modelId?.trim() || capability; + const response = await fetch(url, { + method: "POST", + headers: { + Authorization: `Bearer ${signer.jwt}`, + Accept: "application/json", + "Content-Type": "application/json", + }, + body: JSON.stringify({ + orchestrator: challenge.paymentParams, + type: "lv2v", + ManifestID: challenge.manifestId, + ...(capability ? { capability, model_id: modelId } : {}), + }), + cache: "no-store", + }); + const text = await response.text(); + if (!response.ok) { + throw new RunnerGatewayError( + `Payment mint failed: HTTP ${response.status}${text ? ` — ${text.slice(0, 200)}` : ""}`, + { code: "signer_error", status: 502 } + ); + } + let data: Record; + try { + data = JSON.parse(text) as Record; + } catch { + throw new RunnerGatewayError("Payment response was not valid JSON", { + code: "signer_error", + status: 502, + }); + } + const payment = typeof data.payment === "string" ? data.payment : ""; + const segCreds = + typeof data.segCreds === "string" + ? data.segCreds + : typeof data.seg_creds === "string" + ? data.seg_creds + : ""; + if (!payment || !segCreds) { + throw new RunnerGatewayError( + "Payment response missing payment or segCreds", + { + code: "signer_error", + status: 502, + } + ); + } + return { payment, segCreds }; +} + +async function readJsonResponse( + response: Response +): Promise> { + const text = await response.text(); + if (!response.ok) { + throw new RunnerHttpError(response.status, text); + } + let data: unknown; + try { + data = JSON.parse(text); + } catch { + throw new RunnerGatewayError("Live runner call expected JSON object", { + code: "runner_error", + status: 502, + }); + } + if (!data || typeof data !== "object" || Array.isArray(data)) { + throw new RunnerGatewayError( + `Live runner call expected JSON object, got ${Array.isArray(data) ? "array" : typeof data}`, + { code: "runner_error", status: 502 } + ); + } + return data as Record; +} + +export async function callRunner(input: { + runnerUrl: string; + runner?: LiveRunnerInstance | null; + payload?: Record; + method?: string; + signer?: SignerAuth | null; + /** Live-runner app / capability id for OpenMeter usage attribution. */ + capability?: string; + timeoutMs?: number; + maxPaymentChallengeRetries?: number; +}): Promise { + const runnerUrl = input.runnerUrl.trim(); + if (!runnerUrl) { + throw new RunnerGatewayError("Live runner call requires runner_url", { + code: "runner_error", + status: 400, + }); + } + + const requestPayload = input.payload ?? {}; + const method = (input.method ?? "POST").toUpperCase(); + const signer = input.signer ?? null; + const paymentAttribution = input.capability?.trim() + ? { capability: input.capability.trim() } + : undefined; + const maxRetries = input.maxPaymentChallengeRetries ?? 3; + const attempts = (Math.max(0, maxRetries) + 1) * 2; + + let payerAddress = ""; + if (signer) { + payerAddress = await getSignerAddress(signer); + } + + let challenge: RunnerPaymentChallenge | null = null; + + for (let attempt = 0; attempt < attempts; attempt += 1) { + const headers: Record = { + Accept: "application/json", + "Content-Type": "application/json", + }; + let sessionId = ""; + + if (signer) { + headers[PAYER_ADDRESS_HEADER] = payerAddress; + } + + if (challenge && signer) { + const payment = await getRunnerPayment( + challenge, + signer, + paymentAttribution + ); + headers["Livepeer-Payment"] = payment.payment; + headers["Livepeer-Segment"] = payment.segCreds; + sessionId = challenge.manifestId; + } + + const controller = new AbortController(); + const timeout = setTimeout( + () => controller.abort(), + input.timeoutMs ?? 120_000 + ); + + try { + const response = await fetch(runnerUrl, { + method, + headers, + body: + method === "GET" || method === "HEAD" + ? undefined + : JSON.stringify(requestPayload), + signal: controller.signal, + cache: "no-store", + }); + + if (response.status === 402) { + if (!signer) { + throw new RunnerGatewayError( + "Live runner paid call requires signer", + { + code: "signer_misconfigured", + status: 503, + } + ); + } + const body = await response.text(); + challenge = parseRunnerPaymentChallenge(new RunnerHttpError(402, body)); + continue; + } + + const data = await readJsonResponse(response); + const dataSessionId = + typeof data.session_id === "string" ? data.session_id.trim() : ""; + return { + data, + runnerUrl, + runner: input.runner ?? null, + sessionId: sessionId || dataSessionId, + }; + } catch (error) { + if ( + error instanceof RunnerHttpError && + error.statusCode === 402 && + signer + ) { + challenge = parseRunnerPaymentChallenge(error); + continue; + } + if (error instanceof RunnerGatewayError) throw error; + if (error instanceof RunnerHttpError) throw error; + if (error instanceof Error && error.name === "AbortError") { + throw new RunnerGatewayError("Live runner call timed out", { + code: "runner_timeout", + status: 504, + }); + } + throw error; + } finally { + clearTimeout(timeout); + } + } + + throw new RunnerGatewayError( + "Live runner call exhausted payment challenge retries", + { + code: "payment_challenge_exhausted", + status: 502, + } + ); +} + +export async function callRunnerStream(input: { + runnerUrl: string; + runner?: LiveRunnerInstance | null; + payload?: Record; + method?: string; + signer?: SignerAuth | null; + /** Live-runner app / capability id for OpenMeter usage attribution. */ + capability?: string; + timeoutMs?: number; + maxPaymentChallengeRetries?: number; +}): Promise { + const runnerUrl = input.runnerUrl.trim(); + if (!runnerUrl) { + throw new RunnerGatewayError("Live runner call requires runner_url", { + code: "runner_error", + status: 400, + }); + } + + const requestPayload = input.payload ?? {}; + const method = (input.method ?? "POST").toUpperCase(); + const signer = input.signer ?? null; + const paymentAttribution = input.capability?.trim() + ? { capability: input.capability.trim() } + : undefined; + const maxRetries = input.maxPaymentChallengeRetries ?? 3; + const attempts = (Math.max(0, maxRetries) + 1) * 2; + + let payerAddress = ""; + if (signer) { + payerAddress = await getSignerAddress(signer); + } + + let challenge: RunnerPaymentChallenge | null = null; + + for (let attempt = 0; attempt < attempts; attempt += 1) { + const headers: Record = { + Accept: "text/event-stream, application/json", + "Content-Type": "application/json", + }; + let sessionId = ""; + + if (signer) { + headers[PAYER_ADDRESS_HEADER] = payerAddress; + } + + if (challenge && signer) { + const payment = await getRunnerPayment( + challenge, + signer, + paymentAttribution + ); + headers["Livepeer-Payment"] = payment.payment; + headers["Livepeer-Segment"] = payment.segCreds; + sessionId = challenge.manifestId; + } + + const controller = new AbortController(); + const timeout = setTimeout( + () => controller.abort(), + input.timeoutMs ?? 300_000 + ); + + try { + const response = await fetch(runnerUrl, { + method, + headers, + body: + method === "GET" || method === "HEAD" + ? undefined + : JSON.stringify(requestPayload), + signal: controller.signal, + cache: "no-store", + }); + + if (response.status === 402) { + if (!signer) { + throw new RunnerGatewayError( + "Live runner paid call requires signer", + { + code: "signer_misconfigured", + status: 503, + } + ); + } + const body = await response.text(); + challenge = parseRunnerPaymentChallenge(new RunnerHttpError(402, body)); + continue; + } + + if (!response.ok) { + const body = await response.text(); + throw new RunnerHttpError(response.status, body); + } + + return { + response, + runnerUrl, + runner: input.runner ?? null, + sessionId, + }; + } catch (error) { + if ( + error instanceof RunnerHttpError && + error.statusCode === 402 && + signer + ) { + challenge = parseRunnerPaymentChallenge(error); + continue; + } + if (error instanceof RunnerGatewayError) throw error; + if (error instanceof RunnerHttpError) throw error; + if (error instanceof Error && error.name === "AbortError") { + throw new RunnerGatewayError("Live runner stream call timed out", { + code: "runner_timeout", + status: 504, + }); + } + throw error; + } finally { + clearTimeout(timeout); + } + } + + throw new RunnerGatewayError( + "Live runner stream call exhausted payment challenge retries", + { + code: "payment_challenge_exhausted", + status: 502, + } + ); +} + +export async function reserveSession(input: { + discoveryUrl: string; + app: string; + signer?: SignerAuth | null; + timeoutMs?: number; +}): Promise { + const { discoverRunnerCandidates } = + await import("@/lib/runner-gateway/discovery"); + const candidates = await discoverRunnerCandidates({ + discoveryUrl: input.discoveryUrl, + app: input.app, + }); + + if (candidates.length === 0) { + throw new NoRunnerAvailableError(); + } + + let lastError: unknown; + for (const runner of candidates) { + try { + const result = await callRunner({ + runnerUrl: runner.url, + runner, + payload: {}, + signer: input.signer, + capability: input.app, + timeoutMs: input.timeoutMs ?? 30_000, + }); + const sessionId = + typeof result.data.session_id === "string" + ? result.data.session_id.trim() + : ""; + const appUrl = + typeof result.data.app_url === "string" + ? result.data.app_url.trim() + : ""; + if (!sessionId) { + throw new RunnerGatewayError( + "runner session response missing session_id", + { + code: "runner_error", + status: 502, + } + ); + } + if (!appUrl) { + throw new RunnerGatewayError( + "runner session response missing app_url", + { + code: "runner_error", + status: 502, + } + ); + } + return { + sessionId, + appUrl, + runnerUrl: result.runnerUrl, + runner, + }; + } catch (error) { + lastError = error; + } + } + + if (lastError instanceof Error) { + throw lastError; + } + throw new NoRunnerAvailableError(); +} diff --git a/lib/runner-gateway/discovery.ts b/lib/runner-gateway/discovery.ts new file mode 100644 index 0000000..d6dae74 --- /dev/null +++ b/lib/runner-gateway/discovery.ts @@ -0,0 +1,127 @@ +import { RunnerGatewayError } from "@/lib/runner-gateway/errors"; + +export type LiveRunnerInstance = { + url: string; + app: string; + runnerId: string; + mode: string; + orchestratorUrl: string; + raw: Record; +}; + +type DiscoveryEntry = Record; + +function normalizeFilterValues(value: string | string[] | undefined): string[] { + if (!value) return []; + const values = Array.isArray(value) ? value : [value]; + return values.map((item) => item.trim()).filter(Boolean); +} + +function appendQueryValues( + url: string, + pairs: Array<[string, string]> +): string { + if (pairs.length === 0) return url; + const parsed = new URL(url); + for (const [key, val] of pairs) { + parsed.searchParams.append(key, val); + } + return parsed.toString(); +} + +function stringValue(value: unknown): string { + return typeof value === "string" ? value.trim() : ""; +} + +function runnerCandidatesFromDiscovery( + entries: DiscoveryEntry[] +): LiveRunnerInstance[] { + const candidates: LiveRunnerInstance[] = []; + for (const entry of entries) { + const orchestratorUrl = stringValue(entry.address); + const runners = entry.runners; + if (!Array.isArray(runners)) continue; + + for (const runner of runners) { + if (!runner || typeof runner !== "object") continue; + const record = runner as Record; + const url = stringValue(record.url); + const app = stringValue(record.app); + if (!url || !app) continue; + candidates.push({ + url, + app, + runnerId: stringValue(record.runner_id), + mode: stringValue(record.mode), + orchestratorUrl, + raw: record, + }); + } + } + return candidates; +} + +export async function discoverRunners(input: { + discoveryUrl: string; + app?: string | string[]; + gpu?: string | string[]; + headers?: Record; +}): Promise { + const appFilters = normalizeFilterValues(input.app); + const gpuFilters = normalizeFilterValues(input.gpu); + + let endpoint = input.discoveryUrl.trim(); + const queryPairs: Array<[string, string]> = []; + for (const item of appFilters) queryPairs.push(["app", item]); + for (const item of gpuFilters) queryPairs.push(["gpu", item]); + endpoint = appendQueryValues(endpoint, queryPairs); + + const response = await fetch(endpoint, { + method: "GET", + headers: { + Accept: "application/json", + ...input.headers, + }, + cache: "no-store", + }); + + const bodyText = await response.text(); + if (!response.ok) { + throw new RunnerGatewayError( + `Discovery failed: HTTP ${response.status}${bodyText ? ` — ${bodyText.slice(0, 200)}` : ""}`, + { code: "discovery_failed", status: 502 } + ); + } + + let data: unknown; + try { + data = JSON.parse(bodyText); + } catch { + throw new RunnerGatewayError("Discovery response was not valid JSON", { + code: "discovery_failed", + status: 502, + }); + } + + if (!Array.isArray(data)) { + throw new RunnerGatewayError( + `Discovery response must be a JSON list, got ${typeof data}`, + { code: "discovery_failed", status: 502 } + ); + } + + return data.filter( + (entry): entry is DiscoveryEntry => + Boolean(entry) && typeof entry === "object" && !Array.isArray(entry) + ); +} + +export async function discoverRunnerCandidates(input: { + discoveryUrl: string; + app?: string | string[]; + gpu?: string | string[]; + headers?: Record; +}): Promise { + const entries = await discoverRunners(input); + return runnerCandidatesFromDiscovery(entries); +} diff --git a/lib/runner-gateway/errors.ts b/lib/runner-gateway/errors.ts new file mode 100644 index 0000000..42902f5 --- /dev/null +++ b/lib/runner-gateway/errors.ts @@ -0,0 +1,36 @@ +export class RunnerGatewayError extends Error { + readonly code: string; + readonly status: number; + + constructor( + message: string, + options: { code?: string; status?: number } = {} + ) { + super(message); + this.name = "RunnerGatewayError"; + this.code = options.code ?? "runner_error"; + this.status = options.status ?? 502; + } +} + +export class NoRunnerAvailableError extends RunnerGatewayError { + constructor(message = "No runners available to select") { + super(message, { code: "no_runners", status: 503 }); + this.name = "NoRunnerAvailableError"; + } +} + +export class RunnerHttpError extends RunnerGatewayError { + readonly statusCode: number; + readonly body: string; + + constructor(statusCode: number, body: string, message?: string) { + super(message ?? `HTTP ${statusCode} from runner`, { + code: "runner_http_error", + status: statusCode >= 500 ? 502 : statusCode, + }); + this.name = "RunnerHttpError"; + this.statusCode = statusCode; + this.body = body; + } +} diff --git a/lib/runner-gateway/forward.ts b/lib/runner-gateway/forward.ts new file mode 100644 index 0000000..69288b2 --- /dev/null +++ b/lib/runner-gateway/forward.ts @@ -0,0 +1,107 @@ +import { + getSignerContext, + isRunnerSignerConfigured, +} from "@/lib/console/signer-session-bff"; +import { + callRunner, + callRunnerStream, + reserveSession, + type LiveRunnerSession, +} from "@/lib/runner-gateway/call-runner"; +import { stopRunnerSession } from "@/lib/runner-gateway/stop-session"; +import { RunnerGatewayError } from "@/lib/runner-gateway/errors"; +import "@/lib/runner-gateway/tls"; + +export type ForwardRunnerRequestInput = { + externalUserId: string; + appId: string; + runnerPath: string; + payload: Record; + discoveryUrl: string; +}; + +function readDiscoveryUrl(): string { + const url = process.env.RUNNER_DISCOVERY_URL?.trim(); + if (!url) { + throw new RunnerGatewayError( + "RUNNER_DISCOVERY_URL is required for live-runner gateway calls", + { code: "runner_misconfigured", status: 503 } + ); + } + return url; +} + +function buildSignerAuth( + context: Awaited> +) { + if (!context.signerUrl) { + return null; + } + return { + signerUrl: context.signerUrl, + jwt: context.jwt, + }; +} + +export async function forwardRunnerRequest( + input: ForwardRunnerRequestInput +): Promise { + const discoveryUrl = input.discoveryUrl.trim() || readDiscoveryUrl(); + let signer: ReturnType = null; + if (isRunnerSignerConfigured()) { + const signerContext = await getSignerContext(input.externalUserId); + signer = buildSignerAuth(signerContext); + } + const stream = Boolean(input.payload.stream); + + let session: LiveRunnerSession | null = null; + try { + session = await reserveSession({ + discoveryUrl, + app: input.appId, + signer, + }); + + const runnerUrl = `${session.appUrl.replace(/\/+$/, "")}/${input.runnerPath.replace(/^\/+/, "")}`; + + if (stream) { + const streamResult = await callRunnerStream({ + runnerUrl, + runner: session.runner, + payload: input.payload, + signer, + capability: input.appId, + }); + const headers = new Headers(streamResult.response.headers); + if (!headers.has("Content-Type")) { + headers.set("Content-Type", "text/event-stream"); + } + return new Response(streamResult.response.body, { + status: streamResult.response.status, + headers, + }); + } + + const result = await callRunner({ + runnerUrl, + runner: session.runner, + payload: input.payload, + signer, + capability: input.appId, + }); + + return Response.json(result.data); + } finally { + if (session) { + try { + await stopRunnerSession(session); + } catch { + // Release is best-effort — an unreleased session blocks capacity-1 runners. + } + } + } +} + +export function isRunnerGatewayConfigured(): boolean { + return Boolean(process.env.RUNNER_DISCOVERY_URL?.trim()); +} diff --git a/lib/runner-gateway/index.ts b/lib/runner-gateway/index.ts new file mode 100644 index 0000000..f57e4b4 --- /dev/null +++ b/lib/runner-gateway/index.ts @@ -0,0 +1,24 @@ +export { + RunnerGatewayError, + NoRunnerAvailableError, + RunnerHttpError, +} from "@/lib/runner-gateway/errors"; +export { + discoverRunners, + discoverRunnerCandidates, + type LiveRunnerInstance, +} from "@/lib/runner-gateway/discovery"; +export { + callRunner, + callRunnerStream, + reserveSession, + type LiveRunnerSession, + type LiveRunnerCallResult, + type LiveRunnerStreamResult, +} from "@/lib/runner-gateway/call-runner"; +export { stopRunnerSession } from "@/lib/runner-gateway/stop-session"; +export { + forwardRunnerRequest, + isRunnerGatewayConfigured, + type ForwardRunnerRequestInput, +} from "@/lib/runner-gateway/forward"; diff --git a/lib/runner-gateway/stop-session.ts b/lib/runner-gateway/stop-session.ts new file mode 100644 index 0000000..c22c719 --- /dev/null +++ b/lib/runner-gateway/stop-session.ts @@ -0,0 +1,61 @@ +import { RunnerGatewayError } from "@/lib/runner-gateway/errors"; +import type { LiveRunnerSession } from "@/lib/runner-gateway/call-runner"; + +function joinEndpoint(base: string, suffix: string): string { + const trimmed = base.replace(/\/+$/, ""); + const path = suffix.startsWith("/") ? suffix : `/${suffix}`; + return `${trimmed}${path}`; +} + +export async function stopRunnerSession( + session: LiveRunnerSession, + options?: { timeoutMs?: number } +): Promise { + const runnerUrl = session.runnerUrl.trim(); + const sessionId = session.sessionId.trim(); + if (!runnerUrl) { + throw new RunnerGatewayError( + "Live runner session stop requires runner_url", + { + code: "runner_error", + status: 400, + } + ); + } + if (!sessionId) { + throw new RunnerGatewayError( + "Live runner session stop requires session_id", + { + code: "runner_error", + status: 400, + } + ); + } + + const encoded = encodeURIComponent(sessionId); + const url = joinEndpoint(runnerUrl, `/${encoded}/stop`); + + const controller = new AbortController(); + const timeout = setTimeout( + () => controller.abort(), + options?.timeoutMs ?? 10_000 + ); + + try { + const response = await fetch(url, { + method: "POST", + headers: { Accept: "application/json" }, + signal: controller.signal, + cache: "no-store", + }); + if (!response.ok) { + const body = await response.text().catch(() => ""); + throw new RunnerGatewayError( + `Failed to stop runner session: HTTP ${response.status}${body ? ` — ${body.slice(0, 200)}` : ""}`, + { code: "runner_error", status: 502 } + ); + } + } finally { + clearTimeout(timeout); + } +} diff --git a/lib/runner-gateway/tls.ts b/lib/runner-gateway/tls.ts new file mode 100644 index 0000000..4135494 --- /dev/null +++ b/lib/runner-gateway/tls.ts @@ -0,0 +1,18 @@ +import "server-only"; + +/** + * Local/dev orchestrators (e.g. kiloutcorp.link) often present self-signed TLS. + * When RUNNER_GATEWAY_ALLOW_INSECURE_TLS=1, Node's fetch will accept those certs. + * Do not enable in production. + */ +export function applyRunnerGatewayTlsPolicy(): void { + const allowInsecure = process.env.RUNNER_GATEWAY_ALLOW_INSECURE_TLS === "1"; + const isProduction = + process.env.NODE_ENV === "production" || + process.env.VERCEL_ENV === "production"; + if (allowInsecure && !isProduction) { + process.env.NODE_TLS_REJECT_UNAUTHORIZED = "0"; + } +} + +applyRunnerGatewayTlsPolicy(); diff --git a/next.config.ts b/next.config.ts index 4f79207..967e28f 100644 --- a/next.config.ts +++ b/next.config.ts @@ -24,16 +24,17 @@ const nextConfig: NextConfig = { // /models/[id] to /apps/[id] (one noun — "app" — for the object across // both the consumer catalog and the operator surfaces). { - source: "/models/:id", - destination: "/apps/:id", + source: "/models/:path*", + destination: "/apps/:path*", permanent: true, }, // The operator console folded into the app page as ownership-gated tabs, // so the separate /manage route is gone. Deep-link the console via - // /apps/[id]?tab=overview instead. + // /apps/[...id]?tab=overview instead. `:path*` preserves slash-y + // capability ids (e.g. livepeer-example/hello-world). { - source: "/apps/:id/manage", - destination: "/apps/:id?tab=overview", + source: "/apps/:path*/manage", + destination: "/apps/:path*?tab=overview", permanent: true, }, // Old livepeer.org routes → new site equivalents diff --git a/package.json b/package.json index c61a55e..4dfcc4a 100644 --- a/package.json +++ b/package.json @@ -14,10 +14,12 @@ }, "dependencies": { "@auth0/nextjs-auth0": "^4.27.0", - "@pymthouse/builder-sdk": "^0.6.5", + "@pymthouse/builder-sdk": "0.7.1-rc.0", "framer-motion": "^11.15.0", "geist": "^1.7.0", + "jmuxer": "^2.1.1", "lucide-react": "^1.6.0", + "mux.js": "^6.3.0", "next": "^15.1.0", "react": "^19.0.0", "react-dom": "^19.0.0", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index fb66ebe..81ce89e 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -12,17 +12,23 @@ importers: specifier: ^4.27.0 version: 4.27.0(next@15.5.14(react-dom@19.2.4(react@19.2.4))(react@19.2.4))(react-dom@19.2.4(react@19.2.4))(react@19.2.4) '@pymthouse/builder-sdk': - specifier: ^0.6.5 - version: 0.6.5 + specifier: 0.7.1-rc.0 + version: 0.7.1-rc.0 framer-motion: specifier: ^11.15.0 version: 11.18.2(react-dom@19.2.4(react@19.2.4))(react@19.2.4) geist: specifier: ^1.7.0 version: 1.7.0(next@15.5.14(react-dom@19.2.4(react@19.2.4))(react@19.2.4)) + jmuxer: + specifier: ^2.1.1 + version: 2.1.1 lucide-react: specifier: ^1.6.0 version: 1.7.0(react@19.2.4) + mux.js: + specifier: ^6.3.0 + version: 6.3.0 next: specifier: ^15.1.0 version: 15.5.14(react-dom@19.2.4(react@19.2.4))(react@19.2.4) @@ -92,6 +98,10 @@ packages: react: ^18.0.0 || ~19.0.1 || ~19.1.2 || ^19.2.1 react-dom: ^18.0.0 || ~19.0.1 || ~19.1.2 || ^19.2.1 + '@babel/runtime@7.29.7': + resolution: {integrity: sha512-Nq8OhGWiZIZGV6hLHoyAKLLcJihP/xFeBMGJoUrxTX2psI8dCifzLhZISFb+VWS3wFMRDmCGw5R+dOySCqPLhw==} + engines: {node: '>=6.9.0'} + '@edge-runtime/cookies@5.0.2': resolution: {integrity: sha512-Sd8LcWpZk/SWEeKGE8LT6gMm5MGfX/wm+GPnh1eBEtCpya3vYqn37wYknwAHw92ONoyyREl1hJwxV/Qx2DWNOg==} engines: {node: '>=16'} @@ -408,8 +418,8 @@ packages: '@panva/hkdf@1.2.1': resolution: {integrity: sha512-6oclG6Y3PiDFcoyk8srjLfVKyMfVCKJ27JwNPViuXziFpmdz+MZnZN/aKY0JGXgYuO/VghU0jcOAZgWXZ1Dmrw==} - '@pymthouse/builder-sdk@0.6.5': - resolution: {integrity: sha512-oWQC3y7vTqOKG+fEaZRc7TtbIMJaLZZoS6kCqBzF5M4489wGS31LlE2gwhYDEuVNAeyoJwaK/Uu+mfEe8TwGFw==} + '@pymthouse/builder-sdk@0.7.1-rc.0': + resolution: {integrity: sha512-H1nodCLM7UJsW/OwPW/p1KlwG0G4WRnoQJIgf79cJ9L2rM6kJPkp1vA4rKYLnmLDQ0ODbS9bnxQqaQKtL+b1HA==} engines: {node: '>=20'} '@reduxjs/toolkit@2.11.2': @@ -987,6 +997,9 @@ packages: resolution: {integrity: sha512-35mSku4ZXK0vfCuHEDAwt55dg2jNajHZ1odvF+8SSr82EsZY4QmXfuWso8oEd8zRhVObSN18aM0CjSdoBX7zIw==} engines: {node: '>=0.10.0'} + dom-walk@0.1.2: + resolution: {integrity: sha512-6QvTW9mrGeIegrFXdtQi9pk7O/nSK6lSdXW2eqUspN5LWD7UTji2Fqw5V2YLjBpHEoU9Xl/eUWNpDeZvoyOv2w==} + dunder-proto@1.0.1: resolution: {integrity: sha512-KIN/nDJBQRcXw0MLVhZE9iQHmG68qAVIBg9CqmUYjmQIhgij9U5MFvrqkUL5FbtyyzZuOeOt0zdeRe4UY7ct+A==} engines: {node: '>= 0.4'} @@ -1270,6 +1283,9 @@ packages: resolution: {integrity: sha512-XxwI8EOhVQgWp6iDL+3b0r86f4d6AX6zSU55HfB4ydCEuXLXc5FcYeOu+nnGftS4TEju/11rt4KJPTMgbfmv4A==} engines: {node: '>=10.13.0'} + global@4.4.0: + resolution: {integrity: sha512-wv/LAoHdRE3BeTGz53FAamhGlPLhlssK45usmGFThIi4XqnBmjKQ16u+RNbP7WvigRZDxUsM0J3gcQ5yicaL0w==} + globals@14.0.0: resolution: {integrity: sha512-oahGvuMGQlPw/ivIYBjVSrWAfWLBeku5tpPE2fOPLi+WHffIWbuh2tCjhyQhTBPMf5E9jDEH4FOmTYgYwbKwtQ==} engines: {node: '>=18'} @@ -1459,6 +1475,9 @@ packages: resolution: {integrity: sha512-ekilCSN1jwRvIbgeg/57YFh8qQDNbwDb9xT/qu2DAHbFFZUicIl4ygVaAvzveMhMVr3LnpSKTNnwt8PoOfmKhQ==} hasBin: true + jmuxer@2.1.1: + resolution: {integrity: sha512-3UoON/ghRpXS0xA8mQcXn0aN0Sb1JD3kWfaag4V8P118O1iZOkcLd559QnByyuRPtsP0zn6G+L4H0hZueGbP/g==} + jose@6.2.10: resolution: {integrity: sha512-iiW7J9qRFlGxvCOIBDBDxFePQSn7ZMAnrYGhrrOo6siO/MIqwfyilLR27pkfDgUk+raLuzADS8A3S/KLBisc0g==} @@ -1605,6 +1624,9 @@ packages: resolution: {integrity: sha512-PXwfBhYu0hBCPw8Dn0E+WDYb7af3dSLVWKi3HGv84IdF4TyFoC0ysxFd0Goxw7nSv4T/PzEJQxsYsEiFCKo2BA==} engines: {node: '>=8.6'} + min-document@2.19.2: + resolution: {integrity: sha512-8S5I8db/uZN8r9HSLFVWPdJCvYOejMcEC82VIzNUc6Zkklf/d1gg2psfE79/vyhWOj4+J8MtwmoOz3TmvaGu5A==} + minimatch@10.2.5: resolution: {integrity: sha512-MULkVLfKGYDFYejP07QOurDLLQpcjk7Fw+7jXS2R2czRQzR56yHRveU5NDJEOviH+hETZKSkIk5c+T23GjFUMg==} engines: {node: 18 || 20 || >=22} @@ -1624,6 +1646,11 @@ packages: ms@2.1.3: resolution: {integrity: sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==} + mux.js@6.3.0: + resolution: {integrity: sha512-/QTkbSAP2+w1nxV+qTcumSDN5PA98P0tjrADijIzQHe85oBK3Akhy9AHlH0ne/GombLMz1rLyvVsmrgRxoPDrQ==} + engines: {node: '>=8', npm: '>=5'} + hasBin: true + nanoid@3.3.11: resolution: {integrity: sha512-N8SpfPUnUp1bK+PMYW8qSWdl9U+wwNWI4QKxOYDy9JAro3WMX7p2OeVRF9v+347pnakNevPmiHhNmZ2HbFA76w==} engines: {node: ^10 || ^12 || ^13.7 || ^14 || >=15.0.1} @@ -1763,6 +1790,10 @@ packages: engines: {node: '>=14'} hasBin: true + process@0.11.10: + resolution: {integrity: sha512-cdGef/drWFoydD1JsMzuFf8100nZl+GT+yacc2bEced5f9Rjk4z+WtFUTBu9PhOi9j/jfmBPu0mMEY4wIdAF8A==} + engines: {node: '>= 0.6.0'} + prop-types@15.8.1: resolution: {integrity: sha512-oj87CgZICdulUohogVAR7AjlC0327U4el4L6eAvOqCeudMDVU0NThNaV+b9Df4dXgSP1gXMTnPdhfe/2qDH5cg==} @@ -2102,6 +2133,8 @@ snapshots: react-dom: 19.2.4(react@19.2.4) swr: 2.5.1(react@19.2.4) + '@babel/runtime@7.29.7': {} + '@edge-runtime/cookies@5.0.2': {} '@emnapi/core@1.9.2': @@ -2346,7 +2379,7 @@ snapshots: '@panva/hkdf@1.2.1': {} - '@pymthouse/builder-sdk@0.6.5': + '@pymthouse/builder-sdk@0.7.1-rc.0': dependencies: oauth4webapi: 3.8.7 @@ -2895,6 +2928,8 @@ snapshots: dependencies: esutils: 2.0.3 + dom-walk@0.1.2: {} + dunder-proto@1.0.1: dependencies: call-bind-apply-helpers: 1.0.2 @@ -3328,6 +3363,11 @@ snapshots: dependencies: is-glob: 4.0.3 + global@4.4.0: + dependencies: + min-document: 2.19.2 + process: 0.11.10 + globals@14.0.0: {} globalthis@1.0.4: @@ -3511,6 +3551,8 @@ snapshots: jiti@2.6.1: {} + jmuxer@2.1.1: {} + jose@6.2.10: {} js-tokens@4.0.0: {} @@ -3627,6 +3669,10 @@ snapshots: braces: 3.0.3 picomatch: 2.3.2 + min-document@2.19.2: + dependencies: + dom-walk: 0.1.2 + minimatch@10.2.5: dependencies: brace-expansion: 5.0.5 @@ -3645,6 +3691,11 @@ snapshots: ms@2.1.3: {} + mux.js@6.3.0: + dependencies: + '@babel/runtime': 7.29.7 + global: 4.4.0 + nanoid@3.3.11: {} napi-postinstall@0.3.4: {} @@ -3787,6 +3838,8 @@ snapshots: prettier@3.8.1: {} + process@0.11.10: {} + prop-types@15.8.1: dependencies: loose-envify: 1.4.0 diff --git a/tsconfig.json b/tsconfig.json index d8b9323..d4663f8 100644 --- a/tsconfig.json +++ b/tsconfig.json @@ -23,5 +23,5 @@ } }, "include": ["next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts"], - "exclude": ["node_modules"] + "exclude": ["node_modules", "**/*.test.ts"] } diff --git a/types/jmuxer.d.ts b/types/jmuxer.d.ts new file mode 100644 index 0000000..5939493 --- /dev/null +++ b/types/jmuxer.d.ts @@ -0,0 +1,12 @@ +declare module "jmuxer" { + export default class JMuxer { + constructor(options: Record); + feed(payload: { + video?: Uint8Array; + audio?: Uint8Array; + duration?: number; + }): void; + destroy(): void; + reset(): void; + } +} diff --git a/types/muxjs.d.ts b/types/muxjs.d.ts new file mode 100644 index 0000000..ab82480 --- /dev/null +++ b/types/muxjs.d.ts @@ -0,0 +1,18 @@ +declare module "mux.js" { + type MuxSegment = { + initSegment?: Uint8Array; + data: Uint8Array; + }; + + const muxjs: { + mp4: { + Transmuxer: new () => { + on: (event: "data", handler: (segment: MuxSegment) => void) => void; + push: (chunk: Uint8Array) => void; + flush: () => void; + }; + }; + }; + + export default muxjs; +}