#server.src.pixelweaver.command_applier

#server.src.pixelweaver.command_applier

Authoritative applier for WebSocket (browser) commands.

The browser forwards every dispatched command to the server over the WebSocket command message using the client command schema (see src/lib/core/command-params.generated.ts), which differs from the MCP tool schema (per-pixel RGBA + layerId rather than a single color + layer_id, x0/y0/x1/y1 rather than x1/y1/x2/y2, and so on).

Historically _handle_command only appended the command to history and broadcast it -- it never applied it to authoritative state, so every browser edit was thrown away (blank canvas on reload, blank exports). This module materializes the covered browser commands into the authoritative pixel and role buffers so the server truly is the source of truth.

Layer targeting: pencil/eraser strokes carry a real layerId; shape/fill and effect commands are dispatched with layerId: '' and rely on the app's active layer. The server tracks the active layer per canvas via the forwarded set_active_layer command and falls back to the first layer of the right kind when no active layer is known.

#CommandApplyError

A covered command failed to apply (bad parameters / target).

#_resolve_rgba_layer_id

def _resolve_rgba_layer_id(canvas: CanvasState, params: dict[str, Any]) -> str | None

Resolve the target rgba layer: explicit layerId else the active layer else the first rgba layer.

#apply_ws_command

def apply_ws_command(canvas: CanvasState, command: dict[str, Any], *, project: ProjectState | None=None) -> Any

Apply a browser command to canvas's authoritative state.

Returns one of: - False -- the command type is not covered (the caller records it in history without a server-side mutation). - COVERED_NO_UNDO -- covered and applied but with no undo snapshot (e.g. set_active_layer). - a dict -- covered and applied, carrying a before/after undo record.

Raises CommandApplyError if a covered command cannot be applied.

project is the owning ProjectState; it is required for the project-scoped metadata commands (manifest / part registry / part-set / iso). Pixel/role/structural commands operate on canvas alone.

#_resolve_layer

def _resolve_layer(canvas: CanvasState, params: dict[str, Any]) -> dict[str, Any]

Resolve a target layer dict by the client id param (hard error).

Searches the whole tree so a nested (grouped) layer resolves too.

#_apply_structural

