feat: add durable QEMU create update delete

README.md

@@ -10,9 +10,16 @@ limits are recorded in [the 0.1.0 compatibility report](docs/compatibility-0.1.0
 The bundled PVE 9.2.3 declared contract contains 444 paths and 675 methods.
 Implemented semantics currently include version, ticket login, node listing and
 status, cluster resources, basic QEMU list/config/status/start/stop, and task
-list/status/log, plus API-token list/create/read/update/delete. Mutations require
-the ticket-bound CSRF header and execute through PostgreSQL-leased workers; all
-other declared methods return an explicit unsupported error.
+list/status/log, QEMU create/sync-update/async-update/delete, plus API-token
+list/create/read/update/delete. Mutations require the ticket-bound CSRF header
+and execute through PostgreSQL-leased workers; all other declared methods return
+an explicit unsupported error.
+
+QEMU create, asynchronous config update, and delete return durable UPIDs and use
+the same PostgreSQL resource lock as lifecycle operations. Synchronous config
+PUT uses optimistic versioning. Common fields and unknown version-dependent
+parameters are retained in JSONB; duplicate VMIDs and overlapping operations
+fail with 409.
 
 ## Development
 

app/api/registry.py

@@ -14,7 +14,7 @@ from app.api.errors import ApiError, ContractValidationError
 from app.config import Settings
 from app.contracts.model import Method, Schema, Snapshot
 from app.db.pool import AsyncpgDatabase
-from app.security.acl import AclEntry, authorize, requirement_from_contract
+from app.security.acl import AclEntry, CapabilityRequirement, authorize, requirement_from_contract
 from app.security.auth import parse_api_token, verify_csrf, verify_secret, verify_ticket
 
 Handler = Callable[[Request, dict[str, Any]], Awaitable[Any]]
@@ -159,13 +159,14 @@ async def _authenticate(
     if principal == "root@pam" and token_privileges is None:
         return
     database = cast(AsyncpgDatabase, request.app.state.database)
-    await _authorize(database, principal, token_privileges, method, inputs)
+    await _authorize(database, principal, token_privileges, semantic_path, method, inputs)
 
 
 async def _authorize(
     database: AsyncpgDatabase,
     principal: str,
     token_privileges: frozenset[str] | None,
+    semantic_path: str,
     method: Method,
     inputs: dict[str, Any],
 ) -> None:
@@ -173,6 +174,8 @@ async def _authorize(
     requirement = requirement_from_contract(
         method.permissions, {name: str(value) for name, value in values.items()}
     )
+    if requirement is None and semantic_path == "/nodes/{node}/qemu" and method.verb == "POST":
+        requirement = CapabilityRequirement(f"/vms/{values['vmid']}", frozenset({"VM.Allocate"}))
     if requirement is None:
         return
     rows = await database.pool.fetch(
@@ -201,6 +204,7 @@ async def _authorize(
         requirement.privileges,
         entries,
         token_privileges=token_privileges,
+        require_all=requirement.require_all,
     ):
         raise ApiError(403, "permission check failed")
 
@@ -241,7 +245,11 @@ async def _parse_inputs(request: Request, method: Method) -> dict[str, Any]:
             errors[name] = "property is not defined in schema"
     if errors:
         raise ContractValidationError(dict(sorted(errors.items())))
-    return {"values": parsed, "path": dict(request.path_params)}
+    return {
+        "values": parsed,
+        "path": dict(request.path_params),
+        "provided": tuple(sorted(supplied)),
+    }
 
 
 def _coerce(value: Any, schema: Schema) -> Any:

app/handlers/access.py

@@ -101,6 +101,7 @@ def register_access_handlers(registry: HandlerRegistry) -> None:
 
     async def token_update(request: Request, inputs: dict[str, Any]) -> dict[str, Any]:
         values = _values(inputs)
+        provided = frozenset(str(item) for item in inputs.get("provided", values))
         userid, tokenid = str(values["userid"]), str(values["tokenid"])
         _require_owner(request, userid)
         regenerate = bool(values.get("regenerate", False))
@@ -117,9 +118,9 @@ def register_access_handlers(registry: HandlerRegistry) -> None:
                 extract(epoch from t.expires_at)::bigint AS expire""",
             userid,
             tokenid,
-            values.get("comment"),
-            _expire_value(values),
-            values.get("privsep"),
+            values.get("comment") if "comment" in provided else None,
+            _expire_value(values) if "expire" in provided else None,
+            values.get("privsep") if "privsep" in provided else None,
             hash_secret(secret) if secret is not None else None,
         )
         if row is None:

app/handlers/qemu.py

@@ -11,6 +11,7 @@ from fastapi import Request
 from app.api.errors import ApiError
 from app.api.registry import HandlerRegistry
 from app.db.pool import AsyncpgDatabase
+from app.db.primitives import ConflictError
 from app.tasks.repository import TaskRepository
 from app.tasks.upid import Upid
 
@@ -43,14 +44,16 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None:
     async def qemu_config(request: Request, inputs: dict[str, Any]) -> dict[str, Any]:
         node, vmid = str(_values(inputs)["node"]), str(_values(inputs)["vmid"])
         row = await _database(request).pool.fetchrow(
-            """SELECT r.state FROM resources r JOIN nodes n ON n.id=r.node_id
+            """SELECT r.state, v.config FROM resources r
+            JOIN nodes n ON n.id=r.node_id
+            JOIN virtual_machines v ON v.resource_id=r.id
             WHERE n.name=$1 AND r.kind='qemu' AND r.external_id=$2""",
             node,
             vmid,
         )
         if row is None:
             raise ApiError(404, "virtual machine does not exist")
-        return {"vmid": int(vmid), **_state(row["state"])}
+        return {"vmid": int(vmid), **_state(row["config"]), **_state(row["state"])}
 
     async def qemu_status(request: Request, inputs: dict[str, Any]) -> dict[str, Any]:
         return await qemu_config(request, inputs)
@@ -74,7 +77,17 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None:
             raise ApiError(409, f"cannot {operation} VM while it is {current}")
         timestamp = int(await database.pool.fetchval("SELECT extract(epoch from now())::bigint"))
         pid = int(await database.pool.fetchval("SELECT pg_backend_pid()"))
-        upid = str(Upid(node, pid, pid, timestamp, f"qm{operation}", vmid, "root@pam"))
+        upid = str(
+            Upid(
+                node,
+                pid,
+                pid,
+                timestamp,
+                f"qm{operation}",
+                vmid,
+                str(request.state.principal),
+            )
+        )
         task = await TaskRepository(database.pool).create(
             upid=upid,
             task_type=f"qemu-{operation}",
@@ -84,6 +97,123 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None:
         )
         return task.upid
 
+    async def create(request: Request, inputs: dict[str, Any]) -> str:
+        values = _values(inputs)
+        node, vmid = str(values["node"]), int(values["vmid"])
+        database = _database(request)
+        if not await database.pool.fetchval(
+            "SELECT EXISTS(SELECT 1 FROM nodes WHERE name=$1)", node
+        ):
+            raise ApiError(404, "node does not exist")
+        if await database.pool.fetchval(
+            """SELECT EXISTS(SELECT 1 FROM resources
+            WHERE external_id=$1 AND kind IN ('qemu','lxc'))""",
+            str(vmid),
+        ):
+            raise ApiError(409, "VMID already exists")
+        config = {
+            key: value
+            for key, value in values.items()
+            if key not in {"node", "vmid", "force", "archive", "start"}
+        }
+        return await _create_task(
+            request,
+            node=node,
+            vmid=str(vmid),
+            task_type="qemu-create",
+            payload={"node": node, "vmid": vmid, "config": config},
+        )
+
+    async def update(request: Request, inputs: dict[str, Any], *, asynchronous: bool) -> str | None:
+        values = _values(inputs)
+        node, vmid = str(values["node"]), str(values["vmid"])
+        database = _database(request)
+        row = await database.pool.fetchrow(
+            """SELECT r.id, r.version, r.state, v.config FROM resources r
+            JOIN nodes n ON n.id=r.node_id
+            JOIN virtual_machines v ON v.resource_id=r.id
+            WHERE n.name=$1 AND r.kind='qemu' AND r.external_id=$2""",
+            node,
+            vmid,
+        )
+        if row is None:
+            raise ApiError(404, "virtual machine does not exist")
+        control = {"node", "vmid", "digest", "delete", "revert", "skiplock", "background_delay"}
+        provided = frozenset(str(item) for item in inputs.get("provided", values))
+        changes = {
+            key: value for key, value in values.items() if key in provided and key not in control
+        }
+        delete = str(values.get("delete", "")) if "delete" in provided else ""
+        if asynchronous:
+            return await _create_task(
+                request,
+                node=node,
+                vmid=vmid,
+                task_type="qemu-update",
+                payload={
+                    "node": node,
+                    "vmid": vmid,
+                    "resource_id": str(row["id"]),
+                    "changes": changes,
+                    "delete": delete,
+                },
+            )
+        state = _state(row["state"])
+        config = _state(row["config"])
+        state.update(changes)
+        config.update(changes)
+        for key in delete.split(","):
+            if key:
+                state.pop(key, None)
+                config.pop(key, None)
+        status = await database.pool.execute(
+            """UPDATE resources SET state=$3::jsonb, version=version+1,
+            updated_at=now() WHERE id=$1 AND version=$2""",
+            row["id"],
+            row["version"],
+            json.dumps(state, sort_keys=True),
+        )
+        if status != "UPDATE 1":
+            raise ApiError(409, "configuration changed concurrently")
+        await database.pool.execute(
+            """UPDATE virtual_machines SET config=$2::jsonb
+            WHERE resource_id=$1""",
+            row["id"],
+            json.dumps(config, sort_keys=True),
+        )
+        return None
+
+    async def update_async(request: Request, inputs: dict[str, Any]) -> str:
+        result = await update(request, inputs, asynchronous=True)
+        if not isinstance(result, str):
+            raise RuntimeError("async QEMU update did not create a task")
+        return result
+
+    async def update_sync(request: Request, inputs: dict[str, Any]) -> None:
+        await update(request, inputs, asynchronous=False)
+
+    async def delete(request: Request, inputs: dict[str, Any]) -> str:
+        values = _values(inputs)
+        node, vmid = str(values["node"]), str(values["vmid"])
+        database = _database(request)
+        row = await database.pool.fetchrow(
+            """SELECT r.id, r.state FROM resources r JOIN nodes n ON n.id=r.node_id
+            WHERE n.name=$1 AND r.kind='qemu' AND r.external_id=$2""",
+            node,
+            vmid,
+        )
+        if row is None:
+            raise ApiError(404, "virtual machine does not exist")
+        if str(_state(row["state"]).get("status")) != "stopped":
+            raise ApiError(409, "cannot delete a running virtual machine")
+        return await _create_task(
+            request,
+            node=node,
+            vmid=vmid,
+            task_type="qemu-delete",
+            payload={"node": node, "vmid": vmid, "resource_id": str(row["id"])},
+        )
+
     async def start(request: Request, inputs: dict[str, Any]) -> str:
         return await mutate("start", request, inputs)
 
@@ -124,10 +254,44 @@ def register_qemu_handlers(registry: HandlerRegistry) -> None:
         ]
 
     registry.register("/nodes/{node}/qemu", "GET", qemu_list)
+    registry.register("/nodes/{node}/qemu", "POST", create)
+    registry.register("/nodes/{node}/qemu/{vmid}", "DELETE", delete)
     registry.register("/nodes/{node}/qemu/{vmid}/config", "GET", qemu_config)
+    registry.register("/nodes/{node}/qemu/{vmid}/config", "POST", update_async)
+    registry.register("/nodes/{node}/qemu/{vmid}/config", "PUT", update_sync)
     registry.register("/nodes/{node}/qemu/{vmid}/status/current", "GET", qemu_status)
     registry.register("/nodes/{node}/qemu/{vmid}/status/start", "POST", start)
     registry.register("/nodes/{node}/qemu/{vmid}/status/stop", "POST", stop)
     registry.register("/nodes/{node}/tasks", "GET", task_list)
     registry.register("/nodes/{node}/tasks/{upid}/status", "GET", task_status)
     registry.register("/nodes/{node}/tasks/{upid}/log", "GET", task_log)
+
+
+async def _create_task(
+    request: Request,
+    *,
+    node: str,
+    vmid: str,
+    task_type: str,
+    payload: dict[str, Any],
+) -> str:
+    database = _database(request)
+    timestamp = int(await database.pool.fetchval("SELECT extract(epoch from now())::bigint"))
+    pid = int(await database.pool.fetchval("SELECT pg_backend_pid()"))
+    worker_type = {
+        "qemu-create": "qmcreate",
+        "qemu-delete": "qmdestroy",
+        "qemu-update": "qmconfig",
+    }[task_type]
+    upid = str(Upid(node, pid, pid, timestamp, worker_type, vmid, str(request.state.principal)))
+    try:
+        task = await TaskRepository(database.pool).create(
+            upid=upid,
+            task_type=task_type,
+            payload=payload,
+            resource_key=f"qemu:{vmid}",
+            idempotency_key=request.headers.get("Idempotency-Key"),
+        )
+    except ConflictError as error:
+        raise ApiError(409, str(error)) from error
+    return task.upid

app/main.py

@@ -43,7 +43,13 @@ def create_app(
             return TaskWorker(
                 repository,
                 "simulator-worker",
-                {"qemu-start": handler, "qemu-stop": handler},
+                {
+                    "qemu-create": handler,
+                    "qemu-delete": handler,
+                    "qemu-start": handler,
+                    "qemu-stop": handler,
+                    "qemu-update": handler,
+                },
                 concurrency=resolved.task_worker_concurrency,
                 lease_seconds=resolved.task_lease_seconds,
             )

app/security/acl.py

@@ -57,17 +57,19 @@ def authorize(
     entries: tuple[AclEntry, ...],
     *,
     token_privileges: frozenset[str] | None = None,
+    require_all: bool = True,
 ) -> bool:
     privileges = effective_privileges(principal, path, entries)
     if token_privileges is not None:
         privileges &= token_privileges
-    return required <= privileges
+    return required <= privileges if require_all else bool(required & privileges)
 
 
 @dataclass(frozen=True, slots=True)
 class CapabilityRequirement:
     path: str
     privileges: frozenset[str]
+    require_all: bool = True
 
 
 def requirement_from_contract(
@@ -84,4 +86,7 @@ def requirement_from_contract(
     raw_privileges = check[2]
     if not isinstance(raw_privileges, list):
         return None
-    return CapabilityRequirement(raw_path, frozenset(str(item) for item in raw_privileges))
+    require_all = not (len(check) >= 4 and check[3] == "any")
+    return CapabilityRequirement(
+        raw_path, frozenset(str(item) for item in raw_privileges), require_all
+    )

app/tasks/qemu.py

@@ -4,6 +4,8 @@ from __future__ import annotations
 
 import json
 import uuid
+from collections.abc import Mapping
+from typing import Any, cast
 
 from app.simulation.clock import Clock
 from app.simulation.transitions import VmState, plan_transition
@@ -12,9 +14,15 @@ from app.tasks.worker import TaskHandler
 
 
 def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler:
-    async def execute(task: Task) -> dict[str, str]:
+    async def execute(task: Task) -> dict[str, Any]:
         operation = task.task_type.removeprefix("qemu-")
+        if operation == "create":
+            return await _create(repository, task)
         resource_id = uuid.UUID(str(task.payload["resource_id"]))
+        if operation == "update":
+            return await _update(repository, task, resource_id)
+        if operation == "delete":
+            return await _delete(repository, task, resource_id)
         async with repository.pool.acquire() as connection:
             row = await connection.fetchrow("SELECT state FROM resources WHERE id=$1", resource_id)
             if row is None:
@@ -41,3 +49,85 @@ def qemu_handler(repository: TaskRepository, clock: Clock) -> TaskHandler:
         return {"status": str(transition.after)}
 
     return execute
+
+
+async def _create(repository: TaskRepository, task: Task) -> dict[str, Any]:
+    node, vmid = str(task.payload["node"]), int(task.payload["vmid"])
+    config = dict(task.payload.get("config", {}))
+    resource_id = uuid.uuid4()
+    state = {"status": "stopped", **config}
+    async with repository.pool.acquire() as connection:
+        async with connection.transaction():
+            node_row = await connection.fetchrow(
+                "SELECT id, cluster_id FROM nodes WHERE name=$1", node
+            )
+            if node_row is None:
+                raise ValueError("node disappeared")
+            await connection.execute(
+                """INSERT INTO resources(
+                    id, node_id, cluster_id, kind, external_id, state, metadata
+                ) VALUES($1, $2, $3, 'qemu', $4, $5::jsonb, '{}'::jsonb)""",
+                resource_id,
+                node_row["id"],
+                node_row["cluster_id"],
+                str(vmid),
+                json.dumps(state, sort_keys=True),
+            )
+            await connection.execute(
+                """INSERT INTO virtual_machines(resource_id, cluster_id, vmid, config)
+                VALUES($1, $2, $3, $4::jsonb)""",
+                resource_id,
+                node_row["cluster_id"],
+                vmid,
+                json.dumps(config, sort_keys=True),
+            )
+    await repository.append_log(task.id, f"VM {vmid} created")
+    return {"vmid": vmid, "status": "stopped"}
+
+
+async def _update(repository: TaskRepository, task: Task, resource_id: uuid.UUID) -> dict[str, Any]:
+    changes = dict(task.payload.get("changes", {}))
+    delete_keys = tuple(str(task.payload.get("delete", "")).split(","))
+    async with repository.pool.acquire() as connection:
+        async with connection.transaction():
+            row = await connection.fetchrow(
+                """SELECT r.state, v.config FROM resources r
+                JOIN virtual_machines v ON v.resource_id=r.id WHERE r.id=$1""",
+                resource_id,
+            )
+            if row is None:
+                raise ValueError("resource disappeared")
+            state = _object(row["state"])
+            config = _object(row["config"])
+            config.update(changes)
+            for key in delete_keys:
+                if key:
+                    config.pop(key, None)
+                    state.pop(key, None)
+            state.update(changes)
+            await connection.execute(
+                """UPDATE resources SET state=$2::jsonb, version=version+1,
+                updated_at=now() WHERE id=$1""",
+                resource_id,
+                json.dumps(state, sort_keys=True),
+            )
+            await connection.execute(
+                "UPDATE virtual_machines SET config=$2::jsonb WHERE resource_id=$1",
+                resource_id,
+                json.dumps(config, sort_keys=True),
+            )
+    await repository.append_log(task.id, "VM configuration updated")
+    return {"updated": sorted(changes), "deleted": sorted(key for key in delete_keys if key)}
+
+
+async def _delete(repository: TaskRepository, task: Task, resource_id: uuid.UUID) -> dict[str, Any]:
+    async with repository.pool.acquire() as connection:
+        status = await connection.execute("DELETE FROM resources WHERE id=$1", resource_id)
+        if status != "DELETE 1":
+            raise ValueError("resource disappeared")
+    await repository.append_log(task.id, "VM deleted")
+    return {"deleted": True}
+
+
+def _object(value: object) -> dict[str, Any]:
+    return json.loads(value) if isinstance(value, str) else dict(cast(Mapping[str, Any], value))

docs/compatibility-0.1.0.md

@@ -9,8 +9,8 @@ Proxmox compatibility.
 | Level | Methods | Contract share | Evidence |
 |---|---:|---:|---|
 | Declared and dynamically routed | 675 | 100% | Imported immutable API Viewer artifact |
-| Stateful semantics implemented on current main | 18 | 2.67% | Handler registry and unit/integration tests |
-| Schema-only or explicitly unsupported | 657 | 97.33% | Default 501 fallback |
+| Stateful semantics implemented on current main | 22 | 3.26% | Handler registry and unit/integration tests |
+| Schema-only or explicitly unsupported | 653 | 96.74% | Default 501 fallback |
 | proxmoxer smoke exercised | 9 | 1.33% | Unmodified proxmoxer 2.3 compatibility test |
 
 The smoke set is `POST /access/ticket`, `GET /version`, `GET /nodes`,

docs/original-prompt-gap-plan.md

@@ -55,7 +55,8 @@ plaintext password, ticket, CSRF token, or token secret reaches storage/logs.
 
 ## G4 — QEMU 0.2 verticals
 
-- [ ] Create, update, delete, shutdown, reboot, reset, suspend and resume.
+- [x] Create, synchronous/asynchronous update, and delete.
+- [ ] Shutdown, reboot, reset, suspend and resume.
 - [ ] Snapshots and rollback, clone, local/remote migration, resize and move
   disk, selected agent endpoints, pending/status data.
 - [ ] Persist normalized CPU/memory/common fields plus unknown PVE parameters in

evidence/pve-9.2.3-0.1.0.json

@@ -50,6 +50,30 @@
       "verb": "GET",
       "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "response_field_types", "response_required_fields", "long_task_behavior", "permissions"],
       "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_task_worker.py", "tests/unit/test_upid.py"]
+    },
+    {
+      "path": "/nodes/{node}/qemu",
+      "verb": "POST",
+      "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"],
+      "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py"]
+    },
+    {
+      "path": "/nodes/{node}/qemu/{vmid}",
+      "verb": "DELETE",
+      "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"],
+      "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py"]
+    },
+    {
+      "path": "/nodes/{node}/qemu/{vmid}/config",
+      "verb": "POST",
+      "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "long_task_behavior", "errors_prohibitions", "permissions"],
+      "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_qemu_task.py", "tests/unit/test_acl.py"]
+    },
+    {
+      "path": "/nodes/{node}/qemu/{vmid}/config",
+      "verb": "PUT",
+      "dimensions": ["input_parameters", "parameter_requiredness", "types_constraints", "http_status", "json_structure", "state_semantics", "errors_prohibitions", "permissions"],
+      "sources": ["tests/compatibility/test_proxmoxer.py", "tests/unit/test_qemu_handlers.py", "tests/unit/test_acl.py"]
     }
   ]
 }

