perf(azure): skip redundant worker tag writes

bin/fm-azure-worker-provider.py

@@ -1708,6 +1708,12 @@ def tag_container(controller, name, tags):
         raise ProviderError("exact worker state-container metadata update failed: {}".format(stderr))
 
 
+def carries_tags(resource, expected):
+    """Whether one inventory record already carries every exact assignment tag."""
+    current = resource.get("tags") or {}
+    return all(str(current.get(key)) == str(value) for key, value in expected.items())
+
+
 def cleanup_marker(resource, key, value):
     return (resource.get("tags") or {}).get(key) == value
 
@@ -2675,8 +2681,11 @@ def converge_create_tags(controller, action):
             "staging-request", "staging-result",
         ):
             continue
-        tag_resource(controller, resource["id"], tags)
-    tag_container(controller, expected_names(controller, action["slot"])["state-container"], tags)
+        if not carries_tags(resource, tags):
+            tag_resource(controller, resource["id"], tags)
+    container = worker["resources"].get("state-container")
+    if container is None or not carries_tags(container, tags):
+        tag_container(controller, expected_names(controller, action["slot"])["state-container"], tags)
     snapshot = inventory_slot(controller, action["slot"])
     if snapshot["conflicts"]:
         raise ProviderError("tagged worker inventory contains foreign or unsafe resources")

tests/fm-worker-lifecycle.test.sh

@@ -1684,6 +1684,36 @@ ready_module.run_pilot_create=lambda *_args: (_ for _ in ()).throw(
 resumed=ready_module.create_or_resume({"prefix":"fmtest"}, ready_action)
 assert resumed is ready_worker and ready_calls == ["read"], (resumed, ready_calls)
 
+# The worker deployment already carries the exact assignment tags. Rewriting
+# them makes Azure reconverge the VM extension and managed Run Commands after
+# they have succeeded, adding control-plane work to every cold admission. Only
+# an actually missing or differing tag should produce a write.
+tag_spec=importlib.util.spec_from_file_location("azure_provider_tags", sys.argv[1])
+tag_module=importlib.util.module_from_spec(tag_spec); tag_spec.loader.exec_module(tag_module)
+expected_tags={"alpha":"one","beta":"two"}
+tag_resources={kind:{
+    "id":"/{}".format(kind), "immutable_id":"i-{}".format(kind),
+    "tags":dict(expected_tags), "provisioning_state":"succeeded",
+} for kind in tag_module.REQUIRED_RESOURCE_KINDS}
+tag_worker={"slot":1,"resources":tag_resources}
+tag_action={"slot":1}
+tag_module.action_tags=lambda *_args: expected_tags
+tag_module.inventory_slot=lambda *_args: {"workers":[tag_worker],"conflicts":[]}
+tag_module.worker_by_slot=lambda snapshot, _slot: snapshot["workers"][0]
+tag_module.recorded_exact=lambda _action, worker, **_kwargs: worker["resources"]
+writes=[]
+tag_module.tag_resource=lambda _controller, resource_id, _tags: writes.append(("resource",resource_id))
+tag_module.tag_container=lambda _controller, name, _tags: writes.append(("container",name))
+tag_module.converge_create_tags({"prefix":"fmtest"}, tag_action)
+assert writes == [], writes
+
+tag_resources["vm"]["tags"].pop("beta")
+tag_resources["state-container"]["tags"]["beta"]="stale"
+tag_module.converge_create_tags({"prefix":"fmtest"}, tag_action)
+assert writes == [
+    ("resource", "/vm"), ("container", "worker-state-01"),
+], writes
+
 # A pre-convergence container has empty metadata; it inherits a same-slot
 # exact-fleet sibling's tags (VM first) instead of classifying as foreign,
 # while a bare orphan container keeps its emptiness and still refuses.