diff --git a/CHANGELOG.md b/CHANGELOG.md index ee6ccef1..d3fe7f8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -28,6 +28,7 @@ Packages without a separate changelog are covered by the cross-package notes bel - Rate limit `GET /v1/workspace` in its own bucket so credential validation and launch preflight stay reachable while the workspace's other traffic is at its ceiling. - Put a retry contract on every 429: `Retry-After` and `X-RateLimit-Reset` now distinguish a per-minute throttle that clears in seconds from a plan quota that only clears at the period boundary. - Add opt-in durable task actions with execution fencing, final-result receipts, and restart reconciliation. +- Keep a node's enrollment-time `cloud:*` tags when its broker re-registers, and stop brokers from adding or removing them. `cloud:` is now a reserved node-tag namespace: `cloud:*` tags in `node.register` are ignored with a server-side warning, and re-enrolling is how to change or clear them. - `GET /v1/inbox`: unread counts no longer include archived channels, and mentions are matched with the exact `@handle` token contract (escaped `\@x`, email addresses, and prefix/superstring text are not mentions); mention results are limited to live channels the agent has joined or DMs the agent participates in. ## [8.10.1] - 2026-09-13 @@ -559,6 +560,9 @@ Packages without a separate changelog are covered by the cross-package notes bel Earlier releases are available on the [GitHub releases page](https://github.com/AgentWorkforce/relaycast/releases). +[Unreleased - Minor]: https://github.com/AgentWorkforce/relaycast/compare/v8.10.1...HEAD +[Unreleased - Patch]: https://github.com/AgentWorkforce/relaycast/compare/v8.10.1...HEAD +[Unreleased]: https://github.com/AgentWorkforce/relaycast/compare/v8.11.0...HEAD [Unreleased]: https://github.com/AgentWorkforce/relaycast/compare/v8.11.1...HEAD [8.11.1]: https://github.com/AgentWorkforce/relaycast/compare/v8.11.0...v8.11.1 [8.11.0]: https://github.com/AgentWorkforce/relaycast/compare/v8.10.1...v8.11.0 diff --git a/README.md b/README.md index 57815765..16cb7e6e 100644 --- a/README.md +++ b/README.md @@ -739,6 +739,14 @@ always round-trip untouched. Registrations that omit `repo_keys` entirely are pre-`repo_keys` clients and keep their legacy `repo:` tags, so brokers that support the field should always send it — `[]` included — rather than omitting it. +The `cloud:` tag namespace is reserved for the control plane's lifecycle +identity (sandbox provider, node type, sandbox id, route). Only enrollment +(`POST /v1/nodes`) sets or clears `cloud:*` tags. A broker `node.register` +keeps the node's current `cloud:*` tags, and any `cloud:*` entry in the frame's +own `tags` is ignored and logged server-side rather than stored; the +registration itself still succeeds. To change or remove a `cloud:*` tag, +re-enroll the node. + Nodes are first-class delivery hosts and every agent has a node route. `kind` describes transport (`ws`, `http_push`, or `poll`), `role` describes ownership (`direct` node-of-one or `broker` node-of-many), and `delivery_adapter` diff --git a/openapi.yaml b/openapi.yaml index accef257..44e9e15b 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1216,7 +1216,11 @@ components: that field is the only source of these entries and any `repo:` tag in the register message's own `tags` is dropped. A register message that omits `repo_keys` is a pre-`repo_keys` client, and its - validated `repo:` tags are preserved as-is. + validated `repo:` tags are preserved as-is. `cloud:*` entries are + reserved for control-plane lifecycle identity and are set or + cleared only by enrollment (`POST /v1/nodes`); a broker + `node.register` keeps the node's current `cloud:*` tags and + ignores (and logs) any `cloud:*` tag in its own `tags`. version: type: string status: diff --git a/packages/engine/CHANGELOG.md b/packages/engine/CHANGELOG.md index 985f4b13..4c8b5c4f 100644 --- a/packages/engine/CHANGELOG.md +++ b/packages/engine/CHANGELOG.md @@ -18,6 +18,7 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht - Add optional `EntitlementsProvider.getUsageResetAt()` so a billing-backed provider can report its own period boundary for `Retry-After`; omitting it falls back to the engine's UTC-month period. - `RateLimiter.check` no longer counts a rejected request against its bucket, so a throttled caller's retries cannot inflate the count or hold the window open. `KeyValueStore.increment` takes an optional `ttlSeconds` applied when the key is created. - Persist task ownership and immutable final results with attempt/generation fencing, deadline failure, and replayable fleet receipts; migration 0060 adds task state and action execution mode. +- Preserve server-owned `cloud:*` node tags set at enrollment when a broker node re-registers, and ignore `cloud:*` tags sent in `node.register` (logged as `[node.register] ignored server-owned tags`); only enrollment sets or clears them. The merge happens inside the register UPDATE, so a re-enroll that lands mid-registration is not reverted. - `GET /v1/inbox`: unread counts no longer include archived channels, and mentions are matched with the exact `@handle` token contract (escaped `\@x`, email addresses, and prefix/superstring text are not mentions); mention results are limited to live channels the agent has joined or DMs the agent participates in. Hosts must apply `0061_messages_workspace_length_id_index.sql` so mention keyset batches use `(workspace_id, length(id), id)`. ## [8.10.1] - 2026-09-13 diff --git a/packages/engine/src/__tests__/conformance/nodeServerOwnedTags.test.ts b/packages/engine/src/__tests__/conformance/nodeServerOwnedTags.test.ts new file mode 100644 index 00000000..c63022a5 --- /dev/null +++ b/packages/engine/src/__tests__/conformance/nodeServerOwnedTags.test.ts @@ -0,0 +1,241 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { registerNode } from '../../engine/node.js'; +import { DEFAULT_PROVIDER_NAME } from '../../engine/nodeProvider.js'; +import type { EngineDb } from '../../ports/database.js'; +import { createWorkspace, FakeSocket, makeNodeStack, type TestStack } from './harness.js'; + +// `cloud:*` tags are written by the control plane at enrollment and are the +// node's lifecycle identity. A broker re-registering over the node socket +// never learns them, so registration must keep them, and must not let the +// register frame add, change or remove them. +const ENROLLED_CLOUD_TAGS = [ + 'cloud:sandbox-provider:daytona', + 'cloud:node-type:daytona-jit', + 'cloud:sandbox-id:sbx_ledger_1', +]; + +describe('fleet node server-owned cloud:* tags', () => { + let stack: TestStack; + beforeEach(() => { stack = makeNodeStack(); }); + afterEach(() => stack.close()); + + async function enroll(workspaceKey: string, body: Record) { + const response = await stack.app.request('/v1/nodes', { + method: 'POST', + headers: { 'content-type': 'application/json', authorization: `Bearer ${workspaceKey}` }, + body: JSON.stringify({ node_id: 'node_sandbox', name: 'sandbox-node', version: 'test-node', ...body }), + }); + expect(response.ok).toBe(true); + } + + async function readTags(workspaceKey: string): Promise { + const response = await stack.app.request('/v1/nodes?name=sandbox-node', { + headers: { authorization: `Bearer ${workspaceKey}` }, + }); + expect(response.status).toBe(200); + const body = await response.json() as { data: Array<{ tags: string[] }> }; + return body.data[0].tags; + } + + async function register(workspaceId: string, id: string, fields: Record) { + const socket = new FakeSocket(); + const handle = stack.runtime.realtime.attachNodeSocket(workspaceId, 'node_sandbox', socket); + await handle.handleMessage(JSON.stringify({ + v: 1, + id, + type: 'node.register', + name: 'sandbox-node', + node_id: 'node_sandbox', + capabilities: [], + max_agents: 4, + tags: [], + version: 'test-node', + resume_cursor: null, + ...fields, + })); + const reply = socket.ofType('reply').at(-1) as { ok: boolean; data: { tags: string[] } }; + expect(reply.ok).toBe(true); + await handle.handleClose(); + return reply.data.tags; + } + + it('keeps enrolled cloud:* tags when a broker re-registers with no tags', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-empty'); + await enroll(workspace.workspaceKey, { max_agents: 4, tags: [...ENROLLED_CLOUD_TAGS, 'enrolled'] }); + + // The broker sends `tags: []` because it never learned the enrollment tags. + // Non-server-owned tags keep their existing semantics (the register value + // wins, so `enrolled` goes); the cloud identity survives. + expect(await register(workspace.workspaceId, 'register-empty', { tags: [] })).toEqual(ENROLLED_CLOUD_TAGS); + expect(await readTags(workspace.workspaceKey)).toEqual(ENROLLED_CLOUD_TAGS); + + // And again on a later reconnect. + expect(await register(workspace.workspaceId, 'register-empty-2', { tags: [] })).toEqual(ENROLLED_CLOUD_TAGS); + expect(await readTags(workspace.workspaceKey)).toEqual(ENROLLED_CLOUD_TAGS); + }); + + it('ignores cloud:* tags in the register frame, so a broker cannot forge or change them', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-forged'); + await enroll(workspace.workspaceKey, { max_agents: 4, tags: ENROLLED_CLOUD_TAGS }); + + const tags = await register(workspace.workspaceId, 'register-forged', { + tags: [ + 'linux', + 'cloud:sandbox-id:sbx_someone_else', + 'cloud:node-type:persistent', + 'cloud:relaycast-route:forged', + ], + }); + + expect(tags).toEqual([...ENROLLED_CLOUD_TAGS, 'linux']); + expect(await readTags(workspace.workspaceKey)).toEqual([...ENROLLED_CLOUD_TAGS, 'linux']); + }); + + it('does not let a register frame add cloud:* tags to a node enrolled without them', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-none'); + await enroll(workspace.workspaceKey, { max_agents: 4, tags: ['enrolled'] }); + + const tags = await register(workspace.workspaceId, 'register-add-cloud', { + tags: ['linux', 'cloud:sandbox-id:sbx_forged'], + }); + + expect(tags).toEqual(['linux']); + expect(await readTags(workspace.workspaceKey)).toEqual(['linux']); + }); + + it('keeps non-cloud:* tags replaced by each register, alongside repo_keys', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-other'); + await enroll(workspace.workspaceKey, { max_agents: 4, tags: ENROLLED_CLOUD_TAGS }); + + expect(await register(workspace.workspaceId, 'register-other-1', { + // `cloudy` and `cloud-region:eu` share a stem with the reserved prefix but + // are not `cloud:` tags, so they are ordinary broker tags. + tags: ['linux', 'gpu', 'cloudy', 'cloud-region:eu', 'repo:forged/x'], + repo_keys: ['AgentWorkforce/relaycast'], + })).toEqual([ + ...ENROLLED_CLOUD_TAGS, + 'linux', + 'gpu', + 'cloudy', + 'cloud-region:eu', + 'repo:AgentWorkforce/relaycast', + ]); + + // A later register replaces the broker's tags wholesale, as before. + expect(await register(workspace.workspaceId, 'register-other-2', { + tags: ['linux'], + repo_keys: [], + })).toEqual([...ENROLLED_CLOUD_TAGS, 'linux']); + expect(await readTags(workspace.workspaceKey)).toEqual([...ENROLLED_CLOUD_TAGS, 'linux']); + }); + + it('leaves enrollment as the authority: a re-enroll changes cloud:* tags and register keeps the new set', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-reenroll'); + await enroll(workspace.workspaceKey, { max_agents: 4, tags: ENROLLED_CLOUD_TAGS }); + await register(workspace.workspaceId, 'register-before-reenroll', { tags: [] }); + + const reEnrolled = ['cloud:sandbox-provider:daytona', 'cloud:sandbox-id:sbx_ledger_2']; + await enroll(workspace.workspaceKey, { max_agents: 4, tags: reEnrolled }); + + expect(await register(workspace.workspaceId, 'register-after-reenroll', { tags: [] })).toEqual(reEnrolled); + expect(await readTags(workspace.workspaceKey)).toEqual(reEnrolled); + }); + + it('leaves direct nodes preserving their enrollment tags, without accepting frame cloud:* tags', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-direct'); + await enroll(workspace.workspaceKey, { + role: 'direct', + max_agents: 1, + tags: ['enrolled', 'cloud:sandbox-id:sbx_direct', 'repo:acme/stale'], + }); + + const tags = await register(workspace.workspaceId, 'register-direct', { + max_agents: 1, + tags: ['current', 'cloud:sandbox-id:sbx_forged'], + repo_keys: ['AgentWorkforce/relaycast'], + }); + + // Unchanged direct-node behaviour: non-repo enrollment tags survive and the + // repo advertisement is refreshed. The frame's cloud:* tag is ignored. + expect(tags).toEqual(['enrolled', 'cloud:sandbox-id:sbx_direct', 'current', 'repo:AgentWorkforce/relaycast']); + expect(await readTags(workspace.workspaceKey)).toEqual([ + 'enrolled', + 'cloud:sandbox-id:sbx_direct', + 'current', + 'repo:AgentWorkforce/relaycast', + ]); + }); + + it('keeps a re-enroll that commits after registration read the row but before it writes', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-race'); + await enroll(workspace.workspaceKey, { max_agents: 4, tags: ENROLLED_CLOUD_TAGS }); + const reEnrolled = ['cloud:sandbox-provider:daytona', 'cloud:sandbox-id:sbx_ledger_2']; + + // registerNode reads the row, then writes tags inside runAtomic. Land a + // re-enroll in exactly that window: after the read, before the write + // transaction opens. A merge computed from the earlier read would put + // `sbx_ledger_1` back; the in-statement merge must keep `sbx_ledger_2`. + const db = stack.runtime.deps.db as EngineDb & { withTransaction: (fn: (tx: EngineDb) => Promise) => Promise }; + let reEnrolledMidRegister = false; + const racingDb = new Proxy(db, { + get(target, prop) { + if (prop === 'withTransaction') { + return async (fn: (tx: EngineDb) => Promise): Promise => { + if (!reEnrolledMidRegister) { + reEnrolledMidRegister = true; + await enroll(workspace.workspaceKey, { max_agents: 4, tags: reEnrolled }); + } + return target.withTransaction(fn); + }; + } + const value = Reflect.get(target, prop, target) as unknown; + return typeof value === 'function' ? value.bind(target) : value; + }, + }); + + const result = await registerNode(racingDb, workspace.workspaceId, 'node_sandbox', { + v: 1, + id: 'register-race', + type: 'node.register', + name: 'sandbox-node', + node_id: 'node_sandbox', + capabilities: [], + max_agents: 4, + tags: ['linux'], + version: 'test-node', + resume_cursor: null, + }, { name: DEFAULT_PROVIDER_NAME, instance_id: 'conn_race' }); + + expect(reEnrolledMidRegister).toBe(true); + expect(result.node.tags).toEqual([...reEnrolled, 'linux']); + expect(await readTags(workspace.workspaceKey)).toEqual([...reEnrolled, 'linux']); + }); + + it('logs the cloud:* tags a register frame tried to set, and still registers', async () => { + const workspace = await createWorkspace(stack.app, 'node-cloud-tags-warn'); + await enroll(workspace.workspaceKey, { max_agents: 4, tags: ENROLLED_CLOUD_TAGS }); + const warn = vi.spyOn(console, 'warn').mockImplementation(() => {}); + try { + const tags = await register(workspace.workspaceId, 'register-warn', { + tags: ['linux', 'cloud:region:eu-west', 'cloud:region:eu-west', 'cloud:sandbox-id:sbx_forged'], + }); + expect(tags).toEqual([...ENROLLED_CLOUD_TAGS, 'linux']); + const ignored = warn.mock.calls.filter(([label]) => label === '[node.register] ignored server-owned tags'); + expect(ignored).toEqual([[ + '[node.register] ignored server-owned tags', + { + workspaceId: workspace.workspaceId, + nodeId: 'node_sandbox', + prefix: 'cloud:', + tags: ['cloud:region:eu-west', 'cloud:sandbox-id:sbx_forged'], + }, + ]]); + + warn.mockClear(); + await register(workspace.workspaceId, 'register-no-warn', { tags: ['linux', 'cloudy'] }); + expect(warn.mock.calls.filter(([label]) => label === '[node.register] ignored server-owned tags')).toEqual([]); + } finally { + warn.mockRestore(); + } + }); +}); diff --git a/packages/engine/src/engine/node.ts b/packages/engine/src/engine/node.ts index 0799e416..37f0de9b 100644 --- a/packages/engine/src/engine/node.ts +++ b/packages/engine/src/engine/node.ts @@ -119,15 +119,55 @@ function isRepoTag(tag: string): boolean { // repo advertisements: once a registration carries the field at all - even as an // empty list - every caller-supplied `repo:` tag is dropped rather than merged. // Registrations that omit the field entirely are pre-`repo_keys` clients and -// stay on the legacy tag-only path. Non-repo tags always round-trip. +// stay on the legacy tag-only path. Other tags round-trip, except server-owned +// tags (below), which a registration can never supply. function registrationTags(message: FleetNodeRegisterMessage): string[] { - if (!message.repo_keys) return [...new Set(message.tags)]; + const callerTags = message.tags.filter((tag) => !isServerOwnedTag(tag)); + if (!message.repo_keys) return [...new Set(callerTags)]; return [...new Set([ - ...message.tags.filter((tag) => !isRepoTag(tag)), + ...callerTags.filter((tag) => !isRepoTag(tag)), ...message.repo_keys.map((repoKey) => `repo:${repoKey}`), ])]; } +// `cloud:*` tags are server-owned lifecycle identity (sandbox provider, node +// type, sandbox id, route). The control plane writes them at enrollment through +// POST /v1/nodes; the broker on the node never learns them and re-registers +// with its own tag list, often `[]`. So `node.register` carries the enrolled +// `cloud:*` tags over and ignores any `cloud:*` tag in the frame: a connected +// node must not be able to shed the identity reclaim uses to find it, or forge +// one that makes it look like a different sandbox. Only enrollment sets them, +// and only enrollment clears them: re-enrolling replaces the whole tag set, so +// a `cloud:*` value a pre-reservation broker once registered is removed by +// re-enrolling the node, not by a later register frame. A frame's `cloud:*` +// tags are ignored with a server-side warning rather than rejected, so a +// broker still configured with one keeps coming online. +export const SERVER_OWNED_NODE_TAG_PREFIX = 'cloud:'; + +function isServerOwnedTag(tag: string): boolean { + return tag.startsWith(SERVER_OWNED_NODE_TAG_PREFIX); +} + +// The tags a broker registration writes: the row's current `cloud:*` tags, +// then the broker's own (already stripped of `cloud:*` by registrationTags). +// This is computed inside the UPDATE rather than from a row read earlier, +// because a concurrent re-enroll can replace the `cloud:*` set in between and +// a read-then-write would put the stale set back. runAtomic is not isolated on +// every adapter (it runs unwrapped on D1) and enrollment does not take the +// node lock, so a single statement is the only boundary that holds everywhere. +// `substr` keeps the match case-sensitive like isServerOwnedTag; LIKE would not. +function brokerTagsPreservingServerOwned(callerTags: string[]) { + return sql`( + SELECT json_group_array(value) FROM ( + SELECT 0 AS src, key AS pos, value FROM json_each(${nodes.tags}) + WHERE substr(value, 1, ${SERVER_OWNED_NODE_TAG_PREFIX.length}) = ${SERVER_OWNED_NODE_TAG_PREFIX} + UNION ALL + SELECT 1 AS src, key AS pos, value FROM json_each(${JSON.stringify(callerTags)}) + ORDER BY src, pos + ) + )`; +} + function supportsProviderDeliveryReadiness(registry: NodeConnectionRegistry): boolean { return typeof registry.setProviderDeliveryReadiness === 'function' && typeof registry.markProviderAgentsDeliveryReady === 'function' @@ -512,6 +552,18 @@ export async function registerNode( } const tags = registrationTags(message); + const ignoredServerOwnedTags = [...new Set(message.tags.filter(isServerOwnedTag))]; + if (ignoredServerOwnedTags.length > 0) { + // Not rejected: a broker whose manifest predates the reserved namespace + // must still come online. But dropping part of a valid frame without a + // trace hides the misconfiguration, so record which tags were ignored. + console.warn('[node.register] ignored server-owned tags', { + workspaceId, + nodeId: authenticatedNodeId, + prefix: SERVER_OWNED_NODE_TAG_PREFIX, + tags: ignoredServerOwnedTags, + }); + } const [existingByName] = await db .select() @@ -567,7 +619,7 @@ export async function registerNode( await materializeProviderActions(tx, workspaceId, authenticatedNodeId, provider.name, capabilities); await tx .update(nodes) - .set({ name: message.name, kind: 'ws', role: 'broker', deliveryAdapter: 'ws.node.v1', deliveryConfig: null, tags }) + .set({ name: message.name, kind: 'ws', role: 'broker', deliveryAdapter: 'ws.node.v1', deliveryConfig: null, tags: brokerTagsPreservingServerOwned(tags) }) .where(and(eq(nodes.workspaceId, workspaceId), eq(nodes.id, authenticatedNodeId))); await recomputeNodeAggregate(tx, workspaceId, authenticatedNodeId, { version: message.version, diff --git a/packages/types/src/fleet-wire.ts b/packages/types/src/fleet-wire.ts index 6e26dc63..87900644 100644 --- a/packages/types/src/fleet-wire.ts +++ b/packages/types/src/fleet-wire.ts @@ -168,6 +168,9 @@ export const FleetNodeRegisterMessageSchema = z capabilities: z.array(FleetCapabilitySchema), // Provider-level capacity; the node figure is the aggregate across providers. max_agents: z.number().int().nonnegative(), + // `cloud:*` is reserved for control-plane lifecycle tags set at enrollment. + // The engine ignores (and logs) any `cloud:*` entry here and keeps the + // node's enrolled `cloud:*` tags instead. tags: z.array(FleetNodeTagSchema), // Placement-safe repository identities. The engine persists these as // `repo:` tags so existing node roster readers can consume them.