Refactor monetize reconciliation into the serviceoffer controller

.dockerignore

@@ -0,0 +1,2 @@
+.workspace/
+.worktrees/

CLAUDE.md

@@ -68,29 +68,29 @@ obol
 
 ## Infrastructure Stack
 
-Deployed on `obol stack up` from `internal/embed/infrastructure/`. Key templates in `base/templates/`: `llm.yaml` (LiteLLM + Ollama), `x402.yaml` (verifier + ConfigMap), `obol-agent.yaml` (singleton), `serviceoffer-crd.yaml`, `obol-agent-monetize-rbac.yaml`, `local-path.yaml`. Plus `cloudflared/` chart and `values/` for eRPC, monitoring, frontend.
+Deployed on `obol stack up` from `internal/embed/infrastructure/`. Key templates in `base/templates/`: `llm.yaml` (LiteLLM + Ollama), `x402.yaml` (verifier + serviceoffer-controller), `obol-agent.yaml` (singleton), `serviceoffer-crd.yaml`, `registrationrequest-crd.yaml`, `obol-agent-monetize-rbac.yaml`, `local-path.yaml`. Plus `cloudflared/` chart and `values/` for eRPC, monitoring, frontend.
 
-Components: eRPC (`erpc` ns), Frontend (`obol-frontend` ns), Cloudflared (`traefik` ns), Monitoring/Prometheus (`monitoring` ns), Reloader, LiteLLM (`llm` ns), x402-verifier (`x402` ns), obol-agent (`openclaw-obol-agent` ns), ServiceOffer CRD.
+Components: eRPC (`erpc` ns), Frontend (`obol-frontend` ns), Cloudflared (`traefik` ns), Monitoring/Prometheus (`monitoring` ns), LiteLLM (`llm` ns), x402-verifier (`x402` ns), serviceoffer-controller (`x402` ns), obol-agent (`openclaw-obol-agent` ns), ServiceOffer + RegistrationRequest CRDs.
 
 ## Monetize Subsystem
 
 Payment-gated access to cluster services via x402 (HTTP 402 micropayments, USDC on Base/Base Sepolia, Traefik ForwardAuth).
 
-**Sell-side flow**: `obol sell http` → creates ServiceOffer CR → agent reconciles 6 stages: ModelReady → UpstreamHealthy → PaymentGateReady (x402 Middleware + pricing route) → RoutePublished (HTTPRoute) → Registered (ERC-8004 on-chain) → Ready. Traefik routes `/services/<name>/*` through ForwardAuth to upstream.
+**Sell-side flow**: `obol sell http` → creates ServiceOffer CR → serviceoffer-controller reconciles ModelReady → UpstreamHealthy → PaymentGateReady (x402 Middleware) → RoutePublished (HTTPRoute) → Registered (RegistrationRequest + optional ERC-8004 side effects) → Ready. Traefik routes `/services/<name>/*` through ForwardAuth to upstream.
 
 **Buy-side flow**: `buy.py probe` sees 402 pricing → `buy.py buy` pre-signs ERC-3009 auths into ConfigMaps → LiteLLM serves static `paid/<remote-model>` aliases through the in-pod `x402-buyer` sidecar → each paid request spends one auth and forwards to the remote seller.
 
 **CLI**: `obol sell pricing --wallet --chain`, `obol sell inference <name> --model --price|--per-mtok`, `obol sell http <name> --wallet --chain --price|--per-request|--per-mtok --upstream --port --namespace --health-path`, `obol sell list|status|stop|delete`, `obol sell register --name --private-key-file`.
 