tests/compatibility/test_proxmoxer.py

@@ -2,6 +2,7 @@
 
 import os
 from threading import Event
+from typing import Any, cast
 
 import pytest
 from proxmoxer import ProxmoxAPI, ResourceException  # type: ignore[import-untyped]
@@ -12,6 +13,15 @@ pytestmark = [
 ]
 
 
+def wait_task(proxmox: Any, upid: str) -> dict[str, object]:
+    for _attempt in range(100):
+        task = proxmox.nodes("pve1").tasks(upid).status.get()
+        if task["status"] == "stopped":
+            return cast(dict[str, object], task)
+        Event().wait(0.05)
+    raise AssertionError("task did not finish")
+
+
 def test_proxmoxer_read_and_qemu_task_flow() -> None:
     proxmox = ProxmoxAPI(
         os.environ["PROXMOXER_HOST"],
@@ -83,6 +93,29 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None:
             storage_api.nodes("pve1").qemu(vmid).config.get()
         assert hidden.value.status_code == 403
 
+    create_upid = proxmox.nodes("pve1").qemu.post(
+        vmid=150, name="created-by-proxmoxer", cores=2, memory=1024
+    )
+    with pytest.raises(ResourceException) as duplicate_create:
+        proxmox.nodes("pve1").qemu.post(vmid=150, name="duplicate")
+    assert duplicate_create.value.status_code == 409
+    assert wait_task(proxmox, create_upid)["exitstatus"] == "OK"
+    created_config = proxmox.nodes("pve1").qemu("150").config.get()
+    assert created_config["name"] == "created-by-proxmoxer"
+    assert created_config["cores"] == 2
+
+    assert proxmox.nodes("pve1").qemu("150").config.put(name="sync-update", cores=3) is None
+    assert proxmox.nodes("pve1").qemu("150").config.get()["name"] == "sync-update"
+    update_upid = proxmox.nodes("pve1").qemu("150").config.post(name="async-update", memory=2048)
+    assert wait_task(proxmox, update_upid)["exitstatus"] == "OK"
+    assert proxmox.nodes("pve1").qemu("150").config.get()["name"] == "async-update"
+
+    delete_upid = proxmox.nodes("pve1").qemu("150").delete()
+    assert wait_task(proxmox, delete_upid)["exitstatus"] == "OK"
+    with pytest.raises(ResourceException) as deleted_vm:
+        proxmox.nodes("pve1").qemu("150").config.get()
+    assert deleted_vm.value.status_code == 404
+
     if os.getenv("PROXMOXER_MUTATION_TEST") == "1":
         operator_api = ProxmoxAPI(
             os.environ["PROXMOXER_HOST"],
@@ -96,10 +129,5 @@ def test_proxmoxer_read_and_qemu_task_flow() -> None:
         operation = "start" if status["status"] == "stopped" else "stop"
         endpoint = operator_api.nodes("pve1").qemu("101").status(operation)
         upid = endpoint.post()
-        for _attempt in range(100):
-            task = operator_api.nodes("pve1").tasks(upid).status.get()
-            if task["status"] == "stopped":
-                break
-            Event().wait(0.05)
-        assert task["status"] == "stopped"
+        task = wait_task(operator_api, upid)
         assert task["exitstatus"] == "OK"

tests/unit/test_acl.py

@@ -43,3 +43,19 @@ def test_contract_permission_maps_to_capability_requirement() -> None:
     assert requirement is not None
     assert requirement.path == "/vms/100"
     assert requirement.privileges == frozenset({"VM.PowerMgmt"})
+
+    any_permission = Permissions(
+        expression={
+            "check": ["perm", "/vms/{vmid}", ["VM.Config.CPU", "VM.Config.Memory"], "any", 1]
+        }
+    )
+    any_requirement = requirement_from_contract(any_permission, {"vmid": "100"})
+    assert any_requirement is not None
+    assert not any_requirement.require_all
+    assert authorize(
+        "alice@pve",
+        "/vms/100",
+        any_requirement.privileges,
+        (AclEntry("alice@pve", "/vms", frozenset({"VM.Config.CPU"})),),
+        require_all=any_requirement.require_all,
+    )

tests/unit/test_core_handlers.py

@@ -29,7 +29,10 @@ class FakePool:
                     "id": uuid.UUID("00000000-0000-0000-0000-000000000100"),
                     "state": '{"name":"demo","status":"stopped"}',
                 }
-            return {"state": '{"name":"demo","status":"stopped"}'}
+            return {
+                "config": '{"name":"demo"}',
+                "state": '{"name":"demo","status":"stopped"}',
+            }
         return None
 
     async def fetch(self, sql: str, *args: object) -> list[dict[str, object]]:

tests/unit/test_qemu_handlers.py

@@ -0,0 +1,187 @@
+"""Persistent QEMU CRUD semantic handler tests."""
+
+import uuid
+from typing import Any, cast
+
+import pytest
+from fastapi import FastAPI, Request
+
+from app.api.errors import ApiError
+from app.api.registry import HandlerRegistry
+from app.db.pool import AsyncpgDatabase
+from app.db.primitives import ConflictError
+from app.handlers.qemu import register_qemu_handlers
+from app.tasks.repository import Task
+
+
+class QemuPool:
+    def __init__(self) -> None:
+        self.resource_exists = False
+        self.missing = False
+        self.running = False
+        self.commands: list[str] = []
+        self.resource_id = uuid.uuid4()
+
+    async def fetchval(self, sql: str, *args: object) -> bool | int:
+        del args
+        if "pg_backend_pid" in sql:
+            return 123
+        if "extract(epoch" in sql:
+            return 1_700_000_000
+        if "FROM nodes" in sql:
+            return True
+        if "FROM resources" in sql:
+            return self.resource_exists
+        raise AssertionError(sql)
+
+    async def fetchrow(self, sql: str, *args: object) -> dict[str, object] | None:
+        del args
+        if self.missing:
+            return None
+        if "SELECT r.id, r.version" in sql:
+            return {
+                "id": self.resource_id,
+                "version": 1,
+                "state": '{"name":"old","status":"stopped"}',
+                "config": '{"name":"old"}',
+            }
+        if "SELECT r.id, r.state" in sql:
+            status = "running" if self.running else "stopped"
+            return {"id": self.resource_id, "state": f'{{"status":"{status}"}}'}
+        raise AssertionError(sql)
+
+    async def execute(self, sql: str, *args: object) -> str:
+        del args
+        self.commands.append(sql)
+        return "UPDATE 1"
+
+
+class FakeDatabase:
+    def __init__(self, pool: QemuPool) -> None:
+        self.pool = pool
+
+
+def request(pool: QemuPool) -> Request:
+    app = FastAPI()
+    app.state.database = cast(AsyncpgDatabase, FakeDatabase(pool))
+    result = Request(
+        {
+            "type": "http",
+            "app": app,
+            "method": "POST",
+            "path": "/",
+            "headers": [],
+            "query_string": b"",
+            "server": ("test", 80),
+            "client": ("test", 123),
+            "scheme": "http",
+        }
+    )
+    result.state.principal = "root@pam"
+    return result
+
+
+def inputs(**values: object) -> dict[str, Any]:
+    return {"values": values, "provided": tuple(values)}
+
+
+async def test_qemu_create_sync_async_update_and_delete(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    created_payloads: list[dict[str, Any]] = []
+
+    class FakeTaskRepository:
+        def __init__(self, pool: object) -> None:
+            del pool
+
+        async def create(self, **kwargs: Any) -> Task:
+            created_payloads.append(kwargs)
+            return Task(
+                uuid.uuid4(),
+                str(kwargs["upid"]),
+                str(kwargs["task_type"]),
+                "queued",
+                dict(kwargs["payload"]),
+                0,
+                False,
+                0,
+            )
+
+    monkeypatch.setattr("app.handlers.qemu.TaskRepository", FakeTaskRepository)
+    registry = HandlerRegistry()
+    register_qemu_handlers(registry)
+    pool = QemuPool()
+    http_request = request(pool)
+    create = registry.get("/nodes/{node}/qemu", "POST")
+    update_sync = registry.get("/nodes/{node}/qemu/{vmid}/config", "PUT")
+    update_async = registry.get("/nodes/{node}/qemu/{vmid}/config", "POST")
+    delete = registry.get("/nodes/{node}/qemu/{vmid}", "DELETE")
+    assert create and update_sync and update_async and delete
+
+    create_upid = await create(
+        http_request,
+        inputs(node="pve1", vmid=150, name="new", cores=2),
+    )
+    assert create_upid.startswith("UPID:pve1:")
+    assert created_payloads[-1]["task_type"] == "qemu-create"
+
+    assert (
+        await update_sync(
+            http_request,
+            inputs(node="pve1", vmid=150, name="sync", delete="unused"),
+        )
+        is None
+    )
+    assert len(pool.commands) == 2
+
+    update_upid = await update_async(
+        http_request,
+        inputs(node="pve1", vmid=150, memory="2048"),
+    )
+    assert update_upid.startswith("UPID:pve1:")
+    assert created_payloads[-1]["task_type"] == "qemu-update"
+
+    delete_upid = await delete(http_request, inputs(node="pve1", vmid=150))
+    assert delete_upid.startswith("UPID:pve1:")
+    assert created_payloads[-1]["task_type"] == "qemu-delete"
+
+
+async def test_qemu_crud_conflicts_and_missing_resources(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class ConflictingRepository:
+        def __init__(self, pool: object) -> None:
+            del pool
+
+        async def create(self, **_kwargs: object) -> Task:
+            raise ConflictError("resource is locked")
+
+    monkeypatch.setattr("app.handlers.qemu.TaskRepository", ConflictingRepository)
+    registry = HandlerRegistry()
+    register_qemu_handlers(registry)
+    pool = QemuPool()
+    http_request = request(pool)
+    create = registry.get("/nodes/{node}/qemu", "POST")
+    update = registry.get("/nodes/{node}/qemu/{vmid}/config", "PUT")
+    delete = registry.get("/nodes/{node}/qemu/{vmid}", "DELETE")
+    assert create and update and delete
+
+    with pytest.raises(ApiError) as locked:
+        await create(http_request, inputs(node="pve1", vmid=150))
+    assert locked.value.status_code == 409
+
+    pool.resource_exists = True
+    with pytest.raises(ApiError) as duplicate:
+        await create(http_request, inputs(node="pve1", vmid=150))
+    assert duplicate.value.status_code == 409
+
+    pool.missing = True
+    with pytest.raises(ApiError) as missing:
+        await update(http_request, inputs(node="pve1", vmid=150, name="missing"))
+    assert missing.value.status_code == 404
+
+    pool.missing = False
+    pool.running = True
+    with pytest.raises(ApiError) as running:
+        await delete(http_request, inputs(node="pve1", vmid=150))
+    assert running.value.status_code == 409

tests/unit/test_qemu_task.py

@@ -61,6 +61,45 @@ class Repository:
         self.logs.append(message)
 
 
+class Transaction:
+    async def __aenter__(self) -> None:
+        return None
+
+    async def __aexit__(self, *args: object) -> None:
+        return None
+
+
+class CrudConnection:
+    def __init__(self) -> None:
+        self.commands: list[str] = []
+
+    def transaction(self) -> Transaction:
+        return Transaction()
+
+    async def fetchrow(self, sql: str, *args: object) -> dict[str, object] | None:
+        del args
+        if "FROM nodes" in sql:
+            return {"id": uuid.uuid4(), "cluster_id": uuid.uuid4()}
+        if "JOIN virtual_machines" in sql:
+            return {"state": '{"status":"stopped","name":"old"}', "config": '{"name":"old"}'}
+        return None
+
+    async def execute(self, sql: str, *args: object) -> str:
+        del args
+        self.commands.append(sql)
+        return "DELETE 1" if sql.startswith("DELETE") else "UPDATE 1"
+
+
+class CrudRepository:
+    def __init__(self) -> None:
+        self.connection = CrudConnection()
+        self.pool = Pool(cast(Connection, self.connection))
+        self.logs: list[str] = []
+
+    async def append_log(self, _task_id: uuid.UUID, message: str) -> None:
+        self.logs.append(message)
+
+
 async def test_qemu_worker_applies_intermediate_and_final_states() -> None:
     repository = Repository()
     task = Task(
@@ -82,3 +121,57 @@ async def test_qemu_worker_applies_intermediate_and_final_states() -> None:
     assert '"starting"' in repository.connection.states[0]
     assert '"running"' in repository.connection.states[1]
     assert repository.logs == ["VM start started", "VM start completed"]
+
+
+async def test_qemu_worker_create_update_and_delete_are_persistent() -> None:
+    repository = CrudRepository()
+    handler = qemu_handler(cast(TaskRepository, repository), cast(Clock, ImmediateClock()))
+    resource_id = uuid.uuid4()
+
+    created = await handler(
+        Task(
+            uuid.uuid4(),
+            "UPID:create",
+            "qemu-create",
+            "running",
+            {"node": "pve1", "vmid": 150, "config": {"name": "new"}},
+            0,
+            False,
+            1,
+        )
+    )
+    updated = await handler(
+        Task(
+            uuid.uuid4(),
+            "UPID:update",
+            "qemu-update",
+            "running",
+            {
+                "resource_id": str(resource_id),
+                "changes": {"name": "changed", "cores": 4},
+                "delete": "unused",
+            },
+            0,
+            False,
+            1,
+        )
+    )
+    deleted = await handler(
+        Task(
+            uuid.uuid4(),
+            "UPID:delete",
+            "qemu-delete",
+            "running",
+            {"resource_id": str(resource_id)},
+            0,
+            False,
+            1,
+        )
+    )
+
+    assert created == {"vmid": 150, "status": "stopped"}
+    assert updated == {"updated": ["cores", "name"], "deleted": ["unused"]}
+    assert deleted == {"deleted": True}
+    assert any("INSERT INTO resources" in command for command in repository.connection.commands)
+    assert any("UPDATE virtual_machines" in command for command in repository.connection.commands)
+    assert any("DELETE FROM resources" in command for command in repository.connection.commands)