diff --git a/cloudflare-workers/oc-gateway/scripts/mint.ts b/cloudflare-workers/oc-gateway/scripts/mint.ts index 953c4c93..21437f61 100644 --- a/cloudflare-workers/oc-gateway/scripts/mint.ts +++ b/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= 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= 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); diff --git a/cloudflare-workers/oc-gateway/src/budget.ts b/cloudflare-workers/oc-gateway/src/budget.ts index 430ec25d..53afc5e3 100644 --- a/cloudflare-workers/oc-gateway/src/budget.ts +++ b/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") { diff --git a/cloudflare-workers/oc-gateway/src/deploylease.ts b/cloudflare-workers/oc-gateway/src/deploylease.ts index 839f2792..190b0c60 100644 --- a/cloudflare-workers/oc-gateway/src/deploylease.ts +++ b/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; diff --git a/cloudflare-workers/oc-gateway/src/index.ts b/cloudflare-workers/oc-gateway/src/index.ts index 2d420329..daa8b8f9 100644 --- a/cloudflare-workers/oc-gateway/src/index.ts +++ b/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 { 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; diff --git a/cloudflare-workers/oc-gateway/src/token.ts b/cloudflare-workers/oc-gateway/src/token.ts index 311bd99e..6be8a1d7 100644 --- a/cloudflare-workers/oc-gateway/src/token.ts +++ b/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 }; } diff --git a/cloudflare-workers/oc-gateway/test/integration.test.ts b/cloudflare-workers/oc-gateway/test/integration.test.ts index a698ab3b..1ce24ed8 100644 --- a/cloudflare-workers/oc-gateway/test/integration.test.ts +++ b/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 () => { diff --git a/cloudflare-workers/oc-gateway/test/logic.test.ts b/cloudflare-workers/oc-gateway/test/logic.test.ts index 1067d6db..c14049e2 100644 --- a/cloudflare-workers/oc-gateway/test/logic.test.ts +++ b/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()); diff --git a/cmd/oc/internal/commands/agent.go b/cmd/oc/internal/commands/agent.go index 3d95c1d8..cf230107 100644 --- a/cmd/oc/internal/commands/agent.go +++ b/cmd/oc/internal/commands/agent.go @@ -173,6 +173,7 @@ var agentCmd = &cobra.Command{ func init() { registerAgentCrud() registerAgentDeploy() + registerAgentConfig() registerAgentSchedules() rootCmd.AddCommand(sessionCmd) } diff --git a/cmd/oc/internal/commands/agent_config.go b/cmd/oc/internal/commands/agent_config.go new file mode 100644 index 00000000..2260a217 --- /dev/null +++ b/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 --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 ", + 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) +} diff --git a/cmd/oc/internal/commands/agent_config_test.go b/cmd/oc/internal/commands/agent_config_test.go new file mode 100644 index 00000000..8243ad34 --- /dev/null +++ b/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) + } +} diff --git a/cmd/oc/internal/commands/agent_deploy.go b/cmd/oc/internal/commands/agent_deploy.go index d877ca4e..d909e7dc 100644 --- a/cmd/oc/internal/commands/agent_deploy.go +++ b/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"` diff --git a/cmd/oc/internal/commands/agent_deploy_flue.go b/cmd/oc/internal/commands/agent_deploy_flue.go index e13f95d9..d0f26d16 100644 --- a/cmd/oc/internal/commands/agent_deploy_flue.go +++ b/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 { diff --git a/cmd/oc/internal/commands/agent_deploy_flue_test.go b/cmd/oc/internal/commands/agent_deploy_flue_test.go index ba0c059d..cdf56625 100644 --- a/cmd/oc/internal/commands/agent_deploy_flue_test.go +++ b/cmd/oc/internal/commands/agent_deploy_flue_test.go @@ -43,6 +43,7 @@ type fakeCP struct { mu sync.Mutex self string createBody map[string]any + configPutBody map[string]any artifactDigest string artifactSize float64 uploaded []byte @@ -63,6 +64,11 @@ func (f *fakeCP) ServeHTTP(w http.ResponseWriter, r *http.Request) { case r.Method == "POST" && r.URL.Path == "/v3/agents": _ = json.NewDecoder(r.Body).Decode(&f.createBody) writeJSON(map[string]any{"id": "agt_e2e", "name": "e2e-flue", "model": "anthropic/claude-sonnet-5", "runtime": "flue"}) + case r.Method == "PUT" && r.URL.Path == "/v3/agents/agt_e2e/config": + _ = json.NewDecoder(r.Body).Decode(&f.configPutBody) + writeJSON(map[string]any{ + "vars": f.configPutBody["vars"], "deployment_required": true, + }) case r.Method == "POST" && r.URL.Path == "/v3/agents/agt_e2e/artifacts": var body map[string]any _ = json.NewDecoder(r.Body).Decode(&body) @@ -215,6 +221,28 @@ func TestDeployFlueDoEndToEnd(t *testing.T) { if f.getCount < 2 { t.Errorf("expected the poll to observe verifying→ready (got %d GETs)", f.getCount) } + if vars, ok := f.configPutBody["vars"].(map[string]any); !ok || len(vars) != 0 { + t.Errorf("manifest without [vars] should clear desired vars, got %#v", f.configPutBody) + } +} + +func TestSyncManifestVarsReplacesDesiredVars(t *testing.T) { + f := &fakeCP{} + srv := httptest.NewServer(f) + defer srv.Close() + f.self = srv.URL + + sc := client.NewSessionsAPI(srv.URL, "test-key") + cmd := &cobra.Command{} + cmd.SetContext(context.Background()) + m := &manifest{Vars: map[string]string{"PUBLIC_MODE": "careful", "MAX_ITEMS": "12"}} + if err := syncManifestVars(cmd, sc, "agt_e2e", m); err != nil { + t.Fatalf("syncManifestVars: %v", err) + } + vars, _ := f.configPutBody["vars"].(map[string]any) + if vars["PUBLIC_MODE"] != "careful" || vars["MAX_ITEMS"] != "12" { + t.Errorf("vars PUT = %#v", f.configPutBody["vars"]) + } } func TestDeployFlueBlocksOnLeakedKey(t *testing.T) { diff --git a/sdks/flue/src/gateway.test.ts b/sdks/flue/src/gateway.test.ts new file mode 100644 index 00000000..6226dec0 --- /dev/null +++ b/sdks/flue/src/gateway.test.ts @@ -0,0 +1,25 @@ +import { describe, expect, it, vi } from "vitest"; + +const mocks = vi.hoisted(() => ({ registerProvider: vi.fn() })); + +vi.mock("@flue/runtime", () => ({ registerProvider: mocks.registerProvider })); +vi.mock("./cf-env.js", () => ({ + ocResolveEnv: () => ({ OC_GATEWAY: "https://gateway.test" }), +})); + +import { route } from "./gateway.js"; + +describe("Flue gateway binding", () => { + it("binds the deploy token once and a tokenless request cannot overwrite it", async () => { + const next = vi.fn(async () => undefined); + await route({ env: { OC_SESSION_TOKEN: "deploy-token" } } as never, next); + await route({ env: {} } as never, next); + + expect(mocks.registerProvider).toHaveBeenCalledTimes(1); + expect(mocks.registerProvider).toHaveBeenCalledWith("anthropic", { + baseUrl: "https://gateway.test/anthropic", + apiKey: "deploy-token", + }); + expect(next).toHaveBeenCalledTimes(2); + }); +}); diff --git a/sdks/flue/src/gateway.ts b/sdks/flue/src/gateway.ts index eb8c9e9a..3c21dd54 100644 --- a/sdks/flue/src/gateway.ts +++ b/sdks/flue/src/gateway.ts @@ -8,10 +8,9 @@ // The secret DOES live on the per-REQUEST env (`c.env`) — the same source ocSandbox reads at run time. // Fix: bind the apiKey at RUN scope from the exported `route` middleware, reading OC_SESSION_TOKEN from // `c.env` by DIRECT property access (NEVER spread c.env — the CF env is a proxy and spreading it throws) -// and OC_GATEWAY from the ambient snapshot, once per isolate. The token is per-DEPLOY (identical for every -// session), so the MODULE(isolate)-scoped registry holds one static value with NO cross-session race; the -// upstream per-request headers(ctx)/getApiKey(ctx) ask is only needed IF a PER-SESSION token is later -// introduced. registerProvider still takes a static apiKey / getApiKey has no turn context — fine here. +// and OC_GATEWAY from the ambient snapshot, once per isolate. The token is per-DEPLOY (identical for +// every session), so the module-scoped provider registry needs one stable binding and no per-session +// mutation. Exact attribution remains an upstream per-request headers(ctx) concern. import { registerProvider } from "@flue/runtime"; import type { AgentInitializerContext, AgentRouteHandler } from "@flue/runtime"; @@ -27,7 +26,7 @@ export const DEFAULT_MODEL = "anthropic/claude-haiku-4-5"; export interface OcEnv { /** Deployed gateway Worker base URL (set per tenant script by the OC deploy). */ OC_GATEWAY?: string; - /** Signed session/deploy JWT the gateway verifies (`{sub, org, agt, bud}`); never a raw provider key. */ + /** Signed deploy JWT the gateway verifies (`{org, agt, ep}`); never a raw provider key. */ OC_SESSION_TOKEN?: string; /** Telemetry sink for `observe()` (operator panel + spend attribution). */ OC_INGEST?: string; @@ -61,16 +60,16 @@ export function useOcGateway(ctx: AgentInitializerContext): void { bindOcProvider(ocResolveEnv(ctx.env)); } -/** Set once the run-scope apiKey bind lands (module/isolate-scoped; the token is deploy-static so one - * bind serves every co-located session — no per-session race). */ +/** Set only after the run-scope secret lands. A tokenless read can never overwrite a working + * isolate-global provider binding. */ let ocProviderBound = false; /** * The HTTP-transport opt-in every OC-hosted agent MUST export as `route` (an agent is reachable at * `/agents/:name/:id` only when its module exports `route` — flue-app.ts). ALSO the run-scope binder for - * the token seam: on each request (where the request ALS is entered and OC_SESSION_TOKEN is readable) it - * (re)binds the provider's apiKey, once per isolate, BEFORE the turn's model call runs via `next()`. A - * first `?view=updates` read that predates the token just leaves the flag false and retries next request. + * the token seam: on the first request carrying OC_SESSION_TOKEN it binds the provider's apiKey before + * the turn's model call runs via `next()`. The provider registry is isolate-global and the token is + * deploy-static, so later requests must not mutate it. * The OC dispatch Worker is the auth boundary (013 §3 B5), so the transport itself adds none. */ export const route: AgentRouteHandler = async (c, next) => { diff --git a/web/src/api/schemas.ts b/web/src/api/schemas.ts index 661890bc..16469e98 100644 --- a/web/src/api/schemas.ts +++ b/web/src/api/schemas.ts @@ -542,7 +542,9 @@ export const SessionEventSchema = z.object({ // `body` is absent/partial, `content_ref` points at the blob, `body_bytes` is the size. content_ref: z.string().nullish(), body_truncated: z.boolean().nullish(), - body_bytes: z.number().nullish(), + // PostgreSQL bigint values are serialized as strings by the sessions API. + // Coerce here so spilled event bodies remain visible in the live timeline. + body_bytes: z.coerce.number().nullish(), refs: record.nullish(), source: z.string().optional(), turn_id: z.string().nullable().optional(),