-**ServiceOffer CRD** (`obol.org`): Spec fields — `type` (inference|fine-tuning), `model{name,runtime}`, `upstream{service,ns,port,healthPath}`, `payment{scheme,network,payTo,price{perRequest,perMTok,perHour}}`, `path`, `registration{enabled,name,description,image}`. In phase 1, `perMTok` is accepted but enforced as `perRequest = perMTok / 1000`.
+**ServiceOffer CRD** (`obol.org`): Source of truth for monetized service intent. Spec fields — `type` (inference|fine-tuning|http), `model{name,runtime}`, `upstream{service,namespace,port,healthPath}`, `payment{scheme,network,payTo,price{perRequest,perMTok,perHour}}`, `path`, `registration{enabled,name,description,image,skills,domains,supportedTrust}`.
 
-**x402-verifier** (`x402` ns): ForwardAuth middleware. No match → pass through. Match + no payment → 402. Match + payment → verify with facilitator. Config in `x402-pricing` ConfigMap: `wallet`, `chain`, `facilitatorURL`, `verifyOnly`, `routes[]{pattern, price, description, priceModel, perMTok, approxTokensPerRequest, offerNamespace, offerName}`. Exposes `/metrics` and is scraped via `ServiceMonitor`.
+**x402-verifier** (`x402` ns): ForwardAuth middleware only. No match → pass through. Match + no payment → 402. Match + payment → verify with facilitator. Static defaults still come from `x402-pricing`, but live per-offer routes are derived in-memory from published ServiceOffers.
 
-**Agent reconciler** (`internal/embed/skills/monetize/scripts/monetize.py`): Watches ServiceOffer CRs, creates Middleware (`traefik.io`), HTTPRoute, pricing route in ConfigMap, registration resources (ConfigMap + httpd + HTTPRoute at `/.well-known/`). All with ownerReferences for auto-GC.
+**serviceoffer-controller** (`internal/serviceoffercontroller/`): Watches ServiceOffers and RegistrationRequests, adds finalizers, creates Middleware + HTTPRoute, publishes registration resources, and drives tombstone cleanup on delete.
 
-**ERC-8004**: On-chain registration on Base Sepolia Identity Registry (`0xEA0fE4FCF9E3017a24d9Db6e0e39B552c8648B9D`). NFT mint via remote-signer wallet, publishes `/.well-known/agent-registration.json`.
+**ERC-8004**: Registration publication is isolated behind `RegistrationRequest`. The controller serves `/.well-known/agent-registration.json` from dedicated child resources and optionally registers/tombstones on Base Sepolia when an ERC-8004 signing key is configured.
 
-**RBAC**: ClusterRole `openclaw-monetize` grants CRUD on ServiceOffers (`obol.org`), Middlewares (`traefik.io`), HTTPRoutes (`gateway.networking.k8s.io`), ConfigMaps/Services/Deployments, read Pods/Endpoints/logs. Bound to SA `openclaw` in `openclaw-obol-agent` ns. Patched by `obol agent init` via `patchMonetizeBinding()`.
+**RBAC**: The controller owns child-resource and registration write access. The agent retains read access plus minimal ServiceOffer CRUD for compatibility commands only.
 
 ## RPC Gateway
 
