Flue preview reliability: rollout-safe gateway and minimal config CLI

cloudflare-workers/oc-gateway/scripts/mint.ts

@@ -2,16 +2,23 @@
 // prod, binding the token as the tenant script's OC_SESSION_TOKEN env var; this is the dev/e2e helper.
 // On first use it also generates an Ed25519 keypair.
 //
-// Generate + mint (PUBLIC + PRIVATE key printed on stderr; the token on stdout):
+// Generate + mint (the exact CP + gateway provisioning values on stderr; token on stdout):
 //   node --experimental-strip-types scripts/mint.ts --org org_1 --agent agt_1 --ep 1
 //     → set the gateway's GATEWAY_TOKEN_PUBLIC_KEY secret from the printed value.
-// Reuse a private key so the gateway's public key stays fixed across mints:
-//   GATEWAY_TOKEN_PRIVATE_KEY=<b64url-pkcs8> node ... scripts/mint.ts --org org_1 --agent agt_1 --ep 2
+// Reuse the control-plane private value so the gateway public key stays fixed:
+//   V3_GATEWAY_TOKEN_PRIVATE_KEY=<base64-pkcs8-pem> node ... scripts/mint.ts ...
 
 import { mintDeployToken, type DeployClaims } from "../src/token.ts";
 
 const b64url = (buf: ArrayBuffer) => Buffer.from(buf).toString("base64url");
