fix(smart-plugs): Home Assistant scripts linked to a printer run when Bambuddy switches its power plug off or on by schedule, not only by hand

CHANGELOG.md

@@ -164,6 +164,7 @@ All notable changes to Bambuddy will be documented in this file.
 - **The frontend build no longer warns about `path` and `crypto` being externalized for the STEP previewer (#2976)** — `occt-import-js`, the Emscripten build behind STEP previews, requires both modules, but only inside its `ENVIRONMENT_IS_NODE` branches; in the browser it loads its `.wasm` from the URL the preview worker passes and draws randomness from `crypto.getRandomValues`. Vite still externalized both and printed two warnings on every build. `vite.config.ts` now drops exactly those two warnings for that one package through `build.rolldownOptions.onLog`, so an externalization anywhere else, or of any other module, still shows.
 
 ### Fixed
+- **Home Assistant "Run when printer turns off" scripts run when Bambuddy switches the printer off** — Linked scripts ran only when someone switched the printer's plug by hand. Auto-off after a print or drying (also when resumed after a restart), the plug's schedule and AI failure detection turned the printer off without them, and the schedule turned it on without its "on" scripts. They now run on all of these. "On" scripts keep running at every print start as before. For these automatic switches only a plug that powers the printer counts, the scripts run once when two plugs feed the printer, and a script whose **Enabled** toggle is off does not run (by hand either). Failure detection no longer tries to switch off a script.
 - **A plate check that could not run says so** — When the camera gave no image, the analysis failed or there was no calibration to compare against, the check before a print reported an empty plate, logged "Plate is empty" and told nobody. The print still goes ahead, but users who can control printers now get a warning with the reason, the log records it, and **Check Plate** shows it instead of "Plate appears empty".
 - **The archive recovery download gives a large 3MF time** — The retry that fetches a print's 3MF when the first download failed gave up after a flat 90 seconds, so a large file on a slow SD card never made it. Once the printer reports the file's size, the deadline follows it (up to 15 minutes), as the first download's already did.
 - **A broken MakerWorld download is not kept** — Only the HTTP status was checked, so a truncated file or an error page served as a success was stored as the model's 3MF. It is now refused unless it is a readable 3MF.

backend/app/api/routes/smart_plugs.py

@@ -55,7 +55,7 @@ from backend.app.services.secret_fields import (
     reject_placeholder,
     same_value,
 )
-from backend.app.services.smart_plug_manager import publish_plug_state, smart_plug_manager
+from backend.app.services.smart_plug_manager import publish_plug_state, run_linked_scripts, smart_plug_manager
 from backend.app.services.tasmota import tasmota_service
 from backend.app.utils.local_time import utcnow_naive
 
@@ -807,7 +807,7 @@ async def control_smart_plug(
         plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
     )
     if is_main_plug and plug.printer_id and expected_state:
-        await trigger_associated_scripts(plug.printer_id, expected_state, db)
+        await run_linked_scripts(plug.printer_id, expected_state, db)
 
     # MQTT relay - publish smart plug state change. A toggle doesn't say which
     # way it went, so read the state it ended in.
@@ -825,37 +825,6 @@ async def control_smart_plug(
     return {"success": True, "action": control.action}
 
 
-async def trigger_associated_scripts(printer_id: int, plug_state: str, db: AsyncSession):
-    """Trigger scripts linked to a printer based on main plug state change.
-
-    When the main plug turns ON, triggers scripts with auto_on=True.
-    When the main plug turns OFF, triggers scripts with auto_off=True.
-    """
-    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
-    plugs = result.scalars().all()
-
-    # Find scripts that should be triggered
-    for plug in plugs:
-        is_script = plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script.")
-        if not is_script:
-            continue
-
-        should_trigger = False
-        if plug_state == "ON" and plug.auto_on:
-            should_trigger = True
-            logger.info("Auto-triggering script '%s' on printer power-on", plug.name)
-        elif plug_state == "OFF" and plug.auto_off:
-            should_trigger = True
-            logger.info("Auto-triggering script '%s' on printer power-off", plug.name)
-
-        if should_trigger:
-            try:
-                service = await _get_service_for_plug(plug, db)
-                await service.turn_on(plug)  # Scripts are triggered by calling turn_on
-            except Exception as e:
-                logger.error("Failed to trigger script '%s': %s", plug.name, e)
-
-
 @router.get("/{plug_id}/status", response_model=SmartPlugStatus)
 async def get_plug_status(
     plug_id: int,

backend/app/services/energy_plug.py

@@ -53,7 +53,7 @@ def _is_script_entity(plug: SmartPlug) -> bool:
     """A Home Assistant ``script.*`` entity linked to a printer for automation.
 
     Stored as plugs so they can follow the print cycle (see
-    ``trigger_associated_scripts``), but a script has nothing to meter.
+    ``run_linked_scripts``), but a script has nothing to meter.
     """
     return bool(plug.plug_type == "homeassistant" and plug.ha_entity_id and plug.ha_entity_id.startswith("script."))
 

backend/app/services/obico_actions.py

@@ -52,12 +52,20 @@ def _pause_print(printer_id: int) -> None:
 
 
 async def _turn_off_linked_plugs(printer_id: int) -> None:
-    from backend.app.services.smart_plug_manager import publish_plug_state, smart_plug_manager
-
+    from backend.app.services.smart_plug_manager import (
+        _is_script_plug,
+        publish_plug_state,
+        run_linked_scripts_after_automatic_switch,
+        smart_plug_manager,
+    )
+
+    power_plug_off = None
     async with async_session() as db:
         plugs = await smart_plug_manager._get_plugs_for_printer(printer_id, db)
         for plug in plugs:
-            if not plug.enabled:
+            # A script can only be run, not turned off; the "printer turns
+            # off" ones run below, after the plug that feeds the printer.
+            if not plug.enabled or _is_script_plug(plug):
                 continue
             try:
                 service = await smart_plug_manager.get_service_for_plug(plug, db)
@@ -65,9 +73,14 @@ async def _turn_off_linked_plugs(printer_id: int) -> None:
                     await publish_plug_state(
                         plug.id, plug.name, plug.printer_id, on=False, ha_entity_id=plug.ha_entity_id
                     )
+                    if plug.controls_printer_power:
+                        power_plug_off = plug
                 logger.info("Obico action: turned off plug %s for printer %s", plug.name, printer_id)
             except Exception as e:
                 logger.error("Obico action: failed to turn off plug %s: %s", plug.name, e)
+    if power_plug_off is not None:
+        # Once, however many plugs feed the printer.
+        await run_linked_scripts_after_automatic_switch(printer_id, power_plug_off, "OFF")
 
 
 async def _notify(

backend/app/services/smart_plug_manager.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import logging
+import time
 from datetime import datetime, timezone
 from typing import TYPE_CHECKING
 
@@ -60,6 +61,72 @@ async def publish_plug_state(
         logger.debug("MQTT relay: failed to publish the state of plug %s: %s", plug_id, e)
 
 
+def _is_script_plug(plug) -> bool:
+    """A Home Assistant script entity: it can be run, not switched."""
+    return (
+        getattr(plug, "plug_type", None) == "homeassistant"
+        and bool(getattr(plug, "ha_entity_id", None))
+        and plug.ha_entity_id.startswith("script.")
+    )
+
+
+async def run_linked_scripts(printer_id: int, plug_state: str, db: AsyncSession) -> None:
+    """Run the HA scripts linked to a printer after its power plug switched.
+
+    ON runs the scripts set to "Run when printer turns on", OFF those set to
+    "Run when printer turns off". Never raises: a script that fails must not
+    undo the switch that triggered it.
+    """
+    from backend.app.models.smart_plug import SmartPlug
+
+    result = await db.execute(select(SmartPlug).where(SmartPlug.printer_id == printer_id))
+    for plug in result.scalars().all():
+        # "Enabled" is the script's automation switch, as at print start.
+        if not _is_script_plug(plug) or not plug.enabled:
+            continue
+        if plug_state == "ON" and plug.auto_on:
+            logger.info("Auto-triggering script '%s' on printer power-on", plug.name)
+        elif plug_state == "OFF" and plug.auto_off:
+            logger.info("Auto-triggering script '%s' on printer power-off", plug.name)
+        else:
+            continue
+        try:
+            service = await smart_plug_manager.get_service_for_plug(plug, db)
+            await service.turn_on(plug)  # Scripts are triggered by calling turn_on
+        except Exception as e:
+            logger.error("Failed to trigger script '%s': %s", plug.name, e)
+
+
+# (printer_id, state) -> when its scripts last ran after an automatic switch.
+# Two plugs feeding one printer both auto-off after a print; the printer
+# turned off once.
+_recent_automatic_script_runs: dict[tuple[int, str], float] = {}
+_AUTOMATIC_SCRIPT_RUN_WINDOW_S = 120.0
+
+
+async def run_linked_scripts_after_automatic_switch(printer_id: int | None, plug, plug_state: str) -> None:
+    """Linked scripts for a printer's power plug switched by Bambuddy itself.
+
+    Only for a plug that feeds the printer: an accessory switching is not the
+    printer turning on or off. Print start is not routed here -- it runs the
+    "on" scripts itself, as linked plugs with auto-on.
+    """
+    if not printer_id or _is_script_plug(plug) or not getattr(plug, "controls_printer_power", True):
+        return
+    now = time.monotonic()
+    last = _recent_automatic_script_runs.get((printer_id, plug_state))
+    if last is not None and now - last < _AUTOMATIC_SCRIPT_RUN_WINDOW_S:
+        return
+    _recent_automatic_script_runs[(printer_id, plug_state)] = now
+    from backend.app.core.database import async_session
+
+    try:
+        async with async_session() as db:
+            await run_linked_scripts(printer_id, plug_state, db)
+    except Exception as e:
+        logger.warning("Could not run the scripts linked to printer %s: %s", printer_id, e)
+
+
 class SmartPlugManager:
     """Manages smart plug automation and delayed turn-off."""
 
@@ -328,6 +395,7 @@ class SmartPlugManager:
                             await publish_plug_state(
                                 plug.id, plug.name, plug.printer_id, on=True, ha_entity_id=plug.ha_entity_id
                             )
+                            await run_linked_scripts_after_automatic_switch(plug.printer_id, plug, "ON")
 
                 # Check if we should turn off
                 if plug.schedule_off_time == current_time:
@@ -345,6 +413,7 @@ class SmartPlugManager:
                             # Mark printer offline if this plug feeds it (#2629)
                             if plug.printer_id and plug.controls_printer_power:
                                 printer_manager.mark_printer_offline(plug.printer_id)
+                            await run_linked_scripts_after_automatic_switch(plug.printer_id, plug, "OFF")
 
             await db.commit()
 
@@ -601,6 +670,7 @@ class SmartPlugManager:
                     self.rest_off_body = rest_off_body
                     self.rest_method = rest_method
                     self.rest_headers = rest_headers
+                    self.controls_printer_power = controls_printer_power
 
             plug_info = PlugInfo()
             service = await self.get_service_for_plug(plug_info)
@@ -617,6 +687,7 @@ class SmartPlugManager:
                 # plug actually feeds the printer (#2629).
                 if controls_printer_power:
                     printer_manager.mark_printer_offline(printer_id)
+                await run_linked_scripts_after_automatic_switch(printer_id, plug_info, "OFF")
 
         except asyncio.CancelledError:
             logger.debug("Delayed turn-off cancelled for plug %s", plug_id)
@@ -734,6 +805,7 @@ class SmartPlugManager:
                                 self.rest_off_body = rest_off_body
                                 self.rest_method = rest_method
                                 self.rest_headers = rest_headers
+                                self.controls_printer_power = controls_printer_power
 
                         plug_info = PlugInfo()
                         service = await self.get_service_for_plug(plug_info)
@@ -753,6 +825,7 @@ class SmartPlugManager:
                             # when this plug actually feeds the printer (#2629).
                             if controls_printer_power:
                                 printer_manager.mark_printer_offline(printer_id)
+                            await run_linked_scripts_after_automatic_switch(printer_id, plug_info, "OFF")
 
                         break
 
@@ -903,6 +976,7 @@ class SmartPlugManager:
                             )
                             if plug.controls_printer_power:
                                 printer_manager.mark_printer_offline(plug.printer_id)
+                            await run_linked_scripts_after_automatic_switch(plug.printer_id, plug, "OFF")
 
                 if pending_plugs:
                     logger.info("Resumed %s pending auto-off(s)", len(pending_plugs))

backend/tests/unit/test_ha_scripts_on_automatic_switch.py

@@ -0,0 +1,255 @@
+"""HA "Run when printer turns on/off" scripts follow automatic switching too.
+
+They ran only when someone switched the printer's plug by hand. Auto-off
+(delayed, by temperature, resumed after a restart), the plug schedule and AI
+failure detection switched the printer without them. Print start already runs
+the "on" scripts itself, as linked plugs with auto-on, so it is not routed
+through here; nor is the queue's power-on, which print start follows.
+"""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from backend.app.services import smart_plug_manager as spm
+from backend.app.services.smart_plug_manager import SmartPlugManager
+
+
+@pytest.fixture(autouse=True)
+def _fresh_run_memory():
+    spm._recent_automatic_script_runs.clear()
+    yield
+    spm._recent_automatic_script_runs.clear()
+
+
+def _plug(**fields):
+    base = {
+        "id": 1,
+        "name": "Plug",
+        "plug_type": "tasmota",
+        "ha_entity_id": None,
+        "printer_id": 7,
+        "controls_printer_power": True,
+        "auto_on": False,
+        "auto_off": False,
+        "enabled": True,
+    }
+    base.update(fields)
+    return SimpleNamespace(**base)
+
+
+def _script(name, **fields):
+    return _plug(name=name, plug_type="homeassistant", ha_entity_id=f"script.{name}", **fields)
+
+
+class TestWhichScriptsRun:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(("state", "ran"), [("ON", ["lights_on"]), ("OFF", ["fan_off"])])
+    async def test_by_the_direction_of_the_switch(self, state, ran):
+        plugs = [_plug(), _script("lights_on", auto_on=True), _script("fan_off", auto_off=True)]
+        db = MagicMock()
+        db.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: plugs)))
+        service = MagicMock(turn_on=AsyncMock(return_value=True))
+
+        with patch.object(spm.smart_plug_manager, "get_service_for_plug", AsyncMock(return_value=service)):
+            await spm.run_linked_scripts(7, state, db)
+
+        assert [call.args[0].name for call in service.turn_on.await_args_list] == ran
+
+
+class TestOnlyThePrintersPowerPlug:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ("plug", "printer_id", "runs"),
+        [
+            (_plug(), 7, True),
+            (_plug(controls_printer_power=False), 7, False),  # an accessory, e.g. a lamp
+            (_script("x"), 7, False),
+            (_plug(), None, False),
+        ],
+    )
+    async def test_after_an_automatic_switch(self, plug, printer_id, runs):
+        session = MagicMock()
+        session.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+        session.return_value.__aexit__ = AsyncMock(return_value=False)
+        run = AsyncMock()
+
+        with (
+            patch("backend.app.core.database.async_session", session),
+            patch.object(spm, "run_linked_scripts", run),
+        ):
+            await spm.run_linked_scripts_after_automatic_switch(printer_id, plug, "OFF")
+
+        assert run.await_count == (1 if runs else 0)
+
+
+@pytest.fixture
+def scripts():
+    run = AsyncMock()
+    with patch.object(spm, "run_linked_scripts_after_automatic_switch", run):
+        yield run
+
+
+def _service():
+    return MagicMock(turn_on=AsyncMock(return_value=True), turn_off=AsyncMock(return_value=True))
+
+
+class TestEveryAutomaticOff:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize("feeds_printer", [True, False])
+    async def test_the_delayed_auto_off(self, scripts, feeds_printer):
+        manager = SmartPlugManager()
+        with (
+            patch.object(spm, "printer_manager") as pm,
+            patch.object(spm, "publish_plug_state", AsyncMock()),
+            patch.object(manager, "get_service_for_plug", AsyncMock(return_value=_service())),
+            patch.object(manager, "_mark_auto_off_executed", AsyncMock()),
+        ):
+            pm.is_print_active.return_value = False
+            await manager._delayed_off(
+                3, "tasmota", "1.2.3.4", None, None, None, 7, 0, controls_printer_power=feeds_printer
+            )
+
+        printer_id, plug, state = scripts.await_args.args
+        assert (printer_id, state, plug.controls_printer_power) == (7, "OFF", feeds_printer)
+
+    @pytest.mark.asyncio
+    async def test_the_temperature_auto_off(self, scripts):
+        manager = SmartPlugManager()
+        with (
+            patch.object(spm, "printer_manager") as pm,
+            patch.object(spm, "publish_plug_state", AsyncMock()),
+            patch.object(spm.asyncio, "sleep", AsyncMock()),
+            patch.object(manager, "get_service_for_plug", AsyncMock(return_value=_service())),
+            patch.object(manager, "_mark_auto_off_executed", AsyncMock()),
+        ):
+            pm.get_status.return_value = MagicMock(state="FINISH", temperatures={"nozzle": 30})
+            pm.is_print_active.return_value = False
+            await manager._temp_based_off(3, "tasmota", "1.2.3.4", None, None, None, printer_id=7, temp_threshold=55)
+
+        assert scripts.await_args.args[0] == 7 and scripts.await_args.args[2] == "OFF"
+
+    @pytest.mark.asyncio
+    async def test_a_resumed_auto_off(self, scripts):
+        manager = SmartPlugManager()
+        plug = MagicMock(
+            id=1, printer_id=7, off_delay_mode="time", controls_printer_power=True, auto_off_pending_since=None
+        )
+        db = AsyncMock()
+        db.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [plug])))
+        session = MagicMock()
+        session.return_value.__aenter__ = AsyncMock(return_value=db)
+        session.return_value.__aexit__ = AsyncMock(return_value=False)
+        with (
+            patch("backend.app.core.database.async_session", session),
+            patch.object(spm, "printer_manager") as pm,
+            patch.object(spm, "publish_plug_state", AsyncMock()),
+            patch.object(manager, "get_service_for_plug", AsyncMock(return_value=_service())),
+            patch.object(manager, "_mark_auto_off_executed", AsyncMock()),
+        ):
+            pm.is_print_active.return_value = False
+            await manager.resume_pending_auto_offs()
+
+        scripts.assert_awaited_once_with(7, plug, "OFF")
+
+
+class TestTheSchedule:
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(("now", "state"), [("08:00", "ON"), ("22:00", "OFF")])
+    async def test_a_scheduled_switch(self, scripts, now, state):
+        manager = SmartPlugManager()
+        plug = MagicMock(
+            id=1, printer_id=7, schedule_on_time="08:00", schedule_off_time="22:00", controls_printer_power=True
+        )
+        db = AsyncMock()
+        db.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: [plug])))
+        session = MagicMock()
+        session.return_value.__aenter__ = AsyncMock(return_value=db)
+        session.return_value.__aexit__ = AsyncMock(return_value=False)
+        with (
+            patch("backend.app.core.database.async_session", session),
+            patch.object(spm, "datetime") as clock,
+            patch.object(spm, "printer_manager"),
+            patch.object(spm, "publish_plug_state", AsyncMock()),
+            patch.object(manager, "get_service_for_plug", AsyncMock(return_value=_service())),
+        ):
+            clock.now.return_value = MagicMock(strftime=MagicMock(return_value=now))
+            await manager._check_schedules()
+
+        scripts.assert_awaited_once_with(7, plug, state)
+
+
+class TestFailureDetection:
+    @pytest.mark.asyncio
+    async def test_scripts_run_once_and_are_not_turned_off(self, scripts):
+        """Two plugs feed the printer; the linked script is run, not switched."""
+        from backend.app.services import obico_actions
+
+        mains = [_plug(id=1), _plug(id=2)]
+        script = _script("fan_off", auto_off=True, id=3)
+        service = _service()
+        session = MagicMock()
+        session.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+        session.return_value.__aexit__ = AsyncMock(return_value=False)
+        with (
+            patch.object(obico_actions, "async_session", session),
+            patch.object(spm.smart_plug_manager, "_get_plugs_for_printer", AsyncMock(return_value=[*mains, script])),
+            patch.object(spm.smart_plug_manager, "get_service_for_plug", AsyncMock(return_value=service)),
+            patch.object(spm, "publish_plug_state", AsyncMock()),
+        ):
+            await obico_actions._turn_off_linked_plugs(7)
+
+        assert [call.args[0].id for call in service.turn_off.await_args_list] == [1, 2]
+        assert scripts.await_count == 1
+        assert scripts.await_args.args[0] == 7 and scripts.await_args.args[2] == "OFF"
+
+
+class TestReviewCases:
+    @pytest.mark.asyncio
+    async def test_a_script_whose_automation_is_off_does_not_run(self):
+        plugs = [_script("paused", auto_off=True, enabled=False), _script("live", auto_off=True)]
+        db = MagicMock()
+        db.execute = AsyncMock(return_value=MagicMock(scalars=lambda: MagicMock(all=lambda: plugs)))
+        service = MagicMock(turn_on=AsyncMock(return_value=True))
+
+        with patch.object(spm.smart_plug_manager, "get_service_for_plug", AsyncMock(return_value=service)):
+            await spm.run_linked_scripts(7, "OFF", db)
+
+        assert [call.args[0].name for call in service.turn_on.await_args_list] == ["live"]
+
+    @pytest.mark.asyncio
+    async def test_two_plugs_switching_the_printer_run_the_scripts_once(self):
+        """Two plugs feed one printer and both auto-off after a print."""
+        session = MagicMock()
+        session.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+        session.return_value.__aexit__ = AsyncMock(return_value=False)
+        run = AsyncMock()
+
+        with (
+            patch("backend.app.core.database.async_session", session),
+            patch.object(spm, "run_linked_scripts", run),
+        ):
+            await spm.run_linked_scripts_after_automatic_switch(7, _plug(id=1), "OFF")
+            await spm.run_linked_scripts_after_automatic_switch(7, _plug(id=2), "OFF")
+            await spm.run_linked_scripts_after_automatic_switch(7, _plug(id=1), "ON")
+
+        assert [call.args[1] for call in run.await_args_list] == ["OFF", "ON"]
+
+    @pytest.mark.asyncio
+    async def test_failure_detection_that_switched_no_power_plug_runs_nothing(self, scripts):
+        from backend.app.services import obico_actions
+
+        lamp = _plug(id=4, controls_printer_power=False)
+        session = MagicMock()
+        session.return_value.__aenter__ = AsyncMock(return_value=MagicMock())
+        session.return_value.__aexit__ = AsyncMock(return_value=False)
+        with (
+            patch.object(obico_actions, "async_session", session),
+            patch.object(spm.smart_plug_manager, "_get_plugs_for_printer", AsyncMock(return_value=[lamp])),
+            patch.object(spm.smart_plug_manager, "get_service_for_plug", AsyncMock(return_value=_service())),
+            patch.object(spm, "publish_plug_state", AsyncMock()),
+        ):
+            await obico_actions._turn_off_linked_plugs(7)
+
+        scripts.assert_not_awaited()

backend/tests/unit/test_mqtt_relay_events.py

@@ -132,7 +132,7 @@ async def test_an_mqtt_plug_that_stopped_reporting_publishes_nothing(db_session,
 # The plug backends themselves (a REST toggle is a turn_on or turn_off).
 PLUG_BACKENDS = {"tasmota.py", "homeassistant.py", "rest_smart_plug.py", "mqtt_smart_plug.py"}
 # Runs a Home Assistant script through its entity's turn_on: no plug changes state.
-NOT_A_SWITCH = {("api/routes/smart_plugs.py", "trigger_associated_scripts")}
+NOT_A_SWITCH = {("services/smart_plug_manager.py", "run_linked_scripts")}
 
 
 def _called_names(node: ast.AST) -> set[str]: