diff --git a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift index 8d1537d1b016..daa8667297df 100644 --- a/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift +++ b/apps/shared/OpenClawKit/Sources/OpenClawProtocol/GatewayModels.swift @@ -7877,25 +7877,29 @@ public struct SessionMoveGatewayTarget: Codable, Sendable { public struct SessionMoveProfileTarget: Codable, Sendable { public let kind: String public let profileid: String + public let machineclass: String? public init( - profileid: String + profileid: String, + machineclass: String? = nil ) { self.kind = "profile" self.profileid = profileid + self.machineclass = machineclass } private enum CodingKeys: String, CodingKey { case kind case profileid = "profileId" + case machineclass = "machineClass" } public init(from decoder: Decoder) throws { let rawContainer = try decoder.container(keyedBy: GatewayAnyCodingKey.self) let unexpectedKeys = rawContainer.allKeys .map(\.stringValue) - .filter { !Set(["kind", "profileId"]).contains($0) } + .filter { !Set(["kind", "profileId", "machineClass"]).contains($0) } if !unexpectedKeys.isEmpty { throw DecodingError.dataCorrupted( .init( @@ -7915,12 +7919,14 @@ public struct SessionMoveProfileTarget: Codable, Sendable { } self.kind = "profile" self.profileid = try container.decode(String.self, forKey: .profileid) + self.machineclass = try container.decodeIfPresent(String.self, forKey: .machineclass) } public func encode(to encoder: Encoder) throws { var container = encoder.container(keyedBy: CodingKeys.self) try container.encode("profile", forKey: .kind) try container.encode(profileid, forKey: .profileid) + try container.encodeIfPresent(machineclass, forKey: .machineclass) } } diff --git a/docs/gateway/cloud-workers.md b/docs/gateway/cloud-workers.md index e66a3e25fd87..f0d075e1fd5c 100644 --- a/docs/gateway/cloud-workers.md +++ b/docs/gateway/cloud-workers.md @@ -195,7 +195,7 @@ Apply uses the dispatch-time manifest as the merge base. Cloud-only changes are While a fenced result is still reconciling, a new turn waits up to 15 seconds for the prior claim to release. If it is still busy, the turn fails with an actionable “previous cloud turn's workspace result is still reconciling” message and can be retried shortly. On restart, recovery discovers pending and staged results before stale-claim cleanup, completes or retries their local apply, and reclaims dead environments only after preserving the result. The bounded SQLite rollback journal makes an interrupted filesystem apply recoverable without replaying already accepted mutations. -To continue the same session somewhere else, open the **Runs on Cloud** chip and choose **Move session…**. Select the Gateway, a paired device, or another configured cloud profile. The Gateway closes new admission, interrupts any active turn, reconciles the source workspace, destroys the old environment, and then activates the destination. An interrupted turn is never replayed: partial output may disappear, and you send the next turn again after the move. Move intent and bounded errors are durable, so the Control UI shows **Moving to…** or the recovery error after a reconnect, and Gateway restart recovery resumes the same operation before generic placement cleanup. +To continue the same session somewhere else, open the **Runs on Cloud** chip and choose **Move session…**. Select the Gateway, a paired device, or a configured cloud profile and, when available, its machine class. Moving to the current profile with a different class resizes the session by replacing its worker. The Gateway closes new admission, interrupts any active turn, reconciles the source workspace, destroys the old environment, and then activates the destination. An interrupted turn is never replayed: partial output may disappear, and you send the next turn again after the move. The exact target, including a machine override, and bounded errors are durable, so the Control UI shows **Moving to…** or the recovery error after a reconnect, and Gateway restart recovery resumes the same operation before generic placement cleanup. When the work is complete and no turn is running, choose **Stop cloud worker…** from the same chip. The Gateway performs one final workspace reconciliation before it destroys the environment. A placement already in `draining` or `reconciling` is finishing teardown; wait for its badge to become `reclaimed` before deleting the session. @@ -221,7 +221,7 @@ openclaw gateway call sessions.move \ --params '{"key":"agent:main:big-refactor","expected":{"generation":5,"environmentId":"worker:source","ownerEpoch":2},"target":{"kind":"gateway"}}' ``` -Worker targets use `{"kind":"profile","profileId":"aws"}` or `{"kind":"device","deviceId":"paired-device-id"}`. A stale source is rejected rather than moving a newer placement. Successful results end in `local` for the Gateway target or `active` for a worker target. +Worker targets use `{"kind":"profile","profileId":"aws","machineClass":"fast"}` or `{"kind":"device","deviceId":"paired-device-id"}`. Omit `machineClass` to use the profile default. Moving to the same profile with a different class is the resize workflow. A stale source is rejected rather than moving a newer placement. Successful results end in `local` for the Gateway target or `active` for a worker target. Placement moves through a durable state machine (`local → requested → provisioning → syncing → starting → active`), so a Gateway restart mid-dispatch reconciles instead of leaking machines. A failed model turn keeps the active placement available for a retry. Workspace path conflicts keep the local version, apply the rest of the cloud result, and preserve the staged cloud ref for inspection; other reconciliation or lifecycle failures retain their durable recovery fence and diagnostic tail until recovery can safely retry or reclaim the environment. diff --git a/docs/web/control-ui.md b/docs/web/control-ui.md index ab88e755b334..178316511929 100644 --- a/docs/web/control-ui.md +++ b/docs/web/control-ui.md @@ -228,7 +228,7 @@ The sidebar organizes everything around the agent. The identity row at the top i ### Session placement -A selected session running on a worker shows a quiet **Runs on Cloud** chip in the chat header. Open it and choose **Move session…** to continue on the Gateway, an eligible paired device, or another configured cloud profile. The confirmation explains that an active turn is interrupted and never replayed; OpenClaw reconciles the workspace before activating the destination. While the durable operation is in progress, the chip shows **Moving to…**. If recovery is blocked, the chip exposes the bounded error after reconnect so the action never fails silently. **Stop cloud worker…** remains the destructive action for returning the session to a reclaimed state without immediately selecting another runner. +A selected session running on a worker shows a quiet **Runs on Cloud** chip in the chat header. Open it and choose **Move session…** to continue on the Gateway, an eligible paired device, or a configured cloud profile. Profiles with multiple machine classes show a machine picker; choosing the default omits an override, while choosing a different class on the current profile resizes the session. The confirmation explains that an active turn is interrupted and never replayed; OpenClaw reconciles the workspace before activating the destination. While the durable operation is in progress, the chip shows **Moving to…**. If recovery is blocked, the chip exposes the bounded error after reconnect so the action never fails silently. **Stop cloud worker…** remains the destructive action for returning the session to a reclaimed state without immediately selecting another runner. ### Session icons diff --git a/packages/gateway-protocol/src/schema/session-placement.test.ts b/packages/gateway-protocol/src/schema/session-placement.test.ts index f50768e8be84..f30c8d889292 100644 --- a/packages/gateway-protocol/src/schema/session-placement.test.ts +++ b/packages/gateway-protocol/src/schema/session-placement.test.ts @@ -364,7 +364,7 @@ describe("session dispatch protocol schemas", () => { it.each([ { kind: "gateway" }, - { kind: "profile", profileId: "development" }, + { kind: "profile", profileId: "development", machineClass: "beast" }, { kind: "device", deviceId: "device-1" }, ] as const)("accepts the closed $kind move target", (target) => { expect( @@ -384,9 +384,18 @@ describe("session dispatch protocol schemas", () => { validateSessionsMoveParams({ key: "agent:main:dispatch", expected: { generation: 4, environmentId: accepted, ownerEpoch: 7 }, - target: { kind: "profile", profileId: accepted }, + target: { kind: "profile", profileId: accepted, machineClass: "x".repeat(128) }, }), ).toBe(true); + for (const machineClass of ["", "x".repeat(129)]) { + expect( + validateSessionsMoveParams({ + key: "agent:main:dispatch", + expected: { generation: 4, environmentId: "environment-1", ownerEpoch: 7 }, + target: { kind: "profile", profileId: "development", machineClass }, + }), + ).toBe(false); + } for (const value of [rejected, " leading", "trailing "]) { expect( validateSessionsMoveParams({ @@ -407,9 +416,11 @@ describe("session dispatch protocol schemas", () => { it.each([ { kind: "gateway", profileId: "development" }, + { kind: "gateway", machineClass: "beast" }, { kind: "profile" }, { kind: "profile", profileId: "development", deviceId: "device-1" }, { kind: "device" }, + { kind: "device", deviceId: "device-1", machineClass: "beast" }, { kind: "other" }, ])("rejects an invalid or mixed move target %#", (target) => { expect( diff --git a/packages/gateway-protocol/src/schema/session-placement.ts b/packages/gateway-protocol/src/schema/session-placement.ts index 6db5acc4ec16..f5200699c412 100644 --- a/packages/gateway-protocol/src/schema/session-placement.ts +++ b/packages/gateway-protocol/src/schema/session-placement.ts @@ -253,6 +253,7 @@ export const SessionMoveGatewayTargetSchema = closedObject({ export const SessionMoveProfileTargetSchema = closedObject({ kind: Type.Literal("profile"), profileId: WorkerIdentifierSchema, + machineClass: Type.Optional(WorkerMachineClassSchema), }); /** Moves the session to one paired device worker. */ diff --git a/src/gateway/server-methods/chat-history-handler.ts b/src/gateway/server-methods/chat-history-handler.ts index 6be47cede853..eb49f86c61bf 100644 --- a/src/gateway/server-methods/chat-history-handler.ts +++ b/src/gateway/server-methods/chat-history-handler.ts @@ -61,6 +61,7 @@ import { startOptionalServerMethodModelCatalogSnapshotLoad, } from "./optional-model-catalog.js"; import { resolveVisibleActiveSessionRunState } from "./session-active-runs.js"; +import { readSessionPlacementFields } from "./session-placement-read-projection.js"; import type { GatewayRequestHandlerOptions, GatewayRequestHandlers } from "./types.js"; import { assertValidParams } from "./validation.js"; @@ -479,6 +480,10 @@ async function handleChatHistoryRequest({ }); sessionInfo.hasActiveRun = activeRunState.active; sessionInfo.activeRunIds = activeRunState.runIds; + // Clients merge this row into the same store sessions.list fills, so it must + // carry the placement facts that projection adds; without them the merge + // erases a live worker placement and its move intent. + Object.assign(sessionInfo, readSessionPlacementFields(context, entry?.sessionId)); if (Object.hasOwn(historyPage, "activeLeafEntryId")) { sessionInfo.activeLeafEntryId = historyPage.activeLeafEntryId ?? null; } diff --git a/src/gateway/server-methods/sessions-dispatch.test-support.ts b/src/gateway/server-methods/sessions-dispatch.test-support.ts index f41d45cb4e2e..36a81c87bff3 100644 --- a/src/gateway/server-methods/sessions-dispatch.test-support.ts +++ b/src/gateway/server-methods/sessions-dispatch.test-support.ts @@ -145,7 +145,7 @@ export async function invokeSessionMove( expected: { generation: number; environmentId: string; ownerEpoch: number }; target: | { kind: "gateway" } - | { kind: "profile"; profileId: string } + | { kind: "profile"; profileId: string; machineClass?: string } | { kind: "device"; deviceId: string }; }, ) { diff --git a/src/gateway/server-methods/sessions.dispatch.test.ts b/src/gateway/server-methods/sessions.dispatch.test.ts index f8ce583921cd..30fb06fc4876 100644 --- a/src/gateway/server-methods/sessions.dispatch.test.ts +++ b/src/gateway/server-methods/sessions.dispatch.test.ts @@ -411,13 +411,13 @@ describe("sessions.dispatch", () => { }), { expected: { generation: 4, environmentId: "environment-previous", ownerEpoch: 1 }, - target: { kind: "profile", profileId: "test" }, + target: { kind: "profile", profileId: "test", machineClass: "beast" }, }, ); expect(move).toHaveBeenCalledWith( expect.objectContaining({ - target: { kind: "profile", profileId: "test" }, + target: { kind: "profile", profileId: "test", machineClass: "beast" }, }), expect.any(Function), ); diff --git a/src/gateway/server-worker-placement-startup.ts b/src/gateway/server-worker-placement-startup.ts index 7ce9ec1f348a..345098bbcf71 100644 --- a/src/gateway/server-worker-placement-startup.ts +++ b/src/gateway/server-worker-placement-startup.ts @@ -367,7 +367,7 @@ export function createGatewayWorkerPlacementRuntime(params: GatewayWorkerPlaceme const destination = resolveWorkerPlacementDestination({ cfg: config, ...(moveTarget.kind === "profile" - ? { profileId: moveTarget.profileId } + ? { profileId: moveTarget.profileId, machineClass: moveTarget.machineClass } : { deviceId: moveTarget.deviceId }), }); if (!destination.ok || !destination.value) { diff --git a/src/gateway/server.chat.gateway-server-chat-b.test.ts b/src/gateway/server.chat.gateway-server-chat-b.test.ts index 7d5b5e1d4aac..9bcb4ad46263 100644 --- a/src/gateway/server.chat.gateway-server-chat-b.test.ts +++ b/src/gateway/server.chat.gateway-server-chat-b.test.ts @@ -684,6 +684,60 @@ async function prepareMainHistoryHarness(params: { } describe("gateway server chat", () => { + test.each(["chat.history", "chat.startup"] as const)( + "%s projects the session's durable worker placement", + async (method) => { + openDirectChatSession(); + try { + await writeMainSessionStore(); + const placement = { + sessionId: "sess-main", + agentId: "main", + sessionKey: "agent:main:main", + executionMode: "worker-turn", + state: "active", + environmentId: "env-placement", + generation: 7, + activeOwnerEpoch: 12, + workspaceBaseManifestRef: "manifest-base", + remoteWorkspaceDir: "/workspace/main", + workerBundleHash: "ab".repeat(32), + recoveryError: null, + terminalReason: null, + terminalAtMs: null, + turnClaim: null, + createdAtMs: 100, + updatedAtMs: 300, + stateChangedAtMs: 200, + }; + const context = createDirectChatContext({ + workerSessionPlacementService: { + getMany: () => new Map([[placement.sessionId, placement]]), + getPlacementMoves: () => new Map(), + }, + } as unknown as Partial); + const responses: Array<{ ok: boolean; payload?: unknown }> = []; + await callDirectChat(method, { + id: method, + params: makeMainSessionParams(), + respond: captureChatResult(responses), + context, + }); + + expect(responses[0]?.ok).toBe(true); + // Clients merge this row into the same store sessions.list fills, so a + // missing placement here silently erases a live worker placement. + expect( + (responses[0]?.payload as { sessionInfo?: { placement?: { state?: string } } }) + ?.sessionInfo?.placement, + ).toMatchObject({ state: "active", environmentId: "env-placement" }); + } finally { + testState.sessionStorePath = undefined; + clearConfigCache(); + } + }, + ); + test.each(["chat.history", "chat.startup"] as const)( "%s replays the active plan snapshot in inFlightRun", async (method) => { diff --git a/src/gateway/worker-environments/placement-move-intent.ts b/src/gateway/worker-environments/placement-move-intent.ts index 658ba2c0169d..c39305c59271 100644 --- a/src/gateway/worker-environments/placement-move-intent.ts +++ b/src/gateway/worker-environments/placement-move-intent.ts @@ -9,7 +9,7 @@ import { getNodeSqliteKysely, } from "../../infra/kysely-sync.js"; import { generateSecureToken } from "../../infra/secure-random.js"; -import { tableExists } from "../../state/openclaw-state-db-schema-helpers.js"; +import { ensureColumn, tableExists } from "../../state/openclaw-state-db-schema-helpers.js"; import type { DB as StateDatabase, WorkerSessionPlacementMoves, @@ -24,6 +24,7 @@ import { boundedWorkerError } from "./worker-error.js"; const MOVE_SCHEMA_START = "CREATE TABLE IF NOT EXISTS worker_session_placement_moves ("; const MOVE_SCHEMA_END = "\n) STRICT;"; const MOVE_OPERATION_PREFIX = "move:v1:"; +const MOVE_MACHINE_CLASS_MAX_LENGTH = 128; type MoveRow = Selectable; type MoveDatabase = Pick< @@ -60,8 +61,31 @@ function moveSchemaSql(): string { return OPENCLAW_STATE_SCHEMA_SQL.slice(start, endMarkerStart + MOVE_SCHEMA_END.length); } +// Single-slot per-handle memo: getPlacementMoves feeds the sessions read +// projection, so the DDL/PRAGMA ensure must not run per read. +const ensuredMoveSchemaHandles = new WeakSet(); + function ensureWorkerPlacementMoveSchema(db: DatabaseSync): void { + if (ensuredMoveSchemaHandles.has(db)) { + return; + } db.exec(moveSchemaSql()); // sqlite-allow-raw -- Canonical feature-owned additive DDL only. + // Databases that created this table before the column shipped upgrade in place; + // the column is bare and nullable, so old readers stay compatible. + ensureColumn(db, "worker_session_placement_moves", "target_machine_class TEXT"); + ensuredMoveSchemaHandles.add(db); +} + +function ensureExistingWorkerPlacementMoveSchema(db: DatabaseSync): boolean { + if (ensuredMoveSchemaHandles.has(db)) { + return true; + } + // Reads stay lazy: never create the optional table from a read path. + if (!tableExists(db, "worker_session_placement_moves")) { + return false; + } + ensureWorkerPlacementMoveSchema(db); + return true; } function normalizeGeneration(value: number): number { @@ -71,12 +95,14 @@ function normalizeGeneration(value: number): number { return value; } -function boundedIdentifier(value: string, field: string): string { +function boundedIdentifier( + value: string, + field: string, + maximumLength = WORKER_PROTOCOL_MAX_IDENTIFIER_LENGTH, +): string { const normalized = required(value, field); - if (normalized.length > WORKER_PROTOCOL_MAX_IDENTIFIER_LENGTH) { - throw new Error( - `Worker session placement ${field} exceeds ${WORKER_PROTOCOL_MAX_IDENTIFIER_LENGTH} characters`, - ); + if (normalized.length > maximumLength) { + throw new Error(`Worker session placement ${field} exceeds ${maximumLength} characters`); } return normalized; } @@ -95,8 +121,22 @@ function normalizeWorkerPlacementMoveTarget( switch (target.kind) { case "gateway": return { kind: "gateway" }; - case "profile": - return { kind: "profile", profileId: boundedIdentifier(target.profileId, "move profile id") }; + case "profile": { + const machineClass = target.machineClass; + return { + kind: "profile", + profileId: boundedIdentifier(target.profileId, "move profile id"), + ...(machineClass === undefined + ? {} + : { + machineClass: boundedIdentifier( + machineClass, + "move machine class", + MOVE_MACHINE_CLASS_MAX_LENGTH, + ), + }), + }; + } case "device": return { kind: "device", deviceId: boundedIdentifier(target.deviceId, "move device id") }; } @@ -116,14 +156,19 @@ function normalizeWorkerPlacementMoveSource( function targetValues(target: WorkerPlacementMoveTarget): { target_kind: MoveRow["target_kind"]; target_id: MoveRow["target_id"]; + target_machine_class: MoveRow["target_machine_class"]; } { switch (target.kind) { case "gateway": - return { target_kind: target.kind, target_id: null }; + return { target_kind: target.kind, target_id: null, target_machine_class: null }; case "profile": - return { target_kind: target.kind, target_id: target.profileId }; + return { + target_kind: target.kind, + target_id: target.profileId, + target_machine_class: target.machineClass ?? null, + }; case "device": - return { target_kind: target.kind, target_id: target.deviceId }; + return { target_kind: target.kind, target_id: target.deviceId, target_machine_class: null }; } throw new Error("Worker placement move target is invalid"); } @@ -135,10 +180,25 @@ function fromRow(row: MoveRow): WorkerPlacementMoveIntent { ownerEpoch: row.source_owner_epoch, }); let target: WorkerPlacementMoveTarget; + if (row.target_kind !== "profile" && row.target_machine_class !== null) { + throw new Error(`Invalid worker placement move target: ${row.target_kind}`); + } if (row.target_kind === "gateway" && row.target_id === null) { target = { kind: "gateway" }; } else if (row.target_kind === "profile" && row.target_id !== null) { - target = { kind: "profile", profileId: boundedIdentifier(row.target_id, "move profile id") }; + target = { + kind: "profile", + profileId: boundedIdentifier(row.target_id, "move profile id"), + ...(row.target_machine_class === null + ? {} + : { + machineClass: boundedIdentifier( + row.target_machine_class, + "move machine class", + MOVE_MACHINE_CLASS_MAX_LENGTH, + ), + }), + }; } else if (row.target_kind === "device" && row.target_id !== null) { target = { kind: "device", deviceId: boundedIdentifier(row.target_id, "move device id") }; } else { @@ -156,7 +216,7 @@ function fromRow(row: MoveRow): WorkerPlacementMoveIntent { } function findMoveRowBySession(db: DatabaseSync, sessionId: string): MoveRow | undefined { - if (!tableExists(db, "worker_session_placement_moves")) { + if (!ensureExistingWorkerPlacementMoveSchema(db)) { return undefined; } return executeSqliteQueryTakeFirstSync( @@ -169,7 +229,7 @@ function findMoveRowBySession(db: DatabaseSync, sessionId: string): MoveRow | un } function findMoveRowByOperation(db: DatabaseSync, operationId: string): MoveRow | undefined { - if (!tableExists(db, "worker_session_placement_moves")) { + if (!ensureExistingWorkerPlacementMoveSchema(db)) { return undefined; } return executeSqliteQueryTakeFirstSync( @@ -208,6 +268,10 @@ function deleteExactMove(db: DatabaseSync, intent: WorkerPlacementMoveIntent): v values.target_id === null ? statement.where("target_id", "is", null) : statement.where("target_id", "=", values.target_id); + statement = + values.target_machine_class === null + ? statement.where("target_machine_class", "is", null) + : statement.where("target_machine_class", "=", values.target_machine_class); const result = executeSqliteQuerySync(db, statement); if (result.numAffectedRows !== 1n) { throw new Error(`Session ${intent.sessionId} placement move changed before completion`); @@ -265,7 +329,7 @@ export function createPlacementMoveOps(runtime: PlacementStoreRuntime) { ]; const results = new Map(); const db = read(); - if (!tableExists(db, "worker_session_placement_moves")) { + if (!ensureExistingWorkerPlacementMoveSchema(db)) { return results; } for (let offset = 0; offset < normalizedIds.length; offset += 250) { @@ -286,7 +350,7 @@ export function createPlacementMoveOps(runtime: PlacementStoreRuntime) { listPlacementMoves(): WorkerPlacementMoveIntent[] { const db = read(); - if (!tableExists(db, "worker_session_placement_moves")) { + if (!ensureExistingWorkerPlacementMoveSchema(db)) { return []; } return executeSqliteQuerySync( @@ -390,6 +454,10 @@ export function createPlacementMoveOps(runtime: PlacementStoreRuntime) { values.target_id === null ? statement.where("target_id", "is", null) : statement.where("target_id", "=", values.target_id); + statement = + values.target_machine_class === null + ? statement.where("target_machine_class", "is", null) + : statement.where("target_machine_class", "=", values.target_machine_class); return executeSqliteQuerySync(db, statement).numAffectedRows === 1n; }); }, diff --git a/src/gateway/worker-environments/placement-move-schema.test.ts b/src/gateway/worker-environments/placement-move-schema.test.ts index 1b3bec21a439..b7018accddda 100644 --- a/src/gateway/worker-environments/placement-move-schema.test.ts +++ b/src/gateway/worker-environments/placement-move-schema.test.ts @@ -7,6 +7,7 @@ import { openOpenClawStateDatabase, } from "../../state/openclaw-state-db.js"; import { getOpenClawStateRuntimeSchema } from "../../state/openclaw-state-schema-compatibility.js"; +import { OPENCLAW_STATE_SCHEMA_SQL } from "../../state/openclaw-state-schema.js"; import { createWorkerSessionPlacementStore } from "./placement-store.js"; const tempDirs = useAutoCleanupTempDirTracker(afterEach); @@ -24,8 +25,14 @@ describe("worker placement move schema", () => { const metadataBefore = database.db .prepare("SELECT schema_version, updated_at FROM schema_meta WHERE meta_key = 'primary'") .get(); + const previousSchema = OPENCLAW_STATE_SCHEMA_SQL.replace(" target_machine_class TEXT,\n", ""); + const moveSchemaStart = previousSchema.indexOf( + "CREATE TABLE IF NOT EXISTS worker_session_placement_moves (", + ); + const moveSchemaEnd = previousSchema.indexOf(") STRICT;", moveSchemaStart); database.db.exec(` DROP TABLE worker_session_placement_moves; + ${previousSchema.slice(moveSchemaStart, moveSchemaEnd + ") STRICT;".length)} INSERT INTO worker_environments ( environment_id, provider_id, profile_id, profile_snapshot_json, provision_operation_id, lease_id, state, owner_epoch, @@ -50,8 +57,11 @@ describe("worker placement move schema", () => { const begun = store.beginPlacementMove({ sessionId: "session-move", source: { generation: 4, environmentId: "environment-source", ownerEpoch: 7 }, - target: { kind: "profile", profileId: "profile-destination" }, + target: { kind: "profile", profileId: "profile-destination", machineClass: "beast" }, }); + expect(database.db.prepare("PRAGMA table_info(worker_session_placement_moves)").all()).toEqual( + expect.arrayContaining([expect.objectContaining({ name: "target_machine_class" })]), + ); const databasePath = database.path; closeOpenClawStateDatabaseForTest(); diff --git a/src/gateway/worker-environments/placement-store.move.test.ts b/src/gateway/worker-environments/placement-store.move.test.ts index 070440a00950..cac5a963a7d9 100644 --- a/src/gateway/worker-environments/placement-store.move.test.ts +++ b/src/gateway/worker-environments/placement-store.move.test.ts @@ -177,6 +177,69 @@ describe("worker session placement moves", () => { ).toThrow("already has a conflicting placement move"); }); + it("persists a profile machine class and joins only the exact target", () => { + const active = advanceToActive(); + seedAttachedEnvironment({ + environmentId: active.environmentId, + sessionId: active.sessionId, + ownerEpoch: active.activeOwnerEpoch, + }); + const source = { + generation: active.generation, + environmentId: active.environmentId, + ownerEpoch: active.activeOwnerEpoch, + }; + const target = { + kind: "profile", + profileId: "profile-destination", + machineClass: "beast", + } as const; + const begun = store.beginPlacementMove({ + sessionId: SESSION.sessionId, + source, + target: { ...target, machineClass: " beast " }, + }); + + expect(store.getPlacementMove(SESSION.sessionId)).toMatchObject({ target }); + expect( + store.beginPlacementMove({ sessionId: SESSION.sessionId, source, target }), + ).toMatchObject({ joined: true, intent: { operationId: begun.intent.operationId, target } }); + expect(() => + store.beginPlacementMove({ + sessionId: SESSION.sessionId, + source, + target: { ...target, machineClass: "fast" }, + }), + ).toThrow("already has a conflicting placement move"); + }); + + it("rejects a machine class stored for a non-profile target", () => { + const active = advanceToActive(); + seedAttachedEnvironment({ + environmentId: active.environmentId, + sessionId: active.sessionId, + ownerEpoch: active.activeOwnerEpoch, + }); + store.beginPlacementMove({ + sessionId: SESSION.sessionId, + source: { + generation: active.generation, + environmentId: active.environmentId, + ownerEpoch: active.activeOwnerEpoch, + }, + target: { kind: "gateway" }, + }); + database.db + .prepare( + "UPDATE worker_session_placement_moves SET target_machine_class = 'beast' WHERE session_id = ?", + ) + .run(SESSION.sessionId); + + expect(() => store.getPlacementMove(SESSION.sessionId)).toThrow( + "Invalid worker placement move target: gateway", + ); + }); + it("keeps invalid move attempts from creating optional storage", () => { database.db.exec("DROP TABLE worker_session_placement_moves"); const active = advanceToActive(); @@ -276,7 +339,7 @@ describe("worker session placement moves", () => { environmentId: source.environmentId, ownerEpoch: source.activeOwnerEpoch, }, - target: { kind: "profile", profileId: "profile-destination" }, + target: { kind: "profile", profileId: "profile-destination", machineClass: "beast" }, }); const reconciling = store.startReconcile({ sessionId: SESSION.sessionId, @@ -322,7 +385,7 @@ describe("worker session placement moves", () => { expect(store.getPlacementMove(SESSION.sessionId)).toBeUndefined(); }); - it("retries a failed destination from local without reclaiming the old source", async () => { + it("restart-recovers a profile move with its persisted machine class", async () => { const source = advanceToActive(); seedAttachedEnvironment({ environmentId: source.environmentId, @@ -336,7 +399,7 @@ describe("worker session placement moves", () => { environmentId: source.environmentId, ownerEpoch: source.activeOwnerEpoch, }, - target: { kind: "profile", profileId: "profile-destination" }, + target: { kind: "profile", profileId: "profile-destination", machineClass: "beast" }, }); const reconciling = store.startReconcile({ sessionId: source.sessionId, @@ -369,27 +432,41 @@ describe("worker session placement moves", () => { const reclaimSource = vi.fn(async () => { throw new Error("failed destination must not reclaim the old source"); }); + const restartedStore = createWorkerSessionPlacementStore({ database, now: () => nowMs }); const moves = createWorkerPlacementMoveService({ - placements: store, + placements: restartedStore, environments: { get: () => undefined }, runMoveBarrier: async ({ begin }) => begin(), dispatch, reclaimSource, - resolveDestination: async () => ({ - profileId: "profile-destination", - executionMode: "worker-turn", - }), + resolveDestination: async (_identity, target) => { + if (target.kind !== "profile") { + throw new Error("expected profile move target"); + } + return { + profileId: target.profileId, + executionMode: "worker-turn", + machineClass: target.machineClass, + }; + }, }); await moves.recoverAll(); expect(reclaimSource).not.toHaveBeenCalled(); expect(dispatch).toHaveBeenCalledOnce(); - expect(store.get(source.sessionId)).toMatchObject({ + expect(dispatch).toHaveBeenCalledWith( + expect.objectContaining({ + machineClass: "beast", + idempotencyKey: `session-move:${begun.intent.operationId}:dispatch`, + }), + undefined, + ); + expect(restartedStore.get(source.sessionId)).toMatchObject({ state: "local", generation: local.generation + 4, }); - expect(store.getPlacementMove(source.sessionId)).toMatchObject({ + expect(restartedStore.getPlacementMove(source.sessionId)).toMatchObject({ operationId: begun.intent.operationId, lastError: dispatchError.message, }); diff --git a/src/gateway/worker-environments/service-contract.ts b/src/gateway/worker-environments/service-contract.ts index 5a1d9fc57d1f..1e2b32da8c33 100644 --- a/src/gateway/worker-environments/service-contract.ts +++ b/src/gateway/worker-environments/service-contract.ts @@ -104,7 +104,7 @@ export type WorkerPlacementDispatchRequest = { export type WorkerPlacementMoveDestination = Pick< WorkerPlacementDispatchRequest, - "profileId" | "executionMode" | "deviceId" | "inheritedProfile" + "profileId" | "executionMode" | "deviceId" | "machineClass" | "inheritedProfile" >; export type WorkerPlacementReclaimRequest = { diff --git a/src/state/openclaw-state-db.generated.d.ts b/src/state/openclaw-state-db.generated.d.ts index 441608da7321..00303d0283ba 100644 --- a/src/state/openclaw-state-db.generated.d.ts +++ b/src/state/openclaw-state-db.generated.d.ts @@ -1604,6 +1604,7 @@ export interface WorkerSessionPlacementMoves { source_owner_epoch: number; target_id: string | null; target_kind: string; + target_machine_class: string | null; updated_at_ms: number; } diff --git a/src/state/openclaw-state-schema.sql b/src/state/openclaw-state-schema.sql index c25b8a826046..4b1130be85ed 100644 --- a/src/state/openclaw-state-schema.sql +++ b/src/state/openclaw-state-schema.sql @@ -2268,6 +2268,9 @@ CREATE TABLE IF NOT EXISTS worker_session_placement_moves ( source_owner_epoch INTEGER NOT NULL CHECK (source_owner_epoch >= 1), target_kind TEXT NOT NULL CHECK (target_kind IN ('gateway', 'profile', 'device')), target_id TEXT, + -- Keep this nullable column constraint-free so lazy ALTER TABLE produces the + -- same shape as fresh databases; placement-move code validates its value. + target_machine_class TEXT, last_error TEXT, created_at_ms INTEGER NOT NULL, updated_at_ms INTEGER NOT NULL, diff --git a/ui/src/components/session-placement-move-dialog.ts b/ui/src/components/session-placement-move-dialog.ts index 211431e866fd..304b0d408d63 100644 --- a/ui/src/components/session-placement-move-dialog.ts +++ b/ui/src/components/session-placement-move-dialog.ts @@ -3,11 +3,13 @@ import type { SessionMoveTarget } from "../../../packages/gateway-protocol/src/i import { t } from "../i18n/index.ts"; import { formatUiError } from "../lib/format-error.ts"; import { + renderCloudMachineMenuItems, renderCloudProfileMenuItems, renderSessionMenuItem, } from "../pages/new-session/cloud-target.ts"; import type { DraftCloudProfile, DraftNode } from "../pages/new-session/discovery.ts"; import { isDraftNodeSessionEligible } from "../pages/new-session/discovery.ts"; +import { DraftCloudMachineState } from "../pages/new-session/draft-cloud-machine-state.ts"; import "../styles/new-session.css"; import { icons } from "./icons.ts"; import "./modal-dialog.ts"; @@ -51,6 +53,7 @@ export function showSessionPlacementMoveDialog( let loadError: string | null = null; let catalog: Catalog = { profiles: [], nodes: [] }; let selected: SessionMoveTarget = { kind: "gateway" }; + const cloudMachines = new DraftCloudMachineState(); const finish = (result: SessionMoveTarget | null) => { render(nothing, host); @@ -66,7 +69,15 @@ export function showSessionPlacementMoveDialog( const submit = (event: Event) => { event.preventDefault(); - finish(selected); + if (selected.kind !== "profile") { + finish(selected); + return; + } + const machineClass = cloudMachines.resolve(selected.profileId); + finish({ + ...selected, + ...(machineClass ? { machineClass } : {}), + }); }; function paint() { @@ -131,12 +142,43 @@ export function showSessionPlacementMoveDialog(
${t("newSession.cloud")}
- ${renderCloudProfileMenuItems({ - profiles: catalog.profiles, - selectedId: selected.kind === "profile" ? selected.profileId : "", - submitting: false, - icon: icons.server, - onSelect: (profileId) => select({ kind: "profile", profileId }), + ${catalog.profiles.map((profile) => { + const profileSelected = + selected.kind === "profile" && selected.profileId === profile.id; + const machines = profile.machines ?? []; + const selectedMachineId = + cloudMachines.resolve(profile.id) || + machines.find((machine) => machine.default === true)?.id || + ""; + return html` + ${renderCloudProfileMenuItems({ + profiles: [profile], + selectedId: profileSelected ? profile.id : "", + submitting: false, + icon: icons.server, + onSelect: (profileId) => select({ kind: "profile", profileId }), + })} + ${profileSelected && machines.length > 0 + ? html` +
+ ${t("newSession.machine")} +
+ ${renderCloudMachineMenuItems({ + machines, + selectedId: selectedMachineId, + submitting: false, + onSelect: (machineId) => + cloudMachines.select( + profile.id, + machineId, + catalog.profiles, + false, + paint, + ), + })} + ` + : nothing} + `; })} ` : nothing} diff --git a/ui/src/e2e/session-placement.move.e2e.test.ts b/ui/src/e2e/session-placement.move.e2e.test.ts index 5d18b44542c7..9e41874993eb 100644 --- a/ui/src/e2e/session-placement.move.e2e.test.ts +++ b/ui/src/e2e/session-placement.move.e2e.test.ts @@ -127,6 +127,70 @@ suite.define(() => { } }); + it.each([ + { machineId: "fast", expectedMachineClass: "fast" }, + { machineId: "standard", expectedMachineClass: undefined }, + ])( + "moves to a cloud profile with machine $machineId", + async ({ machineId, expectedMachineClass }) => { + const context = await suite.newBrowserContext(contextOptions()); + const page = await context.newPage(); + const gateway = await installMockGateway(page, { + featureMethods: ["chat.startup", "environments.list", "node.list", "sessions.move"], + historyMessages: [{ role: "assistant", content: "Placement machine proof." }], + methodResponses: { + "sessions.list": chatSessionListResponse([activeSession()]), + "environments.list": { + profiles: [ + { + id: "aws", + providerId: "crabbox", + trust: "disposable", + machines: [ + { id: "standard", label: "Standard", default: true }, + { id: "fast", label: "Fast" }, + ], + }, + ], + environments: [], + }, + "node.list": { nodes: [] }, + }, + sessionKey: "agent:main:placement-move", + }); + + try { + await page.goto(`${suite.server.baseUrl}chat`); + await gateway.deferNext("sessions.move"); + await page.getByRole("button", { name: "Runs on Cloud" }).click(); + await page.getByText("Move session…", { exact: true }).click(); + await page.locator('[data-value="cloud:aws"]').click(); + await page.locator(`[data-value="machine:${machineId}"]`).click(); + await page.getByRole("button", { name: "Move session", exact: true }).click(); + + const request = await gateway.waitForRequest("sessions.move"); + expect(request.params).toEqual({ + key: "agent:main:placement-move", + agentId: "main", + expected: { generation: 4, environmentId: "worker:source", ownerEpoch: 7 }, + target: { + kind: "profile", + profileId: "aws", + ...(expectedMachineClass ? { machineClass: expectedMachineClass } : {}), + }, + }); + await gateway.resolveDeferred("sessions.move", { + ok: true, + key: "agent:main:placement-move", + sessionId: "session-placement-move", + placement: { state: "active", generation: 10 }, + }); + } finally { + await suite.closeBrowserContext(context); + } + }, + ); + it("keeps a move failure visible and retryable", async () => { const context = await suite.newBrowserContext(contextOptions()); const page = await context.newPage(); diff --git a/ui/src/pages/chat/chat-pane-placement.test.ts b/ui/src/pages/chat/chat-pane-placement.test.ts index b463c5a08e7d..0a7a170c3aed 100644 --- a/ui/src/pages/chat/chat-pane-placement.test.ts +++ b/ui/src/pages/chat/chat-pane-placement.test.ts @@ -167,6 +167,64 @@ describe("chat pane placement", () => { expect(refreshReplacement).toHaveBeenCalledWith("main"); }); + it("moves an active placement to a selected profile machine", async () => { + const request = vi.fn(async (method: string) => { + if (method === "environments.list") { + return { + profiles: [ + { + id: "aws", + providerId: "crabbox", + machines: [ + { id: "standard", label: "Standard", default: true }, + { id: "beast", label: "Beast" }, + ], + }, + ], + environments: [], + }; + } + if (method === "node.list") { + return { nodes: [] }; + } + return { ok: true }; + }); + const refreshReplacement = vi.fn(async () => undefined); + const { pane } = createTestChatPane({ + client: { request } as unknown as GatewayBrowserClient, + sessions: { refreshReplacement } as unknown as SessionCapability, + }); + pane.context.gateway.snapshot.hello = { + features: { methods: ["sessions.move"] }, + auth: { role: "operator", scopes: ["operator.admin"] }, + } as never; + const session = activePlacementSession(); + + const moving = pane.moveHeaderPlacement(session); + await vi.waitFor(() => { + expect(document.body.querySelector('[data-value="cloud:aws"]')).not.toBeNull(); + }); + document.body.querySelector('[data-value="cloud:aws"]')?.click(); + document.body.querySelector('[data-value="machine:beast"]')?.click(); + const moveButton = [...document.body.querySelectorAll("button")].find( + (button) => button.textContent?.trim() === "Move session", + ); + moveButton?.click(); + await moving; + + expect(request).toHaveBeenCalledWith("sessions.move", { + key: session.key, + agentId: "main", + expected: { + generation: 1, + environmentId: "worker:one", + ownerEpoch: 1, + }, + target: { kind: "profile", profileId: "aws", machineClass: "beast" }, + }); + expect(refreshReplacement).toHaveBeenCalledWith("main"); + }); + it("does not reclaim when the operator cancels", async () => { const request = vi.fn(async () => ({ ok: true })); const { pane } = createTestChatPane({