@@ -127,7 +127,7 @@ k3d: 1 server, ports 80:80 + 8080:80 + 443:443 + 8443:443, `rancher/k3s:v1.35.1-
 
 Skills = SKILL.md + optional scripts/references, embedded in `obol` binary (`internal/embed/skills/`, 23 skills). Delivered via host-path PVC injection to `$DATA_DIR/openclaw-<id>/openclaw-data/.openclaw/skills/`. Categories: Infrastructure (ethereum-networks, ethereum-local-wallet, obol-stack, distributed-validators, monetize, discovery), Ethereum Dev (addresses, building-blocks, concepts, gas, indexing, l2s, orchestration, security, standards, ship, testing, tools, wallets), Frontend (frontend-playbook, frontend-ux, qa, why).
 
-**Monetize skill** (`internal/embed/skills/monetize/`): `monetize.py` 6-stage reconciliation loop.
+**Monetize skill** (`internal/embed/skills/monetize/`): thin compatibility wrapper around ServiceOffer CRUD, controller waiting, and `/skill.md` publication.
 
 **Remote-signer wallet**: `GenerateWallet()` in `internal/openclaw/wallet.go`. secp256k1 → Web3 V3 keystore, remote-signer REST API at port 9000 in same ns.
 

Dockerfile.serviceoffer-controller

@@ -0,0 +1,10 @@
+FROM golang:1.25-alpine AS builder
+WORKDIR /src
+COPY go.mod go.sum ./
+RUN go mod download
+COPY . .
+RUN CGO_ENABLED=0 go build -o /serviceoffer-controller ./cmd/serviceoffer-controller
+
+FROM gcr.io/distroless/static-debian12
+COPY --from=builder /serviceoffer-controller /serviceoffer-controller
+ENTRYPOINT ["/serviceoffer-controller"]

cmd/obol/sell.go

@@ -351,7 +351,8 @@ func sellHTTPCommand(cfg *config.Config) *cli.Command {
 		Name:      "http",
 		Usage:     "Sell any local HTTP service with x402 payments",
 		ArgsUsage: "<name>",
-		Description: `Publishes a payment gated HTTP API to any service within the stack, along with a SKILL.md detailing how to use it.
+		Description: `Creates a payment gated ServiceOffer in the cluster. The serviceoffer-controller reconciles it through:
+health-check → payment gate → route publishing → optional ERC-8004 registration.
 Include --register to have the service listed on EIP8004 onchain agent registry.
 
 Example:
@@ -535,6 +536,13 @@ Example:
 
 			ns := cmd.String("namespace")
 
+			if cmd.String("upstream") == "" {
+				return fmt.Errorf("upstream service name required: use --upstream <service-name>\n\n  Example: obol sell http %s --upstream my-svc --port 8080 --wallet 0x... --chain base-sepolia --price 0.001", name)
+			}
+			if cmd.Int("port") == 0 {
+				return fmt.Errorf("upstream port required: use --port <port-number>\n\n  Example: obol sell http %s --upstream my-svc --port 8080 --wallet 0x... --chain base-sepolia --price 0.001", name)
+			}
+
 			priceTable, err := resolvePriceTable(cmd, true)
 			if err != nil {
 				return err
@@ -640,17 +648,19 @@ Example:
 				"spec": spec,
 			}
 
-			if err := kubectlApply(cfg, manifest); err != nil {
+			applyOut, err := kubectlApplyOutput(cfg, manifest)
+			if err != nil {
 				return err
 			}
-
-			fmt.Printf("ServiceOffer %s/%s created (type: http)\n", ns, name)
-
+			action := "created"
+			if strings.Contains(applyOut, "configured") || strings.Contains(applyOut, "unchanged") {
+				action = "updated"
+			}
+			fmt.Printf("ServiceOffer %s/%s %s (type: http)\n", ns, name, action)
 			if priceTable.PerMTok != "" {
 				fmt.Printf("Requests will be charged at %s\n", formatPriceTableSummary(priceTable))
 			}
-
-			fmt.Printf("The agent will reconcile: health-check → payment gate → route\n")
+			fmt.Printf("The serviceoffer-controller will reconcile: health-check → payment gate → route → registration\n")
 			fmt.Printf("Check status: obol sell status %s -n %s\n", name, ns)
 
 			// Ensure tunnel is active for public access.
@@ -992,7 +1002,7 @@ Examples:
 func sellStopCommand(cfg *config.Config) *cli.Command {
 	return &cli.Command{
 		Name:      "stop",
-		Usage:     "Stop selling a service",
+		Usage:     "Pause a ServiceOffer without deleting it",
 		ArgsUsage: "<name>",
 		Flags: []cli.Flag{
 			&cli.StringFlag{
@@ -1013,19 +1023,16 @@ func sellStopCommand(cfg *config.Config) *cli.Command {
 			}
 			ns := cmd.String("namespace")
 
-			fmt.Printf("Stopping the service offering %s/%s...\n", ns, name)
-
-			removePricingRoute(cfg, name)
-
-			patchJSON := `{"status":{"conditions":[{"type":"Ready","status":"False","reason":"Stopped","message":"Offer stopped by user"}]}}`
+			fmt.Printf("Pausing ServiceOffer %s/%s...\n", ns, name)
 
+			patchJSON := `{"metadata":{"annotations":{"obol.org/paused":"true"}}}`
 			err := kubectlRun(cfg, "patch", "serviceoffers.obol.org", name, "-n", ns,
-				"--type=merge", "--subresource=status", "-p", patchJSON)
+				"--type=merge", "-p", patchJSON)
 			if err != nil {
-				return fmt.Errorf("failed to patch status: %w", err)
+				return fmt.Errorf("failed to pause serviceoffer: %w", err)
 			}
 
-			fmt.Printf("Service offering %s/%s stopped.\n", ns, name)
+			fmt.Printf("ServiceOffer %s/%s paused.\n", ns, name)
 			return nil
 		},
 	}
@@ -1066,49 +1073,19 @@ func sellDeleteCommand(cfg *config.Config) *cli.Command {
 			ns := cmd.String("namespace")
 
 			if !cmd.Bool("force") {
-				msg := fmt.Sprintf("Delete the service offering %s/%s? This will:\n  - Remove the associated Middleware and HTTPRoute\n  - Remove the pricing route from the x402 verifier\n  - Deactivate the ERC-8004 registration (if registered)", ns, name)
+				msg := fmt.Sprintf(
+					"Delete ServiceOffer %s/%s? This will:\n  - Remove the associated Middleware and HTTPRoute\n  - Remove x402 enforcement for the service\n  - Deactivate the ERC-8004 registration (if registered)\n  - Let the serviceoffer-controller finalizer clean up published state",
+					ns,
+					name,
+				)
 				if !u.Confirm(msg, false) {
 					fmt.Println("Aborted.")
 					return nil
 				}
 			}
 
-			removePricingRoute(cfg, name)
-
-			soOut, err := kubectlOutput(cfg, "get", "serviceoffers.obol.org", name, "-n", ns,
-				"-o", "jsonpath={.status.agentId}")
-			if err == nil && strings.TrimSpace(soOut) != "" {
-				agentID := strings.TrimSpace(soOut)
-				fmt.Printf("Deactivating ERC-8004 registration (agent %s)...\n", agentID)
-
-				cmName := fmt.Sprintf("so-%s-registration", name)
-
-				rawJSON, readErr := kubectlOutput(cfg, "get", "configmap", cmName, "-n", ns,
-					"-o", `jsonpath={.data.agent-registration\.json}`)
-				if readErr != nil || strings.TrimSpace(rawJSON) == "" {
-					fmt.Printf("  No registration document found. Agent %s NFT persists on-chain.\n", agentID)
-				} else {
-					var regDoc map[string]any
-					if jsonErr := json.Unmarshal([]byte(rawJSON), &regDoc); jsonErr != nil {
-						fmt.Printf("  Warning: corrupt registration JSON, skipping deactivation: %v\n", jsonErr)
-					} else {
-						regDoc["active"] = false
-
-						patchJSON, _ := json.Marshal(map[string]any{
-							"data": map[string]string{
-								"agent-registration.json": mustMarshal(regDoc),
-							},
-						})
-						if patchErr := kubectlRun(cfg, "patch", "configmap", cmName, "-n", ns,
-							"-p", string(patchJSON), "--type=merge"); patchErr != nil {
-							fmt.Printf("  Warning: could not deactivate agent registration: %v\n", patchErr)
-						} else {
-							fmt.Printf("  Registration deactivated (active=false). On-chain NFT persists.\n")
-						}
-					}
-				}
-			}
-
+			fmt.Printf("Deleting ServiceOffer %s/%s...\n", ns, name)
+			fmt.Printf("The serviceoffer-controller finalizer will clean up published routes and registration state.\n")
 			if err := kubectlRun(cfg, "delete", "serviceoffers.obol.org", name, "-n", ns); err != nil {
 				return err
 			}
@@ -1731,15 +1708,19 @@ func sellInfoCommand(cfg *config.Config) *cli.Command {
 // kubectl helpers
 // ---------------------------------------------------------------------------
 
-func kubectlApply(cfg *config.Config, manifest any) error {
+func kubectlApply(cfg *config.Config, manifest interface{}) error {
+	_, err := kubectlApplyOutput(cfg, manifest)
+	return err
+}
+
+func kubectlApplyOutput(cfg *config.Config, manifest interface{}) (string, error) {
 	raw, err := json.Marshal(manifest)
 	if err != nil {
-		return fmt.Errorf("failed to marshal manifest: %w", err)
+		return "", fmt.Errorf("failed to marshal manifest: %w", err)
 	}
 
 	bin, kc := kubectl.Paths(cfg)
-
-	return kubectl.Apply(bin, kc, raw)
+	return kubectl.ApplyOutput(bin, kc, raw)
 }
 
 func kubectlOutput(cfg *config.Config, args ...string) (string, error) {
@@ -2019,29 +2000,3 @@ func buildInferenceServiceOfferSpec(d *inference.Deployment, pt schemas.PriceTab
 
 	return spec
 }
-
-// removePricingRoute removes the x402-verifier pricing route for the given offer.
-func removePricingRoute(cfg *config.Config, name string) {
-	urlPath := "/services/" + name
-
-	pricingCfg, err := x402verifier.GetPricingConfig(cfg)
-	if err != nil {
-		return
-	}
-
-	updatedRoutes := make([]x402verifier.RouteRule, 0, len(pricingCfg.Routes))
-	for _, r := range pricingCfg.Routes {
-		if !strings.Contains(r.Pattern, urlPath) {
-			updatedRoutes = append(updatedRoutes, r)
-		}
-	}
-
-	if len(updatedRoutes) < len(pricingCfg.Routes) {
-		pricingCfg.Routes = updatedRoutes
-		if err := x402verifier.WritePricingConfig(cfg, pricingCfg); err != nil {
-			fmt.Printf("Warning: failed to remove pricing route: %v\n", err)
-		} else {
-			fmt.Printf("Removed pricing route for %s\n", urlPath)
-		}
-	}
-}

cmd/serviceoffer-controller/main.go

@@ -0,0 +1,47 @@
+package main
+
+import (
+	"context"
+	"flag"
+	"log"
+	"os"
+	"os/signal"
+	"syscall"
+
+	"github.com/ObolNetwork/obol-stack/internal/serviceoffercontroller"
+	"k8s.io/client-go/rest"
+	"k8s.io/client-go/tools/clientcmd"
+)
+
+func main() {
+	kubeconfig := flag.String("kubeconfig", "", "Path to kubeconfig for out-of-cluster runs")
+	workers := flag.Int("workers", 2, "Number of reconcile workers")
+	flag.Parse()
+
+	cfg, err := loadConfig(*kubeconfig)
+	if err != nil {
+		log.Fatalf("load kubernetes config: %v", err)
+	}
+
+	controller, err := serviceoffercontroller.New(cfg)
+	if err != nil {
+		log.Fatalf("create controller: %v", err)
+	}
+
+	ctx, cancel := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
+	defer cancel()
+
+	if err := controller.Run(ctx, *workers); err != nil {
+		log.Fatalf("run controller: %v", err)
+	}
+}
+
+func loadConfig(kubeconfig string) (*rest.Config, error) {
+	if kubeconfig != "" {
+		return clientcmd.BuildConfigFromFlags("", kubeconfig)
+	}
+	if env := os.Getenv("KUBECONFIG"); env != "" {
+		return clientcmd.BuildConfigFromFlags("", env)
+	}
+	return rest.InClusterConfig()
+}