diff --git a/README.md b/README.md
index cd794f8..3bd195c 100644
--- a/README.md
+++ b/README.md
@@ -72,7 +72,14 @@ Minimum practical read privileges:
 - `VM.Audit` for VM and container inventory/config visibility.
 - `SDN.Audit` if you use Proxmox SDN zones, VNets, EVPN, or related network objects.
 
-For future write-enabled firewall orchestration, create a separate token or role and do not reuse the read-only token. Only enable write mode after previews and audit logging have been verified in your environment.
+For write-enabled firewall orchestration, create a separate token or role and do not reuse the read-only token. Only enable write mode after previews and audit logging have been verified in your environment.
+
+Minimum practical write privileges for VM/LXC-level firewall rules:
+
+- `VM.Audit` so NexaFabric can resolve guests and inspect existing rules.
+- `VM.Config.Network` on `/vms` or on the narrow VM/LXC paths you want NexaFabric to manage.
+
+NexaFabric writes only rules that carry a `NexaFabric policy=...` comment marker. During apply it removes and replaces its own marked rules for the selected policy, leaving manually created Proxmox firewall rules untouched.
 
 ### 3. Enable QEMU Guest Agent For VM IP Discovery
 
@@ -123,11 +130,20 @@ NexaFabric policy preview does not require Proxmox firewall writes. It compiles
 Before enabling real firewall apply workflows:
 
 - Ensure Proxmox firewall is enabled intentionally at the Datacenter, node, and guest level where you want enforcement.
+- Ensure the firewall checkbox is enabled on the relevant VM/LXC network interfaces, otherwise Proxmox may store rules without enforcing them for that interface.
 - Keep NexaFabric clusters in `read_only` mode until previews are reviewed.
 - Use `audit` mode policies first to see what would be allowed or blocked.
 - Confirm that Proxmox API token permissions match the exact write operations you plan to allow.
 - Keep backups of Proxmox firewall configuration before enabling automation.
 
+Live apply currently supports VM/LXC-level rules where the enforcement side is a concrete workload:
+
+- `ingress` policies apply to the destination VM/LXC.
+- `egress` policies apply to the source VM/LXC.
+- The opposite side can be `any`, an IP/CIDR, or another workload with an assigned IP in IPAM.
+- Security group and network-wide targets remain preview-only until they can be expanded safely.
+- `audit` mode does not write blocking Proxmox rules; it records the intended result without enforcement.
+
 NexaFabric is designed to read first, simulate second, and only apply after explicit confirmation.
 
 ### 7. Network Flow Visibility
@@ -178,4 +194,4 @@ nginx/         Reverse proxy example
 
 ## Safety Model
 
-NexaFabric never applies firewall changes without a preview, validation, and audit record. The included Proxmox provider is designed around read-only inventory first. Write-enabled orchestration is intentionally routed through explicit dry-run and apply workflows.
+NexaFabric never applies firewall changes without a preview, validation, and audit record. The included Proxmox provider is designed around read-only inventory first, then explicit dry-run and apply workflows for concrete VM/LXC firewall rules.
diff --git a/backend/app/api/v1/router.py b/backend/app/api/v1/router.py
index e68abce..1fa1daf 100644
--- a/backend/app/api/v1/router.py
+++ b/backend/app/api/v1/router.py
@@ -1,7 +1,7 @@
 from datetime import datetime
 import csv
 import io
-from ipaddress import ip_interface
+from ipaddress import ip_address, ip_interface, ip_network
 
 from fastapi import APIRouter, Depends, HTTPException
 from fastapi.responses import StreamingResponse
@@ -139,6 +139,18 @@ def ip_address_payload(db: Session, address: IpAddress) -> dict:
     }
 
 
+def is_ip_or_cidr(value: str) -> bool:
+    try:
+        ip_network(value, strict=False)
+        return True
+    except ValueError:
+        try:
+            ip_address(value)
+            return True
+        except ValueError:
+            return False
+
+
 def cleanup_discovered_container_networks(db: Session) -> int:
     removed = 0
     discovered_networks = db.scalars(select(Network).where(Network.name == "discovered-ipam")).all()
