runner: Add fixed pricing

ai/runner/live_runner.go

@@ -115,8 +115,9 @@ func normalizeLiveRunnerPriceInfo(priceInfo LiveRunnerPriceInfo) (LiveRunnerPric
 	switch unit {
 	case "hour":
 	case "720p":
+	case "fixed":
 	default:
-		return LiveRunnerPriceInfo{}, fmt.Errorf("price_info.unit must be hour or 720p")
+		return LiveRunnerPriceInfo{}, fmt.Errorf("price_info.unit must be hour, 720p, or fixed")
 	}
 	priceInfo.Currency = currency
 	priceInfo.Unit = unit
@@ -243,6 +244,7 @@ type liveRunner struct {
 type liveRunnerSession struct {
 	// Protected by the parent liveRunner.mu.
 	createdAt time.Time
+	priceInfo LiveRunnerPriceInfo
 	channels  map[string]*liveRunnerTrickleChannel
 	proxies   map[string]proxyTarget
 }
@@ -957,19 +959,21 @@ func (r *LiveRunnerRegistry) ReserveSession(runnerID string, optSessionID ...str
 	if !isReadyStatus(runner.Status) {
 		return "", "", &RunnerError{StatusCode: http.StatusNotFound, Message: "runner not found"}
 	}
-	if len(runner.sessions) >= runner.Capacity {
-		return "", "", &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: "no capacity available for runner"}
-	}
 	id := ""
 	if len(optSessionID) > 0 {
 		// use supplied id if given - usually ticket param's auth token id / manifest id
 		id = optSessionID[0]
 	}
 	if id != "" {
 		if _, exists := runner.sessions[id]; exists {
+			// Consider another status code if session enumeration becomes a concern.
 			return "", "", &RunnerError{StatusCode: http.StatusConflict, Message: "session already exists"}
 		}
-	} else {
+	}
+	if len(runner.sessions) >= runner.Capacity {
+		return "", "", &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: "no capacity available for runner"}
+	}
+	if id == "" {
 		id = "session_" + randomStr(liveRunnerIDRandomBytes)
 		for {
 			if _, exists := runner.sessions[id]; !exists {
@@ -980,6 +984,7 @@ func (r *LiveRunnerRegistry) ReserveSession(runnerID string, optSessionID ...str
 	}
 	session := &liveRunnerSession{
 		createdAt: time.Now(),
+		priceInfo: runner.PriceInfo,
 		channels:  make(map[string]*liveRunnerTrickleChannel),
 		proxies:   make(map[string]proxyTarget),
 	}
@@ -1015,14 +1020,27 @@ func (r *LiveRunnerRegistry) PaymentInfo(runnerID string) (*LiveRunnerPriceInfo,
 	}
 	priceInfo, err := runner.convertPrice()
 	if err != nil {
-		return nil, err
+		return nil, &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: fmt.Sprintf("live runner price unavailable: %v", err)}
 	}
 	if _, err := priceInfo.priceRat(); err != nil {
-		return nil, nil
+		return nil, &RunnerError{StatusCode: http.StatusServiceUnavailable, Message: fmt.Sprintf("live runner price unavailable: %v", err)}
 	}
 	return &priceInfo, nil
 }
 
+func (r *LiveRunnerRegistry) SessionPriceInfo(runnerID, sessionID string) (LiveRunnerPriceInfo, error) {
+	runner, unlock, err := r.lockLiveRunner(runnerID)
+	if err != nil {
+		return LiveRunnerPriceInfo{}, err
+	}
+	defer unlock()
+	session, err := runner.liveRunnerSessionLocked(sessionID)
+	if err != nil {
+		return LiveRunnerPriceInfo{}, err
+	}
+	return session.priceInfo, nil
+}
+
 func (r *LiveRunnerRegistry) ReleaseSession(runnerID, sessionID string) error {
 	runner, unlock, err := r.lockLiveRunner(runnerID)
 	if err != nil {
@@ -1753,15 +1771,19 @@ func newConverterForRunner(priceInfo LiveRunnerPriceInfo) (*core.AutoConvertedPr
 		return nil, err
 	}
 
-	usdPerHour, err := priceInfo.priceRat()
+	usdPrice, err := priceInfo.priceRat()
 	if err != nil {
 		return nil, err
 	}
-	denom := int64(3600)
-	if priceInfo.Unit == "720p" {
-		denom *= 1280 * 720 * 30 // 720p @ 30fps
+	switch priceInfo.Unit {
+	case "hour":
+		usdPrice = new(big.Rat).Quo(usdPrice, new(big.Rat).SetInt64(3600))
+	case "720p":
+		denom := int64(3600 * 1280 * 720 * 30) // 1h of 720p @ 30fps
+		usdPrice = new(big.Rat).Quo(usdPrice, new(big.Rat).SetInt64(denom))
+	case "fixed":
+		// Fixed prices are already denominated per request.
 	}
-	usdPrice := new(big.Rat).Quo(usdPerHour, new(big.Rat).SetInt64(denom))
 	return core.NewAutoConvertedPrice("USD", usdPrice, nil)
 }
 
@@ -1773,17 +1795,27 @@ func (runner *liveRunner) convertPrice() (LiveRunnerPriceInfo, error) {
 	if err != nil {
 		return LiveRunnerPriceInfo{}, err
 	}
-	convertedUnit := "seconds"
-	if runner.PriceInfo.Unit == "720p" {
-		convertedUnit = "720p-pixel-seconds"
+	if price.Sign() <= 0 {
+		return LiveRunnerPriceInfo{}, fmt.Errorf("converted live runner price must be at least one wei")
 	}
 	return LiveRunnerPriceInfo{
 		Price:    json.Number(fmt.Sprintf("%d", price.Num().Int64())),
 		Currency: "wei",
-		Unit:     convertedUnit,
+		Unit:     convertedLiveRunnerPriceUnit(runner.PriceInfo.Unit),
 	}, nil
 }
 
+func convertedLiveRunnerPriceUnit(unit string) string {
+	switch strings.ToLower(strings.TrimSpace(unit)) {
+	case "720p":
+		return "720p-pixel-seconds"
+	case "fixed":
+		return "fixed"
+	default:
+		return "seconds"
+	}
+}
+
 func isReadyStatus(status string) bool {
 	return status == "" || status == "ready"
 }

ai/runner/live_runner_test.go

@@ -1118,6 +1118,7 @@ func TestLiveRunnerRegistry_RegisterStaticRunnersUpsertsWithoutDroppingSession(t
 		App:       "live-video-to-video/scope",
 		Capacity:  2,
 		HealthURL: healthSrv.URL,
+		PriceInfo: LiveRunnerPriceInfo{Price: json.Number("10"), Unit: "fixed"},
 	}
 	resp1, err := registry.RegisterStaticRunners(StaticLiveRunnerConfig{Runners: []StaticLiveRunnerConfigEntry{entry}})
 	if err != nil {
@@ -1130,6 +1131,7 @@ func TestLiveRunnerRegistry_RegisterStaticRunnersUpsertsWithoutDroppingSession(t
 	}
 
 	entry.RunnerURL = "https://runner-updated.example.com"
+	entry.PriceInfo = LiveRunnerPriceInfo{Price: json.Number("20"), Unit: "hour"}
 	resp2, err := registry.RegisterStaticRunners(StaticLiveRunnerConfig{Runners: []StaticLiveRunnerConfigEntry{entry}})
 	if err != nil {
 		t.Fatal(err)
@@ -1140,6 +1142,13 @@ func TestLiveRunnerRegistry_RegisterStaticRunnersUpsertsWithoutDroppingSession(t
 	if _, err := registry.RunnerEndpointForSession(runnerID, sessionID); err != nil {
 		t.Fatalf("expected active session to survive healthy static upsert: %v", err)
 	}
+	sessionPriceInfo, err := registry.SessionPriceInfo(runnerID, sessionID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if sessionPriceInfo.Price.String() != "10" || sessionPriceInfo.Unit != "fixed" {
+		t.Fatalf("expected static session price snapshot with price 10 and unit fixed, got: %+v", sessionPriceInfo)
+	}
 }
 
 func TestLiveRunnerRegistry_RegisterStaticRunnersConcurrentUpserts(t *testing.T) {

doc/live-runner.md

@@ -194,10 +194,12 @@ On an on-chain network, every runner must provide a positive `price_info.price`.
 The registration price is denominated in USD:
 
 - `currency` defaults to `usd`; no other registration currency is accepted.
-- `unit` defaults to `hour` and accepts `hour` or `720p`.
+- `unit` defaults to `hour` and accepts `hour`, `720p`, or `fixed`.
 - `hour` is converted to a wei-per-second discovery/payment price.
 - `720p` is converted using 720p at 30 fps and is advertised in
   `720p-pixel-seconds`.
+- `fixed` is converted directly from USD to wei without a time or pixel
+  divisor and is advertised as `fixed`.
 
 The conversion follows the node's USD/ETH price feed. On an offchain network,
 runner prices are not advertised and no payment challenge is required.
@@ -293,9 +295,7 @@ UTF-8 and at most 1,024 UTF-8-encoded bytes. An empty value clears previously
 registered metadata and is omitted from discovery.
 
 On-chain discovery includes the converted wei price. Offchain runner discovery
-omits it. If the orchestrator also has a serverless AI worker, the response may
-contain synthetic live-video-to-video runner records alongside registered live
-runners.
+omits it.
 
 #### Remote signer discovery
 
@@ -391,7 +391,7 @@ Required` with payment parameters. The client retries with
 `Livepeer-Payment` and `Livepeer-Segment`, then periodically refreshes payment at
 `POST /apps/{runner_id}/session/{session_id}/payment`. The orchestrator accounts
 usage on its configured payment interval and releases the session if payment
-fails.
+fails. If the payment unit is `fixed` then payment is only processed once.
 
 The current single-shot proxy path does not perform a payment challenge or
 accounting. Use persistent mode when on-chain payment enforcement is required.
@@ -420,7 +420,7 @@ are currently ignored. Labels in one submitted batch must be unique.
 | `mode` | string | Default: `persistent` | `persistent`, `single-shot`, or the normalized alias `single_shot`. |
 | `capacity` | integer | Default: `1` when zero | Maximum concurrent sessions. Negative values are invalid. |
 | `gpu` | object or integer | Optional | Object fields are `id`, `name`, and `vram_mb`. An integer is treated as a local device index; a negative index records only that numeric ID and skips hardware lookup. |
-| `price_info` | object | Required onchain; ignored for offchain payment/discovery | `price` is a positive decimal. `currency` defaults to and must equal `usd`. `unit` defaults to `hour` and accepts `hour` or `720p`. |
+| `price_info` | object | Required onchain; ignored for offchain payment/discovery | `price` is a positive decimal. `currency` defaults to and must equal `usd`. `unit` defaults to `hour` and accepts `hour`, `720p`, or `fixed`. |
 
 ### CLI flags
 
@@ -537,9 +537,9 @@ session.
 
 | Method and path | Caller / authentication | Description | Principal responses |
 | --- | --- | --- | --- |
-| `POST /apps/{runner_id}/session` | Client; no auth offchain. Onchain uses `Livepeer-Payer-Address` for the initial challenge and `Livepeer-Payment` plus `Livepeer-Segment` to reserve. | Reserves a persistent session. Returns `session_id`, `app_url`, and `control_url`; `proxy: true` registrations receive a random proxy-template `app_url`. A priced runner first returns a payment challenge. | `200`, `402`, `404`, `503` |
+| `POST /apps/{runner_id}/session` | Client; no auth offchain. Onchain uses `Livepeer-Payer-Address` for the initial challenge and `Livepeer-Payment` plus `Livepeer-Segment` to reserve. | Reserves a persistent session. Returns `session_id`, `app_url`, and `control_url`; `proxy: true` registrations receive a random proxy-template `app_url`. A priced runner first returns a payment challenge. | `200`, `400`, `402`, `404`, `409`, `503` |
 | `POST /apps/{runner_id}/session/{session_id}/stop` | Client; session is identified by the URL | Releases a persistent session, its channels, and proxies. | `204`, `404` |
-| `POST /apps/{runner_id}/session/{session_id}/payment` | Paying client; `Livepeer-Payment` and `Livepeer-Segment` | Adds payment for an active session. The payment manifest must match `session_id`. | `200`, `400`, `403`, `404` |
+| `POST /apps/{runner_id}/session/{session_id}/payment` | Paying client; `Livepeer-Payment` and `Livepeer-Segment` | Adds payment for an active session. The payment manifest must match `session_id`. Fixed-payment sessions do not need to call this. | `200`, `400`, `403`, `404`, `409` |
 | `ANY /apps/{runner_id}/session/{session_id}/app/{app_path...}` | Client; access is by the reserved public URL | Proxies any HTTP method, SSE response, or WebSocket upgrade to a persistent runner. | Upstream status, `404`, `502` |
 | `ANY /apps/{runner_id}/app/{app_path...}` | Client; no application-level authentication | Reserves a single-shot session, proxies one request, then releases it. The current path does not enforce runner payment. | Upstream status, `400`, `404`, `503`, `502` |
 

doc/remote-signer.md

@@ -125,7 +125,7 @@ Remote discovery filters data before exposing it:
   configured capability/model maximum from `-maxPricePerCapability`, falling
   back to the global maximum when applicable;
 - runner records require a URL, a non-empty `app`, and a positive wei price;
-- runner price units must be `seconds` or `720p-pixel-seconds`; and
+- runner price units must be `seconds`, `720p-pixel-seconds`, or `fixed`; and
 - the global `-maxPricePerUnit`, when set, also limits runner prices.
 
 Each valid runner's `app` is added to the address's capability list and can be