def _apply_structural(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> Any

Apply a covered structural command using the shared structural_ops path.

Translates the client (camelCase) command schema onto the schema-neutral structural_ops functions and returns their structural undo record.

#_apply_frame_multi

def _apply_frame_multi(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> Any

Apply paste_frame / delete_selected_frames as a batched multi-frame op.

Returns a composite frame_batch undo record (or COVERED_NO_UNDO when the embedded work is empty -- an empty clipboard or an empty selection is a client no-op, not a silent drop).

#_apply_paste_frame

def _apply_paste_frame(canvas: CanvasState, params: dict[str, Any]) -> Any

Insert the client-embedded copied frames after the insertion index.

Each embedded frame carries a client-minted id and its per-layer pixel buffers (base64). Mirror of the client paste_frame: frame i is inserted after afterIndex + i (so the copied run lands contiguously after the current frame). Empty frames empty clipboard no-op.

#_apply_delete_selected_frames

def _apply_delete_selected_frames(canvas: CanvasState, params: dict[str, Any]) -> Any

Remove the client-embedded target frames (by id).

Mirror of the client delete_selected_frames + frameModel.removeFrames: the last frame is never removed (if the targets cover every frame, the lowest-index one survives). Empty selection / single-frame canvas == no-op.

#_merge_layer_into_below

def _merge_layer_into_below(canvas: CanvasState, top: dict[str, Any]) -> str

Composite top onto the rgba pixel layer directly below it IN THE SAME PARENT (per frame, honoring top's visibility/opacity/blend), drop top's buffers, remove top from its sibling list, and repoint the active layer when it was top. Returns the below layer id.

Tree-aware faithful port of the client layerTree.mergeDown: the "below" is resolved via findLayerContext (the sibling directly below within the same enclosing group), NOT the flat root stack -- so merging a nested leaf works. A hidden top contributes nothing but is still removed; a missing bottom buffer starts transparent.

#_apply_layer_composite

def _apply_layer_composite(canvas: CanvasState, cmd_type: str) -> Any

Apply merge_down / flatten_image with true source-over compositing.

Both are tree-aware (a nested leaf resolves its sibling-below correctly). Whole-layer-tree snapshot undo (structure + pixels restore together).

#_clone_subtree

def _clone_subtree(layer: dict[str, Any], id_map: dict[str, str]) -> dict[str, Any]

Deep-copy layer, remapping every node id via id_map (source id -> clone id). Mirrors the client deepCloneLayer which mints a fresh id for every node; the client embeds the mapping so both trees share identities.

#_apply_layer_tree

def _apply_layer_tree(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> Any

Apply a tree-restructuring command with a whole-layer-tree snapshot undo.

All four honor client-embedded object ids so the server tree's identities match the browser's (buffers keyed by id resolve; later commands referencing the ids apply). Snapshot before + after => a single layers_snapshot record reverts structure AND per-frame buffers together.

#_apply_duplicate_layer

def _apply_duplicate_layer(canvas: CanvasState, params: dict[str, Any]) -> None

Clone a leaf/group subtree above the source, WITH its per-frame buffers.

The client embeds idMap (every source node id -> clone id) and newId (the top clone). Buffers are copied per frame for every mapped leaf so the duplicate carries the source's pixels (the client copies them too -- see the duplicate_layer command). Mirror of the client duplicateLayer: clone is inserted directly above the source, its name gets a " copy" suffix, and the active layer moves to the clone when it is a pixel leaf.

#_apply_flatten_group

def _apply_flatten_group(canvas: CanvasState, params: dict[str, Any]) -> None

Composite a group's rgba subtree into a single replacement pixel leaf.

The client embeds the replacement id as newId. Mirror of the client flattenGroup + its composite loop: the replacement takes the group's position and its visible/opacity/blend, the group's children are composited per frame (faithful layer_composite tree walk) into a buffer stored under the replacement id, and the descendant buffers are dropped.

#_reorder_frame_order

def _reorder_frame_order(canvas: CanvasState, from_index: int, to_index: int) -> list[str]

Target id order for a single-frame move (mirrors frameModel.reorderFrame).

#_reorder_frames_order

def _reorder_frames_order(canvas: CanvasState, from_indices: list[int], to_index: int) -> list[str]

Target id order for a bulk move (mirrors frameModel.reorderFrames).

#_apply_rgba_embedded

def _apply_rgba_embedded(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply a clipboard/selection pixel op from its embedded pixels list.

cut / delete_selection clear the selection (each pixel carries a=0), paste writes the clipboard contents, move_selection carries the merged final state (cleared sources + moved destinations). Every case is a flat list of absolute per-pixel RGBA writes the client stabilized into params, so the server just replays them.

#_apply_rgba_primitive

def _apply_rgba_primitive(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply a self-contained drawing-primitive command.

These carry every input in params; the server reproduces the client algorithm (plugins/builtin/drawing-primitives-plugin.ts and advanced-fill-tool.ts) against its own authoritative buffer.

#_apply_copy_region

def _apply_copy_region(buffer: _Buffer, params: dict[str, Any]) -> None

Copy a rectangular region (reads all source pixels before writing, so overlapping src/dst regions are handled -- mirrors the client).

#_apply_replace_color

def _apply_replace_color(buffer: _Buffer, params: dict[str, Any]) -> None

Scan the whole buffer, replacing every exact-RGBA match of oldColor with newColor (mirrors the client).

#_apply_dimension

def _apply_dimension(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply a whole-canvas dimension command with a full before/after undo.

rotate/scale/crop hard-error while any role-kind layer exists (mirrors the client's assertRgbaPixelLayers guard); resize remaps role planes and is allowed with role layers.

#_apply_layer_transform

def _apply_layer_transform(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> Any

Apply the frame-tier rotate / scale effect (single active layer).

Faithful mirror of plugins/builtin/effects/{rotate,scale}.ts: transform ONLY the active layer's current-frame buffer, then resize the whole canvas to the result dims. If the active layer is a role layer, the client's getActiveBuffer returns null and the effect is a no-op (COVERED_NO_UNDO). Undo is a whole- canvas snapshot so the resize reverts.

#_apply_rgba_computed

def _apply_rgba_computed(canvas: CanvasState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply a computed drawing tool (gradient/noise/dither/pattern-stamp).

Reproduces the client algorithm in ws_drawing and overwrites the target pixels, matching the browser's local result byte-for-byte.

#_apply_frame_tag

def _apply_frame_tag(project: ProjectState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply add_tag / remove_tag via frame_tag_ops (meta_snapshot undo).

add_tag replays the client-minted id + color (embedded in params); remove_tag removes by id. Invalid input (bad range, missing parent, unknown id) is a hard CommandApplyError, never a silent no-op.

#_apply_level

def _apply_level(project: ProjectState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply a covered level-editor command via level_ops (meta_snapshot undo).

Translates the client (camelCase) command schema onto the schema-neutral level_ops functions. The three placement commands forward the target layer's name (layerName) so a materialized layer round-trips faithfully; the object-creating ones forward the client-minted object id (id). Invalid input / missing target is a hard CommandApplyError (apply-or-reject).

#_apply_variant

def _apply_variant(project: ProjectState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply a covered variant-preset command via variant_ops (meta_snapshot).

Translates the client (camelCase) command schema onto the schema-neutral variant_ops functions. add_variant_preset replays the client-minted id; generate_variants replays the client-computed presets embedded in createdPresets (a valid empty list is a no-op the client still forwards). Invalid input / missing preset is a hard CommandApplyError (apply-or-reject).

#_apply_metadata

def _apply_metadata(project: ProjectState, cmd_type: str, params: dict[str, Any]) -> dict[str, Any]

Apply a covered project/canvas metadata command via metadata_ops.

Translates the client (camelCase) command schema onto the schema-neutral metadata_ops functions and returns their meta-snapshot undo record. All validation lives in the state model's validated setters; invalid input is a hard CommandApplyError, never a silent partial write.

#_apply_create_part_family

def _apply_create_part_family(project: ProjectState, params: dict[str, Any]) -> dict[str, Any]

Apply create_part_family via the shared structural_ops creation path.

The browser stabilizes the created canvas instance ids into canvasIds; the server keys canvases by the deterministic <partSet>.<i> name, so the ids ride the record as identity metadata. Returns the canvases_added undo record so undo removes the family and redo re-adds it with identical content.

#_apply_manifest_transform

def _apply_manifest_transform(project: ProjectState, params: dict[str, Any]) -> dict[str, Any]

Apply apply_manifest_transform via the shared metadata_ops transform path.

Self-contained params: manifest (the full new manifest) + remaps (per-class old->new role maps). Returns the compound undo record so undo reverts BOTH the manifest and the remapped role bytes.

#_apply_character

def _apply_character(project: ProjectState, cmd_type: str, params: dict[str, Any]) -> Any

Apply a covered character-preview command via character_ops.

Translates the client command schema onto the schema-neutral character_ops functions and returns their meta-snapshot undo record. Validation lives in ProjectState.set_character_preview; invalid input is a hard error.