From bbc73c36bd6e4a2f857d2c0afb8320340c81bd93 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Mon, 27 Jul 2026 15:37:36 -0400 Subject: [PATCH] feat(plugins): beam local coding sessions into a read-only catalog (#112323) * feat(plugins): add read-only session beam * test(plugins): tighten Beam HTTP fixtures * docs(plugins): refresh Beam inventory counts * fix(plugins): keep Beam wire types private * fix(plugins): stabilize Beam transcript paging * fix(plugins): honor Beam Control UI base paths --- .github/labeler.yml | 6 + docs/docs.json | 1 + docs/docs_map.md | 21 ++ docs/plugins/beam.md | 147 +++++++++ docs/plugins/plugin-inventory.md | 4 +- docs/plugins/reference.md | 2 +- docs/plugins/reference/beam.md | 23 ++ extensions/beam/index.ts | 23 ++ extensions/beam/openclaw.plugin.json | 12 + extensions/beam/package.json | 15 + extensions/beam/src/beam.test.ts | 329 ++++++++++++++++++++ extensions/beam/src/http.ts | 133 ++++++++ extensions/beam/src/session-catalog.ts | 143 +++++++++ extensions/beam/src/store.ts | 25 ++ extensions/beam/src/types.ts | 167 ++++++++++ pnpm-lock.yaml | 6 + src/plugin-state/plugin-state-store.test.ts | 23 ++ 17 files changed, 1078 insertions(+), 2 deletions(-) create mode 100644 docs/plugins/beam.md create mode 100644 docs/plugins/reference/beam.md create mode 100644 extensions/beam/index.ts create mode 100644 extensions/beam/openclaw.plugin.json create mode 100644 extensions/beam/package.json create mode 100644 extensions/beam/src/beam.test.ts create mode 100644 extensions/beam/src/http.ts create mode 100644 extensions/beam/src/session-catalog.ts create mode 100644 extensions/beam/src/store.ts create mode 100644 extensions/beam/src/types.ts diff --git a/.github/labeler.yml b/.github/labeler.yml index 3a8592e5143c..4f0f83f21b6f 100644 --- a/.github/labeler.yml +++ b/.github/labeler.yml @@ -80,6 +80,12 @@ - "extensions/workboard/**" - "docs/plugins/workboard.md" - "docs/plugins/reference/workboard.md" +"plugin: beam": + - changed-files: + - any-glob-to-any-file: + - "extensions/beam/**" + - "docs/plugins/beam.md" + - "docs/plugins/reference/beam.md" "plugin: migrate-hermes": - changed-files: - any-glob-to-any-file: diff --git a/docs/docs.json b/docs/docs.json index 90ee7a01fbac..957e606b3c82 100644 --- a/docs/docs.json +++ b/docs/docs.json @@ -1316,6 +1316,7 @@ "plugins/workboard", "plugins/webhooks", "plugins/admin-http-rpc", + "plugins/beam", "plugins/voice-call", "plugins/vault", "plugins/onepassword", diff --git a/docs/docs_map.md b/docs/docs_map.md index 30b3834d37a9..a21496df4e16 100644 --- a/docs/docs_map.md +++ b/docs/docs_map.md @@ -5711,6 +5711,18 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H2: Internals and reference - H2: Related +## plugins/beam.md + +- Route: /plugins/beam +- Headings: + - H2: Enable + - H2: Authentication + - H2: Request + - H2: Storage and visibility + - H2: Security boundary + - H2: Troubleshooting + - H2: Related + ## plugins/building-extensions.md - Route: /plugins/building-extensions @@ -6309,6 +6321,15 @@ Do not edit it by hand; run `pnpm docs:map:gen`. - H2: Surface - H2: Related docs +## plugins/reference/beam.md + +- Route: /plugins/reference/beam +- Headings: + - H1: Beam plugin + - H2: Distribution + - H2: Surface + - H2: Related docs + ## plugins/reference/bonjour.md - Route: /plugins/reference/bonjour diff --git a/docs/plugins/beam.md b/docs/plugins/beam.md new file mode 100644 index 000000000000..53634800005c --- /dev/null +++ b/docs/plugins/beam.md @@ -0,0 +1,147 @@ +--- +summary: "Publish redacted local coding sessions into a shared read-only OpenClaw catalog" +read_when: + - Sharing a Claude Code or Codex session with trusted Gateway operators + - Configuring an authenticated session-ingest endpoint without connecting a node + - Auditing what Beam stores and exposes +title: "Beam plugin" +--- + +The bundled `beam` plugin receives a sanitized coding-session snapshot over authenticated HTTP and presents it in the Control UI's existing external-session catalog. The source computer sends text out; OpenClaw never connects back to that computer and receives no filesystem, terminal, tool, or node capability. + +Beam ships with OpenClaw but is disabled by default. When enabled, it registers: + +- `POST /api/v1/beam/sessions` +- the read-only **Beam** session catalog in the Control UI sidebar + +## Enable + +```bash +openclaw plugins enable beam +openclaw gateway restart +``` + +Equivalent config: + +```json5 +{ + plugins: { + entries: { + beam: { enabled: true }, + }, + }, +} +``` + +Disable the plugin when the ingest route is not needed: + +```bash +openclaw plugins disable beam +openclaw gateway restart +``` + +## Authentication + +The receiver uses normal Gateway HTTP authentication. It is not an anonymous upload endpoint. + +- With `gateway.auth.mode: "trusted-proxy"`, send requests through the configured identity-aware proxy. Beam relies on Gateway authentication but does not persist proxy identity headers as uploader attribution. +- With token or password auth, send `Authorization: Bearer `. +- Do not enable Beam with `gateway.auth.mode: "none"` unless another private ingress fully authenticates every request. + +A Cloudflare Access-protected deployment can authenticate a local CLI without exposing a GitHub token: + +```bash +cloudflared access login https://gateway.example.com +cloudflared access curl https://gateway.example.com/api/v1/beam/sessions \ + -H 'Content-Type: application/json' \ + --data-binary @sanitized-beam.json +``` + +The `beam` skill in [openclaw/agent-skills](https://github.com/openclaw/agent-skills) handles local transcript discovery, redaction, Cloudflare Access login, and upload for Claude Code and Codex. + +## Request + +```http +POST /api/v1/beam/sessions +Content-Type: application/json +``` + +```json +{ + "version": 1, + "beamId": "0123456789abcdef0123456789abcdef", + "source": "claude", + "title": "Fix the upload flow", + "updatedAt": "2026-07-20T12:00:00.000Z", + "completed": false, + "items": [ + { "type": "userMessage", "text": "Fix the upload flow." }, + { "type": "agentMessage", "text": "Implemented and tested." }, + { "type": "other", "text": "3 read, 2 write, 1 execute; raw tool outputs dropped: 4" } + ] +} +``` + +The schema is closed. Beam rejects unknown fields, invalid item types, empty text, more than 200 items, item text over 6,000 characters, non-JSON requests, and bodies over 56 KiB. + +A successful upload returns the stable Beam id and a relative Control UI URL: + +```json +{ + "ok": true, + "beamId": "0123456789abcdef0123456789abcdef", + "url": "/chat?session=catalog%3Abeam%3Agateway%3A0123456789abcdef0123456789abcdef" +} +``` + +Uploading the same `beamId` updates the existing catalog row. A completed upload sets the row status to `completed`; earlier updates display as `live`. + +## Storage and visibility + +Beam stores sanitized payloads in OpenClaw's shared SQLite-backed plugin state: + +- at most 500 sessions +- seven-day retention refreshed by each update +- oldest-entry eviction when the catalog reaches its bound +- server receipt time controls catalog ordering; clients cannot move themselves ahead with a forged timestamp + +The catalog is intentionally shared across the Gateway operator domain. Every client with `operator.read` can view every beamed session, while uploads require `operator.write` or `operator.admin`. Uploader identity is not retained, and any write-authorized operator that knows a Beam id can update that row. OpenClaw operator scopes are not tenant isolation; use a separate Gateway when sessions must be isolated between teams or machines. + +## Security boundary + +Beam is passive session publication, not remote control. + +- It has no `continueSession`, archive, terminal, tool, or node capability. +- It accepts text-only normalized transcript items, not HTML, scripts, archives, attachments, or server-fetched URLs. +- The official skill removes raw tool results, reasoning, prompts, local paths, credentials, cookies, and auth material before upload. +- The receiver still treats every transcript as untrusted text. Copying a beamed transcript into a new agent session is a separate operator action. +- Requests are rate-limited and concurrency-limited before the body is read. + +## Troubleshooting + +`404 Not Found` + +: The Beam plugin is disabled, the Gateway has not restarted since enablement, or the request is reaching another Gateway. + +`401 Unauthorized` + +: The request did not satisfy Gateway HTTP auth. Check the bearer credential or trusted-proxy/Access session. + +`405 Method Not Allowed` + +: The receiver accepts only `POST`. + +`413 Payload Too Large` + +: The serialized request exceeded 56 KiB. The official skill drops older sanitized messages until the snapshot fits. + +`429 Too Many Requests` + +: The authenticated client exceeded the bounded request or concurrency limit. Retry after the current minute window. + +## Related + +- [Control UI](/web/control-ui) +- [Operator scopes](/gateway/operator-scopes) +- [Trusted proxy auth](/gateway/trusted-proxy-auth) +- [Plugin management](/plugins/manage-plugins) diff --git a/docs/plugins/plugin-inventory.md b/docs/plugins/plugin-inventory.md index e9fe7ff3fee2..a791b751f0cc 100644 --- a/docs/plugins/plugin-inventory.md +++ b/docs/plugins/plugin-inventory.md @@ -51,7 +51,7 @@ Each entry lists the package, distribution route, and description. ## Core npm package -70 plugins +71 plugins - **[admin-http-rpc](/plugins/reference/admin-http-rpc)** (`@openclaw/admin-http-rpc`) - included in OpenClaw. OpenClaw admin HTTP RPC endpoint. @@ -61,6 +61,8 @@ Each entry lists the package, distribution route, and description. - **[azure-speech](/plugins/reference/azure-speech)** (`@openclaw/azure-speech`) - included in OpenClaw. Azure AI Speech text-to-speech (MP3, native Ogg/Opus voice notes, PCM telephony). +- **[beam](/plugins/reference/beam)** (`@openclaw/beam`) - included in OpenClaw. Read-only coding-session Beam receiver. + - **[bonjour](/plugins/reference/bonjour)** (`@openclaw/bonjour`) - included in OpenClaw. Advertise the local OpenClaw gateway over Bonjour/mDNS. - **[browser](/plugins/reference/browser)** (`@openclaw/browser-plugin`) - included in OpenClaw. Adds agent-callable tools. diff --git a/docs/plugins/reference.md b/docs/plugins/reference.md index a5c6e8709e28..b30732554ed7 100644 --- a/docs/plugins/reference.md +++ b/docs/plugins/reference.md @@ -15,5 +15,5 @@ This page is generated from `extensions/*/package.json` and pnpm plugins:inventory:gen ``` -Use [Plugin inventory](/plugins/plugin-inventory) to browse all 143 +Use [Plugin inventory](/plugins/plugin-inventory) to browse all 144 generated plugin reference pages by distribution, package, and description. diff --git a/docs/plugins/reference/beam.md b/docs/plugins/reference/beam.md new file mode 100644 index 000000000000..1e2561757d9a --- /dev/null +++ b/docs/plugins/reference/beam.md @@ -0,0 +1,23 @@ +--- +summary: "Read-only coding-session Beam receiver." +read_when: + - You are installing, configuring, or auditing the beam plugin +title: "Beam plugin" +--- + +# Beam plugin + +Read-only coding-session Beam receiver. + +## Distribution + +- Package: `@openclaw/beam` +- Install route: included in OpenClaw + +## Surface + +plugin + +## Related docs + +- [beam](/plugins/beam) diff --git a/extensions/beam/index.ts b/extensions/beam/index.ts new file mode 100644 index 000000000000..e7b3a4bcdbda --- /dev/null +++ b/extensions/beam/index.ts @@ -0,0 +1,23 @@ +import { definePluginEntry } from "openclaw/plugin-sdk/plugin-entry"; +import { createBeamRequestHandler } from "./src/http.js"; +import { createBeamSessionCatalog } from "./src/session-catalog.js"; +import { createBeamStore } from "./src/store.js"; + +export default definePluginEntry({ + id: "beam", + name: "Beam", + description: "Receive redacted local coding sessions as a read-only catalog", + register(api) { + const store = createBeamStore(api.runtime); + api.registerSessionCatalog(createBeamSessionCatalog(store)); + api.registerHttpRoute({ + path: "/api/v1/beam/sessions", + auth: "gateway", + match: "exact", + handler: createBeamRequestHandler({ + store, + resolveControlUiBasePath: () => api.runtime.config.current().gateway?.controlUi?.basePath, + }), + }); + }, +}); diff --git a/extensions/beam/openclaw.plugin.json b/extensions/beam/openclaw.plugin.json new file mode 100644 index 000000000000..8ee7038fd763 --- /dev/null +++ b/extensions/beam/openclaw.plugin.json @@ -0,0 +1,12 @@ +{ + "id": "beam", + "activation": { + "onStartup": false, + "onConfigPaths": ["plugins.entries.beam"] + }, + "configSchema": { + "type": "object", + "additionalProperties": false, + "properties": {} + } +} diff --git a/extensions/beam/package.json b/extensions/beam/package.json new file mode 100644 index 000000000000..e75f302d4c09 --- /dev/null +++ b/extensions/beam/package.json @@ -0,0 +1,15 @@ +{ + "name": "@openclaw/beam", + "version": "2026.7.2", + "private": true, + "description": "Read-only coding-session Beam receiver", + "type": "module", + "devDependencies": { + "@openclaw/plugin-sdk": "workspace:*" + }, + "openclaw": { + "extensions": [ + "./index.ts" + ] + } +} diff --git a/extensions/beam/src/beam.test.ts b/extensions/beam/src/beam.test.ts new file mode 100644 index 000000000000..192b42ef9575 --- /dev/null +++ b/extensions/beam/src/beam.test.ts @@ -0,0 +1,329 @@ +import http from "node:http"; +import { afterEach, describe, expect, it } from "vitest"; +import { createBeamRequestHandler } from "./http.js"; +import { createBeamSessionCatalog } from "./session-catalog.js"; +import type { BeamStore } from "./store.js"; +import { BEAM_MAX_BODY_BYTES, parseBeamUpload, type BeamStoredSession } from "./types.js"; + +type BeamUploadFixture = Omit; + +function sampleUpload(overrides: Record = {}): BeamUploadFixture { + return { + version: 1, + beamId: "0123456789abcdef0123456789abcdef", + source: "claude", + title: "Fix the upload flow", + updatedAt: "2026-07-20T12:00:00.000Z", + completed: false, + items: [ + { type: "userMessage", text: "Please fix the upload flow." }, + { type: "agentMessage", text: "Implemented and tested." }, + ], + ...overrides, + } as BeamUploadFixture; +} + +const writeClient = () => ({ clientIp: "127.0.0.1", scopes: ["operator.write"] }); + +function memoryStore(): BeamStore & { values: Map } { + const values = new Map(); + return { + values, + put: async (session) => { + values.set(session.beamId, session); + }, + get: async (beamId) => values.get(beamId), + list: async () => [...values.values()], + }; +} + +const servers: http.Server[] = []; +afterEach(async () => { + await Promise.all( + servers.splice(0).map( + (server) => + new Promise((resolve) => { + server.close(() => resolve()); + }), + ), + ); +}); + +async function requestStatus( + url: string, + options: { method: string; headers: Record; body: string }, +): Promise { + return await new Promise((resolve, reject) => { + const request = http.request( + url, + { method: options.method, headers: options.headers }, + (response) => { + resolve(response.statusCode); + response.resume(); + }, + ); + request.on("error", reject); + request.end(options.body); + }); +} + +async function serve(handler: ReturnType): Promise { + const server = http.createServer((req, res) => { + void handler(req, res) + .then((handled) => { + if (!handled && !res.writableEnded) { + res.statusCode = 404; + res.end("Not Found"); + } + }) + .catch((error: unknown) => { + if (!res.writableEnded) { + res.statusCode = 500; + res.end(error instanceof Error ? error.message : "test handler failed"); + } + }); + }); + servers.push(server); + await new Promise((resolve) => { + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + if (!address || typeof address === "string") { + throw new Error("test server did not bind a TCP port"); + } + return `http://127.0.0.1:${address.port}/api/v1/beam/sessions`; +} + +describe("Beam payload validation", () => { + it("accepts the closed normalized payload", () => { + const result = parseBeamUpload(sampleUpload()); + expect(result).toEqual({ ok: true, value: sampleUpload() }); + }); + + it("accepts timezone-bearing ISO timestamps with four-digit low years", () => { + expect(parseBeamUpload(sampleUpload({ updatedAt: "0099-01-01T00:00:00Z" }))).toEqual({ + ok: true, + value: sampleUpload({ updatedAt: "0099-01-01T00:00:00Z" }), + }); + }); + + it("rejects unknown fields, non-ISO timestamps, and oversized transcript entries", () => { + expect(parseBeamUpload(sampleUpload({ arbitrary: "junk" }))).toEqual({ + ok: false, + error: "request body must be a closed Beam object", + }); + for (const updatedAt of [ + "1", + "2026/07/20", + "2026-07-20T12:00:00", + "2026-02-30T12:00:00Z", + "2026-04-31T00:00:00Z", + ]) { + expect(parseBeamUpload(sampleUpload({ updatedAt }))).toEqual({ + ok: false, + error: "updatedAt must be an ISO timestamp", + }); + } + expect( + parseBeamUpload(sampleUpload({ items: [{ type: "userMessage", text: "x".repeat(6_001) }] })), + ).toEqual({ + ok: false, + error: "transcript item text must be 1-6000 characters", + }); + }); +}); + +describe("Beam receiver", () => { + it("stores authenticated uploads and preserves creation time across updates", async () => { + const store = memoryStore(); + let now = 100; + const endpoint = await serve( + createBeamRequestHandler({ store, now: () => now, resolveClient: writeClient }), + ); + const first = await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(sampleUpload()), + }); + expect(first.status).toBe(200); + expect(await first.json()).toEqual({ + ok: true, + beamId: "0123456789abcdef0123456789abcdef", + url: "/chat?session=catalog%3Abeam%3Agateway%3A0123456789abcdef0123456789abcdef", + }); + expect(store.values.get("0123456789abcdef0123456789abcdef")).toMatchObject({ + createdAt: 100, + receivedAt: 100, + }); + + now = 200; + await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(sampleUpload({ completed: true })), + }); + expect(store.values.get("0123456789abcdef0123456789abcdef")).toMatchObject({ + createdAt: 100, + receivedAt: 200, + completed: true, + }); + }); + + it("returns a catalog URL beneath the configured Control UI base path", async () => { + const store = memoryStore(); + const endpoint = await serve( + createBeamRequestHandler({ + store, + resolveClient: writeClient, + resolveControlUiBasePath: () => "/openclaw/", + }), + ); + const response = await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(sampleUpload()), + }); + + expect(response.status).toBe(200); + expect(await response.json()).toEqual({ + ok: true, + beamId: "0123456789abcdef0123456789abcdef", + url: "/openclaw/chat?session=catalog%3Abeam%3Agateway%3A0123456789abcdef0123456789abcdef", + }); + }); + + it("requires operator.write before reading the upload body", async () => { + const store = memoryStore(); + const endpoint = await serve( + createBeamRequestHandler({ + store, + resolveClient: () => ({ clientIp: "127.0.0.1", scopes: ["operator.read"] }), + }), + ); + const response = await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify(sampleUpload()), + }); + + expect(response.status).toBe(403); + expect(await response.json()).toEqual({ ok: false, error: "operator.write is required" }); + expect(store.values.size).toBe(0); + }); + + it("rejects method, media type, malformed JSON, and oversized bodies", async () => { + const store = memoryStore(); + const endpoint = await serve(createBeamRequestHandler({ store, resolveClient: writeClient })); + expect((await fetch(endpoint)).status).toBe(405); + expect( + ( + await fetch(endpoint, { + method: "POST", + headers: { "content-type": "text/plain" }, + body: "{}", + }) + ).status, + ).toBe(415); + expect( + ( + await fetch(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: "{", + }) + ).status, + ).toBe(400); + expect( + await requestStatus(endpoint, { + method: "POST", + headers: { "content-type": "application/json" }, + body: JSON.stringify({ padding: "x".repeat(BEAM_MAX_BODY_BYTES) }), + }), + ).toBe(413); + expect(store.values.size).toBe(0); + }); +}); + +describe("Beam session catalog", () => { + it("lists newest sessions and reads paginated transcript items without mutation capabilities", async () => { + const store = memoryStore(); + await store.put({ + ...sampleUpload({ truncated: true }), + createdAt: 100, + receivedAt: 200, + }); + await store.put({ + ...sampleUpload({ + beamId: "fedcba9876543210fedcba9876543210", + title: "Older Codex session", + source: "codex", + completed: true, + }), + createdAt: 50, + receivedAt: 100, + }); + const catalog = createBeamSessionCatalog(store); + + const [host] = await catalog.list({ limitPerHost: 1 }); + expect(host).toBeDefined(); + if (!host) { + throw new Error("Beam catalog did not return its gateway host"); + } + expect(host.sessions).toHaveLength(1); + expect(host.sessions[0]).toMatchObject({ + threadId: "0123456789abcdef0123456789abcdef", + status: "live", + source: "claude", + canContinue: false, + canArchive: false, + }); + expect(host.nextCursor).toBe("1"); + + const transcript = await catalog.read({ + hostId: "gateway", + threadId: "0123456789abcdef0123456789abcdef", + limit: 1, + }); + expect(transcript.items).toEqual([ + expect.objectContaining({ type: "agentMessage", text: "Implemented and tested." }), + ]); + expect(transcript.items[0]).not.toHaveProperty("truncated"); + expect(transcript.nextCursor).toEqual(expect.any(String)); + + const older = await catalog.read({ + hostId: "gateway", + threadId: "0123456789abcdef0123456789abcdef", + limit: 1, + cursor: transcript.nextCursor, + }); + expect(older.items).toEqual([ + expect.objectContaining({ type: "userMessage", text: "Please fix the upload flow." }), + ]); + expect(older.nextCursor).toBeUndefined(); + + const current = store.values.get("0123456789abcdef0123456789abcdef"); + if (!current) { + throw new Error("Beam test store lost the current session"); + } + await store.put({ + ...current, + items: [ + ...current.items.slice(1), + { type: "agentMessage", text: "Appended after first page." }, + ], + receivedAt: 200, + }); + + await expect( + catalog.read({ + hostId: "gateway", + threadId: "0123456789abcdef0123456789abcdef", + limit: 1, + cursor: transcript.nextCursor, + }), + ).rejects.toThrow("stale Beam transcript cursor"); + expect(catalog.continueSession).toBeUndefined(); + expect(catalog.archive).toBeUndefined(); + expect(catalog.openTerminal).toBeUndefined(); + }); +}); diff --git a/extensions/beam/src/http.ts b/extensions/beam/src/http.ts new file mode 100644 index 000000000000..6fdaa5256183 --- /dev/null +++ b/extensions/beam/src/http.ts @@ -0,0 +1,133 @@ +import type { IncomingMessage, ServerResponse } from "node:http"; +import { getPluginRuntimeGatewayRequestScope } from "openclaw/plugin-sdk/plugin-runtime"; +import { + beginWebhookRequestPipelineOrReject, + createFixedWindowRateLimiter, + createWebhookInFlightLimiter, + readJsonWebhookBodyOrReject, +} from "openclaw/plugin-sdk/webhook-ingress"; +import type { BeamStore } from "./store.js"; +import { BEAM_HOST_ID, BEAM_MAX_BODY_BYTES, parseBeamUpload } from "./types.js"; + +function sendJson(res: ServerResponse, status: number, value: unknown): void { + res.statusCode = status; + res.setHeader("Cache-Control", "no-store"); + res.setHeader("Content-Type", "application/json; charset=utf-8"); + res.end(JSON.stringify(value)); +} + +function firstHeader(req: IncomingMessage, name: string): string | undefined { + const value = req.headers[name]; + return (Array.isArray(value) ? value[0] : value)?.trim() || undefined; +} + +type BeamRequestClient = { + clientIp: string; + scopes: readonly string[]; +}; + +function currentRequestClient(req: IncomingMessage): BeamRequestClient { + const client = getPluginRuntimeGatewayRequestScope()?.client; + return { + clientIp: client?.clientIp ?? req.socket.remoteAddress ?? "unknown", + scopes: client?.connect?.scopes ?? [], + }; +} + +function canPublish(scopes: readonly string[]): boolean { + return scopes.includes("operator.write") || scopes.includes("operator.admin"); +} + +function normalizeControlUiBasePath(value: unknown): string { + if (typeof value !== "string") { + return ""; + } + const trimmed = value.trim(); + if (!trimmed || trimmed === "/") { + return ""; + } + const withLeadingSlash = trimmed.startsWith("/") ? trimmed : `/${trimmed}`; + return withLeadingSlash.replace(/\/+$/, ""); +} + +function catalogSessionUrl(beamId: string, basePath: unknown): string { + const sessionKey = `catalog:beam:${BEAM_HOST_ID}:${beamId}`; + return `${normalizeControlUiBasePath(basePath)}/chat?session=${encodeURIComponent(sessionKey)}`; +} + +export function createBeamRequestHandler(params: { + store: BeamStore; + now?: () => number; + resolveClient?: (req: IncomingMessage) => BeamRequestClient; + resolveControlUiBasePath?: () => unknown; +}): (req: IncomingMessage, res: ServerResponse) => Promise { + const rateLimiter = createFixedWindowRateLimiter({ + windowMs: 60_000, + maxRequests: 60, + maxTrackedKeys: 2_048, + }); + const inFlightLimiter = createWebhookInFlightLimiter({ + maxInFlightPerKey: 2, + maxTrackedKeys: 2_048, + }); + + return async (req, res) => { + const client = params.resolveClient?.(req) ?? currentRequestClient(req); + if (!canPublish(client.scopes)) { + sendJson(res, 403, { ok: false, error: "operator.write is required" }); + return true; + } + const pipeline = beginWebhookRequestPipelineOrReject({ + req, + res, + allowMethods: ["POST"], + requireJsonContentType: true, + rateLimiter, + rateLimitKey: client.clientIp, + inFlightLimiter, + inFlightKey: client.clientIp, + }); + if (!pipeline.ok) { + return true; + } + + try { + const contentLength = Number(firstHeader(req, "content-length")); + if (Number.isFinite(contentLength) && contentLength > BEAM_MAX_BODY_BYTES) { + sendJson(res, 413, { ok: false, error: "Payload Too Large" }); + return true; + } + const body = await readJsonWebhookBodyOrReject({ + req, + res, + maxBytes: BEAM_MAX_BODY_BYTES, + timeoutMs: 10_000, + emptyObjectOnEmpty: false, + invalidJsonMessage: "invalid Beam request body", + }); + if (!body.ok) { + return true; + } + const parsed = parseBeamUpload(body.value); + if (!parsed.ok) { + sendJson(res, 400, { ok: false, error: parsed.error }); + return true; + } + const receivedAt = params.now?.() ?? Date.now(); + const existing = await params.store.get(parsed.value.beamId); + await params.store.put({ + ...parsed.value, + createdAt: existing?.createdAt ?? receivedAt, + receivedAt, + }); + sendJson(res, 200, { + ok: true, + beamId: parsed.value.beamId, + url: catalogSessionUrl(parsed.value.beamId, params.resolveControlUiBasePath?.()), + }); + return true; + } finally { + pipeline.release(); + } + }; +} diff --git a/extensions/beam/src/session-catalog.ts b/extensions/beam/src/session-catalog.ts new file mode 100644 index 000000000000..6bf78fb57489 --- /dev/null +++ b/extensions/beam/src/session-catalog.ts @@ -0,0 +1,143 @@ +import { createHash } from "node:crypto"; +import type { + SessionCatalogProvider, + SessionCatalogTranscriptItem, +} from "openclaw/plugin-sdk/session-catalog"; +import type { BeamStore } from "./store.js"; +import { BEAM_HOST_ID, type BeamStoredSession } from "./types.js"; + +const DEFAULT_LIMIT = 50; +const MAX_LIMIT = 100; + +function boundedLimit(value: number | undefined): number { + return Math.min(MAX_LIMIT, Math.max(1, value ?? DEFAULT_LIMIT)); +} + +function cursorOffset(value: string | undefined): number { + if (!value || !/^\d+$/.test(value)) { + return 0; + } + const parsed = Number(value); + return Number.isSafeInteger(parsed) && parsed >= 0 ? parsed : 0; +} + +function searchableText(session: BeamStoredSession): string { + return `${session.title}\n${session.source}`.toLowerCase(); +} + +function transcriptItems(session: BeamStoredSession): SessionCatalogTranscriptItem[] { + return session.items.map((item, index) => ({ + id: `${session.beamId}:${index}`, + type: item.type, + text: item.text, + timestamp: session.updatedAt, + })); +} + +type TranscriptCursor = { revision: string; end: number }; + +function transcriptRevision(session: BeamStoredSession): string { + return createHash("sha256").update(JSON.stringify(session.items)).digest("base64url"); +} + +function encodeTranscriptCursor(cursor: TranscriptCursor): string { + return Buffer.from(JSON.stringify(cursor), "utf8").toString("base64url"); +} + +function decodeTranscriptCursor(value: string): TranscriptCursor { + try { + const parsed = JSON.parse(Buffer.from(value, "base64url").toString("utf8")) as unknown; + if ( + parsed && + typeof parsed === "object" && + !Array.isArray(parsed) && + typeof (parsed as TranscriptCursor).revision === "string" && + /^[A-Za-z0-9_-]{43}$/.test((parsed as TranscriptCursor).revision) && + typeof (parsed as TranscriptCursor).end === "number" && + Number.isSafeInteger((parsed as TranscriptCursor).end) && + (parsed as TranscriptCursor).end >= 0 + ) { + return parsed as TranscriptCursor; + } + } catch { + // Reject malformed cursors below. + } + throw new Error("invalid Beam transcript cursor"); +} + +function transcriptPage( + items: SessionCatalogTranscriptItem[], + limit: number, + revision: string, + cursor?: TranscriptCursor, +): { items: SessionCatalogTranscriptItem[]; nextCursor?: string } { + if (cursor && cursor.revision !== revision) { + throw new Error("stale Beam transcript cursor"); + } + const end = Math.min(items.length, Math.max(0, cursor?.end ?? items.length)); + const start = Math.max(0, end - limit); + return { + items: items.slice(start, end), + ...(start > 0 ? { nextCursor: encodeTranscriptCursor({ revision, end: start }) } : {}), + }; +} + +export function createBeamSessionCatalog(store: BeamStore): SessionCatalogProvider { + return { + id: "beam", + label: "Beam", + async list(params) { + const search = params.search?.trim().toLowerCase(); + const sessions = (await store.list()) + .filter((session) => !search || searchableText(session).includes(search)) + .toSorted((left, right) => right.receivedAt - left.receivedAt); + const offset = cursorOffset(params.cursors?.[BEAM_HOST_ID]); + const limit = boundedLimit(params.limitPerHost); + const page = sessions.slice(offset, offset + limit); + return [ + { + hostId: BEAM_HOST_ID, + label: "Beamed sessions", + kind: "gateway", + connected: true, + sessions: page.map((session) => ({ + threadId: session.beamId, + name: session.title, + status: session.completed ? "completed" : "live", + createdAt: session.createdAt, + updatedAt: session.receivedAt, + recencyAt: session.receivedAt, + source: session.source, + archived: false, + canContinue: false, + canArchive: false, + })), + ...(offset + page.length < sessions.length + ? { nextCursor: String(offset + page.length) } + : {}), + }, + ]; + }, + async read(params) { + if (params.hostId !== BEAM_HOST_ID) { + throw new Error(`unknown Beam host: ${params.hostId}`); + } + const session = await store.get(params.threadId); + if (!session) { + throw new Error(`unknown Beam session: ${params.threadId}`); + } + const page = transcriptPage( + transcriptItems(session), + boundedLimit(params.limit), + transcriptRevision(session), + params.cursor === undefined ? undefined : decodeTranscriptCursor(params.cursor), + ); + return { + hostId: BEAM_HOST_ID, + label: session.title, + threadId: session.beamId, + ...page, + }; + }, + }; +} diff --git a/extensions/beam/src/store.ts b/extensions/beam/src/store.ts new file mode 100644 index 000000000000..c0144bdbd95f --- /dev/null +++ b/extensions/beam/src/store.ts @@ -0,0 +1,25 @@ +import type { PluginRuntime } from "openclaw/plugin-sdk/plugin-runtime"; +import type { BeamStoredSession } from "./types.js"; +import { BEAM_MAX_SESSIONS, BEAM_RETENTION_MS } from "./types.js"; + +export type BeamStore = { + put: (session: BeamStoredSession) => Promise; + get: (beamId: string) => Promise; + list: () => Promise; +}; + +export function createBeamStore(runtime: PluginRuntime): BeamStore { + const store = runtime.state.openKeyedStore({ + namespace: "sessions", + maxEntries: BEAM_MAX_SESSIONS, + overflowPolicy: "evict-oldest", + defaultTtlMs: BEAM_RETENTION_MS, + }); + return { + put: async (session) => { + await store.register(session.beamId, session); + }, + get: (beamId) => store.lookup(beamId), + list: async () => (await store.entries()).map((entry) => entry.value), + }; +} diff --git a/extensions/beam/src/types.ts b/extensions/beam/src/types.ts new file mode 100644 index 000000000000..7fb415d15ba6 --- /dev/null +++ b/extensions/beam/src/types.ts @@ -0,0 +1,167 @@ +import { isRecord } from "openclaw/plugin-sdk/string-coerce-runtime"; + +export const BEAM_HOST_ID = "gateway"; +export const BEAM_MAX_BODY_BYTES = 56 * 1024; +export const BEAM_RETENTION_MS = 7 * 24 * 60 * 60 * 1000; +export const BEAM_MAX_SESSIONS = 500; +const BEAM_MAX_ITEMS = 200; +const BEAM_MAX_ITEM_CHARS = 6_000; + +type BeamTranscriptItem = { + type: "userMessage" | "agentMessage" | "other"; + text: string; +}; + +type BeamUpload = { + version: 1; + beamId: string; + source: string; + title: string; + updatedAt: string; + completed: boolean; + truncated?: boolean; + hookEvent?: string; + items: BeamTranscriptItem[]; +}; + +export type BeamStoredSession = BeamUpload & { + createdAt: number; + receivedAt: number; +}; + +const TOP_LEVEL_KEYS = new Set([ + "version", + "beamId", + "source", + "title", + "updatedAt", + "completed", + "truncated", + "hookEvent", + "items", +]); +const ITEM_KEYS = new Set(["type", "text"]); +const ITEM_TYPES = new Set(["userMessage", "agentMessage", "other"]); + +function hasOnlyKeys(value: Record, allowed: Set): boolean { + return Object.keys(value).every((key) => allowed.has(key)); +} + +function optionalString(value: unknown, maxLength: number): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + return trimmed && trimmed.length <= maxLength ? trimmed : undefined; +} + +function isIsoTimestamp(value: string): boolean { + const match = + /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2})(?:\.\d{1,9})?(?:Z|([+-])(\d{2}):(\d{2}))$/.exec( + value, + ); + if (!match || !Number.isFinite(Date.parse(value))) { + return false; + } + const year = Number(match[1]); + const month = Number(match[2]); + const day = Number(match[3]); + const hour = Number(match[4]); + const minute = Number(match[5]); + const second = Number(match[6]); + const offsetHour = Number(match[8] ?? 0); + const offsetMinute = Number(match[9] ?? 0); + if (hour > 23 || minute > 59 || second > 59 || offsetHour > 23 || offsetMinute > 59) { + return false; + } + const calendar = new Date(0); + calendar.setUTCFullYear(year, month - 1, day); + calendar.setUTCHours(hour, minute, second, 0); + return ( + calendar.getUTCFullYear() === year && + calendar.getUTCMonth() === month - 1 && + calendar.getUTCDate() === day && + calendar.getUTCHours() === hour && + calendar.getUTCMinutes() === minute && + calendar.getUTCSeconds() === second + ); +} + +export function parseBeamUpload( + value: unknown, +): { ok: true; value: BeamUpload } | { ok: false; error: string } { + if (!isRecord(value) || !hasOnlyKeys(value, TOP_LEVEL_KEYS)) { + return { ok: false, error: "request body must be a closed Beam object" }; + } + if (value.version !== 1) { + return { ok: false, error: "version must be 1" }; + } + const beamId = optionalString(value.beamId, 64); + if (!beamId || !/^[a-f0-9]{32}$/i.test(beamId)) { + return { ok: false, error: "beamId must be a 32-character hex id" }; + } + const source = optionalString(value.source, 32); + if (!source || !/^[a-z0-9._-]+$/i.test(source)) { + return { ok: false, error: "source must be a short identifier" }; + } + const title = optionalString(value.title, 160); + if (!title) { + return { ok: false, error: "title must be a non-empty string" }; + } + const updatedAt = optionalString(value.updatedAt, 64); + if (!updatedAt || !isIsoTimestamp(updatedAt)) { + return { ok: false, error: "updatedAt must be an ISO timestamp" }; + } + if (typeof value.completed !== "boolean") { + return { ok: false, error: "completed must be a boolean" }; + } + if (value.truncated !== undefined && typeof value.truncated !== "boolean") { + return { ok: false, error: "truncated must be a boolean" }; + } + const hookEvent = value.hookEvent === undefined ? undefined : optionalString(value.hookEvent, 64); + if (value.hookEvent !== undefined && !hookEvent) { + return { ok: false, error: "hookEvent must be a short string" }; + } + if ( + !Array.isArray(value.items) || + value.items.length === 0 || + value.items.length > BEAM_MAX_ITEMS + ) { + return { ok: false, error: `items must contain 1-${BEAM_MAX_ITEMS} entries` }; + } + const items: BeamTranscriptItem[] = []; + for (const rawItem of value.items) { + if (!isRecord(rawItem) || !hasOnlyKeys(rawItem, ITEM_KEYS)) { + return { ok: false, error: "each transcript item must be a closed object" }; + } + if ( + typeof rawItem.type !== "string" || + !ITEM_TYPES.has(rawItem.type as BeamTranscriptItem["type"]) + ) { + return { ok: false, error: "transcript item type is invalid" }; + } + const text = optionalString(rawItem.text, BEAM_MAX_ITEM_CHARS); + if (!text) { + return { + ok: false, + error: `transcript item text must be 1-${BEAM_MAX_ITEM_CHARS} characters`, + }; + } + items.push({ type: rawItem.type as BeamTranscriptItem["type"], text }); + } + + return { + ok: true, + value: { + version: 1, + beamId: beamId.toLowerCase(), + source: source.toLowerCase(), + title, + updatedAt, + completed: value.completed, + ...(value.truncated === true ? { truncated: true } : {}), + ...(hookEvent ? { hookEvent } : {}), + items, + }, + }; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index e4675fd3e16d..a59d06d1f080 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -481,6 +481,12 @@ importers: specifier: workspace:* version: link:../../packages/plugin-sdk + extensions/beam: + devDependencies: + '@openclaw/plugin-sdk': + specifier: workspace:* + version: link:../../packages/plugin-sdk + extensions/bonjour: dependencies: '@homebridge/ciao': diff --git a/src/plugin-state/plugin-state-store.test.ts b/src/plugin-state/plugin-state-store.test.ts index e2c43591bd20..09ad1f6166ac 100644 --- a/src/plugin-state/plugin-state-store.test.ts +++ b/src/plugin-state/plugin-state-store.test.ts @@ -200,6 +200,29 @@ describe("plugin state keyed store", () => { }); }); + it("refreshes the default TTL when register upserts an existing key", async () => { + await withPluginStateTestState(async () => { + vi.useFakeTimers(); + const store = createPluginStateKeyedStore<{ version: number }>("beam", { + namespace: "sessions", + maxEntries: 10, + defaultTtlMs: 1_000, + }); + vi.setSystemTime(1_000); + await store.register("session", { version: 1 }); + vi.setSystemTime(1_500); + await store.register("session", { version: 2 }); + + await expect(store.entries()).resolves.toEqual([ + { key: "session", value: { version: 2 }, createdAt: 1_500, expiresAt: 2_500 }, + ]); + vi.setSystemTime(2_100); + await expect(store.lookup("session")).resolves.toEqual({ version: 2 }); + vi.setSystemTime(2_501); + await expect(store.lookup("session")).resolves.toBeUndefined(); + }); + }); + it("registerIfAbsent inserts the first value and preserves live duplicates", async () => { await withPluginStateTestState(async () => { vi.useFakeTimers();