-const fromB64url = (s: string) => Buffer.from(s, "base64url");
+const pemFromDer = (der: ArrayBuffer) => {
+  const body = Buffer.from(der).toString("base64").match(/.{1,64}/g)?.join("\n") ?? "";
+  return `-----BEGIN PRIVATE KEY-----\n${body}\n-----END PRIVATE KEY-----\n`;
+};
+const derFromPemB64 = (value: string) => {
+  const pem = Buffer.from(value, "base64").toString("utf8");
+  return Buffer.from(pem.replace(/-----[^-]+-----/g, "").replace(/\s/g, ""), "base64");
+};
 
 function arg(name: string, def?: string): string | undefined {
   const i = process.argv.indexOf(`--${name}`);
@@ -21,14 +28,17 @@ function arg(name: string, def?: string): string | undefined {
 const ED = { name: "Ed25519" } as const;
 
 let privateKey: CryptoKey;
-const existing = process.env.GATEWAY_TOKEN_PRIVATE_KEY;
+const existing = process.env.V3_GATEWAY_TOKEN_PRIVATE_KEY;
 if (existing) {
-  privateKey = await crypto.subtle.importKey("pkcs8", fromB64url(existing), ED, true, ["sign"]);
+  privateKey = await crypto.subtle.importKey("pkcs8", derFromPemB64(existing), ED, true, ["sign"]);
 } else {
   const kp = (await crypto.subtle.generateKey(ED, true, ["sign", "verify"])) as CryptoKeyPair;
   privateKey = kp.privateKey;
-  console.error(`GATEWAY_TOKEN_PUBLIC_KEY=${b64url(await crypto.subtle.exportKey("raw", kp.publicKey))}`);
-  console.error(`GATEWAY_TOKEN_PRIVATE_KEY=${b64url(await crypto.subtle.exportKey("pkcs8", kp.privateKey))}`);
+  const publicValue = b64url(await crypto.subtle.exportKey("raw", kp.publicKey));
+  const privateValue = Buffer.from(pemFromDer(await crypto.subtle.exportKey("pkcs8", kp.privateKey)), "utf8").toString("base64");
+  console.error(`V3_GATEWAY_TOKEN_PRIVATE_KEY=${privateValue}`);
+  console.error(`V3_GATEWAY_TOKEN_PUBLIC_KEY=${publicValue}`);
+  console.error(`GATEWAY_TOKEN_PUBLIC_KEY=${publicValue}`);
 }
 
 const now = Math.floor(Date.now() / 1000);

cloudflare-workers/oc-gateway/src/budget.ts

@@ -59,8 +59,10 @@ export class SpendCounter {
 
     // POST /check {default_budget_micro?} → GATE a NEW call (used ONLY at the hard org+agt grain). On
     // first sight of an unprovisioned key, adopt the gateway's default cap (server-side; never from the
-    // token). Allowed while spent < budget; the last in-flight call can overshoot by at most one call's
-    // cost — bounded and acceptable for "refuse past the limit". Runs BEFORE the model call, DO-serialized.
+    // token). Allowed while spent < budget. Because /add lands after each response, the real overshoot
+    // bound is budget + the sum of every call concurrently in flight at the check boundary (not one
+    // call). The org's OpenRouter key cap remains the hard monetary ceiling. Runs before the call and
+    // is DO-serialized.
     if (url.pathname === "/check") {
       const s = await this.load();
       if (!s.provisioned && typeof body.default_budget_micro === "number") {

cloudflare-workers/oc-gateway/src/deploylease.ts

@@ -37,7 +37,8 @@ export class DeployLease {
     if (url.pathname === "/gate") {
       const l = await this.load();
       const ep = typeof body.ep === "number" ? body.ep : null;
-      if (ep == null) return Response.json({ ok: true, fenced: false, floor: l.floor }); // no epoch → no fence
+      // Rollout compatibility only while unprovisioned. Once a floor exists, omission fails closed.
+      if (ep == null) return Response.json({ ok: l.floor === 0, fenced: l.floor > 0, floor: l.floor, missing_epoch: l.floor > 0 });
       if (ep < l.floor) return Response.json({ ok: false, fenced: true, floor: l.floor });
       if (ep > l.floor) {
         l.floor = ep;

cloudflare-workers/oc-gateway/src/index.ts

@@ -81,6 +81,18 @@ function bearer(h: Headers): string | null {
   return x ? x.trim() : null;
 }
 
+function timingSafeEqual(a: string, b: string): boolean {
+  if (a.length !== b.length) return false;
+  let diff = 0;
+  for (let i = 0; i < a.length; i++) diff |= a.charCodeAt(i) ^ b.charCodeAt(i);
+  return diff === 0;
+}
+
+function validSessionId(value: string | null): string | null {
+  const id = value?.trim() ?? "";
+  return id.length <= 128 && /^ses_[A-Za-z0-9_-]+$/.test(id) ? id : null;
+}
+
 const json = (obj: unknown, status = 200) =>
   new Response(JSON.stringify(obj), { status, headers: { "content-type": "application/json" } });
 
@@ -112,7 +124,7 @@ export default {
 
     // Best-effort per-session attribution — the header may be stale under co-location; used for
     // tracking only, never gated. Absent → we simply skip the per-session record (the call proceeds).
-    const sessionId = req.headers.get("x-oc-session")?.trim() || null;
+    const sessionId = validSessionId(req.headers.get("x-oc-session"));
 
     // (b) lease-epoch fence — a rotated/revoked deploy token stops verifying (per org+agt, DO-serialized).
     const leaseStub = env.DEPLOY_LEASE.get(env.DEPLOY_LEASE.idFromName(`${orgId}:${agentId}`));
@@ -186,7 +198,9 @@ export default {
     ctx.waitUntil(
       meter(meterCopy, isStream, env, {
         agentBudget,
-        sessionCounter: sessionId ? env.SPEND_COUNTER.get(env.SPEND_COUNTER.idFromName(`sess:${sessionId}`)) : null,
+        sessionCounter: sessionId
+          ? env.SPEND_COUNTER.get(env.SPEND_COUNTER.idFromName(`sess:${orgId}:${agentId}:${sessionId}`))
+          : null,
         sessionId, orgId, agentId,
       }),
     );
@@ -208,7 +222,7 @@ function parseUsdMicro(usd?: string): number | null {
 async function admin(req: Request, env: Env, url: URL): Promise<Response> {
   if (!env.GATEWAY_ADMIN_SECRET) return json({ error: { type: "not_found" } }, 404);
   const auth = bearer(req.headers);
-  if (auth !== env.GATEWAY_ADMIN_SECRET) return json({ error: { type: "unauthorized" } }, 401);
+  if (!auth || !timingSafeEqual(auth, env.GATEWAY_ADMIN_SECRET)) return json({ error: { type: "unauthorized" } }, 401);
   if (req.method !== "POST") return json({ error: { type: "method_not_allowed" } }, 405);
   const body = (await req.json().catch(() => ({}))) as Record<string, unknown>;
 

cloudflare-workers/oc-gateway/src/token.ts

@@ -94,6 +94,9 @@ export async function verifyDeployToken(publicKeyB64url: string, token: string,
   if (typeof claims.exp !== "number" || claims.exp <= nowSec) return { ok: false, reason: "expired" };
   if (typeof claims.iat === "number" && claims.iat > nowSec + 60) return { ok: false, reason: "future_iat" };
   if (!claims.org || !claims.agt) return { ok: false, reason: "missing_claims" };
+  if (claims.ep !== undefined && (!Number.isSafeInteger(claims.ep) || claims.ep < 0)) {
+    return { ok: false, reason: "bad_epoch" };
+  }
   return { ok: true, claims };
 }
 

cloudflare-workers/oc-gateway/test/integration.test.ts

@@ -180,8 +180,8 @@ describe("gateway on-path flow (resolved seam)", () => {
     expect(ry.status).toBe(200);
     expect(rz.status).toBe(402); // org+agt cap hit across sessions
     // best-effort per-session tracking recorded each session's own spend separately
-    expect((await stateOf(spend.instances.get("sess:ses_x")!)).spent_micro).toBe(20_000);
-    expect((await stateOf(spend.instances.get("sess:ses_y")!)).spent_micro).toBe(20_000);
+    expect((await stateOf(spend.instances.get("sess:org_1:agt_1:ses_x")!)).spent_micro).toBe(20_000);
+    expect((await stateOf(spend.instances.get("sess:org_1:agt_1:ses_y")!)).spent_micro).toBe(20_000);
     // and the authoritative org+agt grain summed them
     expect((await stateOf(spend.instances.get("agt:org_1:agt_1")!)).spent_micro).toBe(40_000);
   });
@@ -192,7 +192,7 @@ describe("gateway on-path flow (resolved seam)", () => {
     const token = await mint();
     for (let i = 0; i < 3; i++) { const c = ctx(); const r = await worker.fetch(post(token, "ses_hot"), env, c); await drain(c); expect(r.status).toBe(200); }
     // the session accumulated $0.30 but was never blocked (no per-session hard gate)
-    expect((await stateOf(spend.instances.get("sess:ses_hot")!)).spent_micro).toBe(300_000);
+    expect((await stateOf(spend.instances.get("sess:org_1:agt_1:ses_hot")!)).spent_micro).toBe(300_000);
   });
 
   it("fences a superseded deploy lease epoch (401 token_superseded)", async () => {

cloudflare-workers/oc-gateway/test/logic.test.ts

@@ -116,10 +116,12 @@ describe("DeployLease DO — lease-epoch fence + bump (revocation)", () => {
     expect(stale.fenced).toBe(true);
     expect(stale.ok).toBe(false);
   });
-  it("a token with no epoch is never fenced (opt-in fence)", async () => {
+  it("an epoch-less token fails closed once a floor exists", async () => {
     const l = new DeployLease(fakeState());
     await call(l, "/gate", { ep: 5 }); // raise the floor
-    expect((await call(l, "/gate", {})).fenced).toBe(false); // no ep → passes
+    const missing = await call(l, "/gate", {});
+    expect(missing.fenced).toBe(true);
+    expect(missing.ok).toBe(false);
   });
   it("bump revokes without a redeploy (raise the floor above the live token)", async () => {
     const l = new DeployLease(fakeState());

cmd/oc/internal/commands/agent.go

@@ -173,6 +173,7 @@ var agentCmd = &cobra.Command{
 func init() {
 	registerAgentCrud()
 	registerAgentDeploy()
+	registerAgentConfig()
 	registerAgentSchedules()
 	rootCmd.AddCommand(sessionCmd)
 }

cmd/oc/internal/commands/agent_config.go

@@ -0,0 +1,157 @@
+package commands
+
+// Preview Flue configuration has one source of truth: non-secret vars come from agent.toml and
+// write-only secrets come from this command. Both are applied by the next explicit agent deploy.
+
+import (
+	"fmt"
+	"io"
+	"net/url"
+	"os"
+	"strings"
+
+	"github.com/opensandbox/opensandbox/cmd/oc/internal/client"
+	"github.com/spf13/cobra"
+)
+
+type agentConfig struct {
+	Vars               map[string]string `json:"vars"`
+	DeploymentRequired bool              `json:"deployment_required,omitempty"`
+}
+
+type agentSecret struct {
+	Name               string `json:"name"`
+	Last4              string `json:"last4"`
+	UpdatedAt          string `json:"updated_at"`
+	DeploymentRequired bool   `json:"deployment_required,omitempty"`
+}
+
+type agentSecretList struct {
+	Data []agentSecret `json:"data"`
+}
+
+func agentConfigPath(id string) string  { return "/v3/agents/" + id + "/config" }
+func agentSecretsPath(id string) string { return "/v3/agents/" + id + "/secrets" }
+func agentSecretPath(id, name string) string {
+	return agentSecretsPath(id) + "/" + url.PathEscape(name)
+}
+
+// syncManifestVars makes agent.toml the only non-secret config source. An absent [vars] section is
+// an empty desired map, so deleting the section and deploying removes prior bindings.
+func syncManifestVars(cmd *cobra.Command, sc *client.Client, id string, m *manifest) error {
+	vars := m.Vars
+	if vars == nil {
+		vars = map[string]string{}
+	}
+	var saved agentConfig
+	if err := sc.PutJSON(cmd.Context(), agentConfigPath(id), map[string]interface{}{"vars": vars}, &saved); err != nil {
+		return fmt.Errorf("sync agent.toml [vars]: %w", err)
+	}
+	return nil
+}
+
+var agentSecretCmd = &cobra.Command{
+	Use:   "secret",
+	Short: "Manage write-only Flue Worker secrets for the next deploy",
+}
+
+var agentSecretListCmd = &cobra.Command{
+	Use:     "list",
+	Aliases: []string{"ls"},
+	Short:   "List secret metadata (values are never returned)",
+	Args:    cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, _ []string) error {
+		sc, err := sessionsClient(cmd)
+		if err != nil {
+			return err
+		}
+		id, err := targetAgentID(cmd, sc, nil)
+		if err != nil {
+			return err
+		}
+		var res agentSecretList
+		if err := sc.Get(cmd.Context(), agentSecretsPath(id), &res); err != nil {
+			return err
+		}
+		printer.Print(res.Data, func() {
+			if len(res.Data) == 0 {
+				fmt.Println("No secrets.")
+				return
+			}
+			rows := make([][]string, 0, len(res.Data))
+			for _, secret := range res.Data {
+				rows = append(rows, []string{secret.Name, secret.Last4, formatAge(secret.UpdatedAt)})
+			}
+			printer.Table([]string{"NAME", "LAST4", "UPDATED"}, rows)
+		})
+		return nil
+	},
+}
+
+var agentSecretSetCmd = &cobra.Command{
+	Use:   "set <name> --from-stdin",
+	Short: "Save or rotate a Worker secret for the next deploy",
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		fromStdin, _ := cmd.Flags().GetBool("from-stdin")
+		if !fromStdin {
+			return fmt.Errorf("secret values are accepted only via --from-stdin")
+		}
+		raw, err := io.ReadAll(os.Stdin)
+		if err != nil {
+			return fmt.Errorf("reading stdin: %w", err)
+		}
+		value := strings.TrimSuffix(strings.TrimSuffix(string(raw), "\n"), "\r")
+		if value == "" {
+			return fmt.Errorf("secret value cannot be empty")
+		}
+
+		sc, err := sessionsClient(cmd)
+		if err != nil {
+			return err
+		}
+		id, err := targetAgentID(cmd, sc, nil)
+		if err != nil {
+			return err
+		}
+		var saved agentSecret
+		if err := sc.PutJSON(cmd.Context(), agentSecretPath(id, args[0]), map[string]string{"value": value}, &saved); err != nil {
+			return err
+		}
+		printer.Print(saved, func() {
+			fmt.Printf("Secret %s saved. Run `oc agent deploy` to apply it.\n", saved.Name)
+		})
+		return nil
+	},
+}
+
+var agentSecretDeleteCmd = &cobra.Command{
+	Use:     "delete <name>",
+	Aliases: []string{"rm"},
+	Short:   "Remove a Worker secret on the next deploy",
+	Args:    cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		sc, err := sessionsClient(cmd)
+		if err != nil {
+			return err
+		}
+		id, err := targetAgentID(cmd, sc, nil)
+		if err != nil {
+			return err
+		}
+		if err := sc.Delete(cmd.Context(), agentSecretPath(id, args[0])); err != nil {
+			return err
+		}
+		fmt.Printf("Secret %s removed. Run `oc agent deploy` to apply it.\n", args[0])
+		return nil
+	},
+}
+
+func registerAgentConfig() {
+	for _, command := range []*cobra.Command{agentSecretListCmd, agentSecretSetCmd, agentSecretDeleteCmd} {
+		command.Flags().String("agent", "", "Target agent id or name (else the cwd agent.toml)")
+	}
+	agentSecretSetCmd.Flags().Bool("from-stdin", false, "Read the secret value from stdin")
+	agentSecretCmd.AddCommand(agentSecretListCmd, agentSecretSetCmd, agentSecretDeleteCmd)
+	agentCmd.AddCommand(agentSecretCmd)
+}

cmd/oc/internal/commands/agent_config_test.go

@@ -0,0 +1,22 @@
+package commands
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestAgentSecretSetRejectsPositionalValue(t *testing.T) {
+	if err := agentSecretSetCmd.Args(agentSecretSetCmd, []string{"TOKEN", "secret-value"}); err == nil {
+		t.Fatal("expected a positional secret value to be rejected")
+	}
+}
+
+func TestAgentSecretSetRequiresStdinFlag(t *testing.T) {
+	if err := agentSecretSetCmd.Flags().Set("from-stdin", "false"); err != nil {
+		t.Fatal(err)
+	}
+	err := agentSecretSetCmd.RunE(agentSecretSetCmd, []string{"TOKEN"})
+	if err == nil || !strings.Contains(err.Error(), "--from-stdin") {
+		t.Fatalf("error = %v, want --from-stdin requirement", err)
+	}
+}

cmd/oc/internal/commands/agent_deploy.go

@@ -16,8 +16,11 @@ import (
 // ── agent.toml manifest + deploy bundle ──
 
 type manifest struct {
-	Name    string `toml:"name"`
-	Model   string `toml:"model"`
+	Name  string `toml:"name"`
+	Model string `toml:"model"`
+	// Non-secret Flue Worker bindings. The manifest is authoritative: omitting
+	// [vars] clears prior values on deploy. Secrets never belong in agent.toml.
+	Vars    map[string]string `toml:"vars"`
 	Runtime struct {
 		Family string `toml:"family"`
 		Type   string `toml:"type"`

cmd/oc/internal/commands/agent_deploy_flue.go

@@ -71,6 +71,13 @@ func deployFlue(cmd *cobra.Command, sc *client.Client, dir string, m *manifest,
 	if err != nil {
 		return err
 	}
+	// [vars] is part of the deployment input even though the values live in the
+	// agent config resource. Persist it before enqueueing so the off-host runner
+	// cannot race ahead and compose the Worker with stale bindings. Secrets are
+	// intentionally CLI/API only and are resolved by that same runner.
+	if err := syncManifestVars(cmd, sc, id, m); err != nil {
+		return err
+	}
 
 	// 3. Build the app with its own `flue` CLI (a devDependency).
 	if err := runFlueBuild(cmd.Context(), dir); err != nil {