@@ -181,6 +193,107 @@ def import_discovered_ips(db: Session, cluster_id: str, workload: Workload, addr
     return imported
 
 
+def workload_provider_target(db: Session, cluster: Cluster, ref: str) -> tuple[dict, Workload] | None:
+    if not ref.startswith("workload:"):
+        return None
+    workload_id = ref.removeprefix("workload:")
+    workload = db.get(Workload, workload_id)
+    if not workload or workload.cluster_id != cluster.id:
+        return None
+    node = db.get(Node, workload.node_id)
+    if not node:
+        return None
+    kind = "lxc" if workload.kind == "lxc" else "qemu"
+    return (
+        {
+            "node": node.name,
+            "kind": kind,
+            "vmid": workload.external_id,
+            "workload_id": workload.id,
+            "workload_name": workload.name,
+        },
+        workload,
+    )
+
+
+def endpoint_values(db: Session, cluster: Cluster, ref: str) -> tuple[list[str | None], list[str]]:
+    if ref == "any":
+        return [None], []
+    resolved = workload_provider_target(db, cluster, ref)
+    if resolved:
+        _, workload = resolved
+        addresses = db.scalars(select(IpAddress).where(IpAddress.workload_id == workload.id).order_by(IpAddress.address)).all()
+        values = [address.address for address in addresses if address.address]
+        if values:
+            return values, []
+        return [], [f"Workload {workload.name} has no assigned IP address for provider-side source/destination matching."]
+    if is_ip_or_cidr(ref):
+        return [ref], []
+    return [], [f"Endpoint {ref} is not yet resolvable to a Proxmox firewall matcher."]
+
+
+def proxmox_action(action: str) -> str:
+    return {"allow": "ACCEPT", "deny": "DROP", "reject": "REJECT"}.get(action, "ACCEPT")
+
+
+def resolve_firewall_preview(db: Session, cluster: Cluster, preview: FirewallPreview) -> FirewallPreview:
+    warnings = list(preview.warnings)
+    conflicts = list(preview.conflicts)
+    generated_rules: list[dict] = []
+
+    for rule_index, rule in enumerate(preview.generated_rules, start=1):
+        mapped = dict(rule)
+        direction = str(rule.get("direction", "ingress"))
+        target_ref = str(rule.get("destination") if direction == "ingress" else rule.get("source"))
+        target = workload_provider_target(db, cluster, target_ref)
+        if not target:
+            conflicts.append(
+                f"Rule {rule_index} needs a concrete {'destination' if direction == 'ingress' else 'source'} workload for Proxmox live apply."
+            )
+            generated_rules.append(mapped)
+            continue
+
+        provider_target, target_workload = target
+        remote_ref = str(rule.get("source") if direction == "ingress" else rule.get("destination"))
+        remote_values, endpoint_warnings = endpoint_values(db, cluster, remote_ref)
+        warnings.extend(endpoint_warnings)
+        if not remote_values:
+            conflicts.append(f"Rule {rule_index} cannot resolve {remote_ref} to a Proxmox firewall source/destination matcher.")
+            generated_rules.append({**mapped, "provider_target": provider_target})
+            continue
+
+        ports = str(rule.get("ports", "any"))
+        protocol = str(rule.get("protocol", "any"))
+        for remote_value in remote_values:
+            provider_rule = {
+                "type": "in" if direction == "ingress" else "out",
+                "action": proxmox_action(str(rule.get("action", "allow"))),
+                "enable": 1,
+                "comment": (
+                    f"NexaFabric policy={rule.get('policy_id')} version={rule.get('policy_version')} "
+                    f"rule={rule_index} target={target_workload.name}"
+                ),
+            }
+            if protocol != "any":
+                provider_rule["proto"] = protocol
+            if ports != "any":
+                provider_rule["dport"] = ports
+            if remote_value:
+                provider_rule["source" if direction == "ingress" else "dest"] = remote_value
+            if rule.get("logging"):
+                provider_rule["log"] = "info"
+            mapped_rule = {**mapped, "provider_target": provider_target, "provider_rule": provider_rule}
+            generated_rules.append(mapped_rule)
+
+    return FirewallPreview(
+        policy_id=preview.policy_id,
+        dry_run=preview.dry_run,
+        generated_rules=generated_rules,
+        warnings=warnings,
+        conflicts=conflicts,
+    )
+
+
 @api_router.get("/setup/status", response_model=SetupStatus)
 def setup_status(db: Session = Depends(get_db)) -> SetupStatus:
     setting = setup_setting(db)
@@ -757,7 +870,7 @@ async def firewall_preview(policy_id: str, user: CurrentUser, db: Session = Depe
     cluster = db.scalar(select(Cluster).order_by(Cluster.name).limit(1))
     if not policy or not cluster:
         raise HTTPException(status_code=404, detail="Policy or cluster not found")
-    preview = await FirewallOrchestrator().preview(cluster, policy)
+    preview = resolve_firewall_preview(db, cluster, await FirewallOrchestrator().preview(cluster, policy))
     write_audit(db, action="firewall.preview", object_type="policy", object_id=policy.id, user_id=user.id, new_values=preview.model_dump())
     return preview
 
@@ -770,7 +883,7 @@ async def firewall_apply(payload: FirewallApplyRequest, user: CurrentUser, db: S
     cluster = db.get(Cluster, payload.cluster_id) if payload.cluster_id else db.scalar(select(Cluster).order_by(Cluster.name).limit(1))
     if not policy or not cluster:
         raise HTTPException(status_code=404, detail="Policy or cluster not found")
-    preview = await FirewallOrchestrator().preview(cluster, policy)
+    preview = resolve_firewall_preview(db, cluster, await FirewallOrchestrator().preview(cluster, policy))
     if payload.dry_run:
         result = {
             "applied": False,
@@ -778,17 +891,31 @@ async def firewall_apply(payload: FirewallApplyRequest, user: CurrentUser, db: S
             "reason": "Dry run completed. No firewall rules were applied.",
             "rules": preview.generated_rules,
         }
+    elif preview.conflicts:
+        result = {
+            "applied": False,
+            "reason": "Live apply stopped because the preview has unresolved conflicts.",
+            "conflicts": preview.conflicts,
+            "rules": preview.generated_rules,
+        }
     else:
         provider = get_provider(cluster.provider)
-        result = await provider.apply_rules(
-            ProviderConnection(
-                api_url=cluster.api_url,
-                token=cluster.token_ref or "",
-                verify_tls=cluster.verify_tls,
-                read_only=cluster.mode == "read_only",
-            ),
-            preview.generated_rules,
-        )
+        try:
+            result = await provider.apply_rules(
+                ProviderConnection(
+                    api_url=cluster.api_url,
+                    token=cluster.token_ref or "",
+                    verify_tls=cluster.verify_tls,
+                    read_only=cluster.mode == "read_only",
+                ),
+                preview.generated_rules,
+            )
+        except Exception as exc:
+            result = {
+                "applied": False,
+                "reason": f"Provider apply failed: {exc}",
+                "rules": preview.generated_rules,
+            }
     applied = bool(result.get("applied"))
     operation_success = applied or payload.dry_run
     job = Job(
diff --git a/backend/app/services/providers/proxmox.py b/backend/app/services/providers/proxmox.py
index fd53c1a..6a181c9 100644
--- a/backend/app/services/providers/proxmox.py
+++ b/backend/app/services/providers/proxmox.py
@@ -127,11 +127,89 @@ class ProxmoxProvider(Provider):
             "warnings": ["Preview only. No Proxmox firewall changes were sent."],
         }
 
+    def firewall_rules_url(self, connection: ProviderConnection, target: dict[str, Any]) -> str:
+        kind = "lxc" if target.get("kind") == "lxc" else "qemu"
+        node = target["node"]
+        vmid = target["vmid"]
+        return f"{connection.api_url.rstrip('/')}/api2/json/nodes/{node}/{kind}/{vmid}/firewall/rules"
+
+    def policy_marker(self, rule: dict[str, Any]) -> str:
+        return f"NexaFabric policy={rule.get('policy_id')}"
+
+    async def delete_existing_policy_rules(
+        self,
+        client: httpx.AsyncClient,
+        headers: dict[str, str],
+        rules_url: str,
+        marker: str,
+    ) -> list[dict[str, Any]]:
+        existing_response = await client.get(rules_url, headers=headers)
+        existing_response.raise_for_status()
+        existing_rules = existing_response.json().get("data", [])
+        deletions = []
+        for existing_rule in sorted(existing_rules, key=lambda item: int(item.get("pos", 0)), reverse=True):
+            comment = str(existing_rule.get("comment") or "")
+            pos = existing_rule.get("pos")
+            if marker in comment and pos is not None:
+                delete_response = await client.delete(f"{rules_url}/{pos}", headers=headers)
+                delete_response.raise_for_status()
+                deletions.append({"pos": pos, "comment": comment})
+        return deletions
+
     async def apply_rules(self, connection: ProviderConnection, rules: list[dict[str, Any]]) -> dict[str, Any]:
         if connection.read_only:
             return {"applied": False, "reason": "Cluster is read-only", "rules": rules}
+        missing_mapping = [rule for rule in rules if "provider_target" not in rule]
+        if missing_mapping:
+            return {
+                "applied": False,
+                "reason": "Live apply requires every rule to resolve to a concrete Proxmox VM/LXC target.",
+                "rules": missing_mapping,
+            }
+
+        headers = {"Authorization": self.auth_header(connection.token)}
+        grouped: dict[tuple[str, str, str, str], list[dict[str, Any]]] = {}
+        for rule in rules:
+            target = rule["provider_target"]
+            marker = self.policy_marker(rule)
+            key = (str(target["node"]), str(target["kind"]), str(target["vmid"]), marker)
+            grouped.setdefault(key, []).append(rule)
+
+        applied_rules = []
+        deleted_rules = []
+        audit_only_rules = []
+        async with httpx.AsyncClient(verify=connection.verify_tls, timeout=20) as client:
+            for target_rules in grouped.values():
+                target = target_rules[0]["provider_target"]
+                rules_url = self.firewall_rules_url(connection, target)
+                marker = self.policy_marker(target_rules[0])
+                deleted_rules.extend(await self.delete_existing_policy_rules(client, headers, rules_url, marker))
+
+                for rule in target_rules:
+                    if rule.get("audit_only"):
+                        audit_only_rules.append(
+                            {
+                                "policy_id": rule.get("policy_id"),
+                                "target": target,
+                                "reason": "Audit mode does not enforce or write blocking Proxmox rules.",
+                            }
+                        )
+                        continue
+                    provider_rule = rule.get("provider_rule")
+                    if not provider_rule:
+                        return {
+                            "applied": False,
+                            "reason": "Resolved rule is missing provider_rule payload.",
+                            "rule": rule,
+                        }
+                    create_response = await client.post(rules_url, headers=headers, data=provider_rule)
+                    create_response.raise_for_status()
+                    applied_rules.append({"target": target, "rule": provider_rule, "result": create_response.json().get("data")})
+
         return {
-            "applied": False,
-            "reason": "Live Proxmox firewall apply needs rule-to-VM mapping before NexaFabric can safely write provider rules.",
-            "rules": rules,
+            "applied": True,
+            "rules_written": len(applied_rules),
+            "rules_deleted": len(deleted_rules),
+            "audit_only": audit_only_rules,
+            "rules": applied_rules,
         }
diff --git a/backend/tests/test_proxmox_provider.py b/backend/tests/test_proxmox_provider.py
new file mode 100644
index 0000000..467abbb
--- /dev/null
+++ b/backend/tests/test_proxmox_provider.py
@@ -0,0 +1,86 @@
+import pytest
+
+from app.services.providers import proxmox
+from app.services.providers.base import ProviderConnection
+from app.services.providers.proxmox import ProxmoxProvider
+
+
+class FakeResponse:
+    def __init__(self, data: object) -> None:
+        self._data = data
+
+    def json(self) -> dict:
+        return {"data": self._data}
+
+    def raise_for_status(self) -> None:
+        return None
+
+
+class FakeAsyncClient:
+    deleted_urls: list[str] = []
+    posted_payloads: list[dict] = []
+
+    def __init__(self, **_: object) -> None:
+        return None
+
+    async def __aenter__(self) -> "FakeAsyncClient":
+        return self
+
+    async def __aexit__(self, *_: object) -> None:
+        return None
+
+    async def get(self, _: str, **__: object) -> FakeResponse:
+        return FakeResponse(
+            [
+                {"pos": 0, "comment": "manual rule"},
+                {"pos": 1, "comment": "NexaFabric policy=policy-1 version=1 rule=1 target=web"},
+            ]
+        )
+
+    async def delete(self, url: str, **__: object) -> FakeResponse:
+        self.deleted_urls.append(url)
+        return FakeResponse(None)
+
+    async def post(self, _: str, data: dict, **__: object) -> FakeResponse:
+        self.posted_payloads.append(data)
+        return FakeResponse({"pos": 1})
+
+
+@pytest.mark.asyncio
+async def test_apply_rules_replaces_only_marked_nexafabric_rules(monkeypatch: pytest.MonkeyPatch) -> None:
+    FakeAsyncClient.deleted_urls = []
+    FakeAsyncClient.posted_payloads = []
+    monkeypatch.setattr(proxmox.httpx, "AsyncClient", FakeAsyncClient)
+
+    result = await ProxmoxProvider().apply_rules(
+        ProviderConnection(api_url="https://pve.example:8006", token="user@pve!token=secret", read_only=False),
+        [
+            {
+                "policy_id": "policy-1",
+                "audit_only": False,
+                "provider_target": {"node": "pve1", "kind": "qemu", "vmid": "100"},
+                "provider_rule": {
+                    "type": "in",
+                    "action": "ACCEPT",
+                    "enable": 1,
+                    "proto": "tcp",
+                    "dport": "443",
+                    "comment": "NexaFabric policy=policy-1 version=2 rule=1 target=web",
+                },
+            }
+        ],
+    )
+
+    assert result["applied"] is True
+    assert result["rules_deleted"] == 1
+    assert FakeAsyncClient.deleted_urls == ["https://pve.example:8006/api2/json/nodes/pve1/qemu/100/firewall/rules/1"]
+    assert FakeAsyncClient.posted_payloads == [
+        {
+            "type": "in",
+            "action": "ACCEPT",
+            "enable": 1,
+            "proto": "tcp",
+            "dport": "443",
+            "comment": "NexaFabric policy=policy-1 version=2 rule=1 target=web",
+        }
+    ]