diff --git a/data/debian/rules b/data/debian/rules
index 47d26ccb..f32142df 100755
--- a/data/debian/rules
+++ b/data/debian/rules
@@ -20,5 +20,6 @@ override_dh_installsystemd:
 	dh_installsystemd --no-start --name=procdockerstatsd
 	dh_installsystemd --no-start --name=determine-reboot-cause
 	dh_installsystemd --no-start --name=process-reboot-cause
+	dh_installsystemd --no-start --name=gnoi-shutdown
 	dh_installsystemd $(HOST_SERVICE_OPTS) --name=sonic-hostservice
 
diff --git a/data/debian/sonic-host-services-data.gnoi-shutdown.service b/data/debian/sonic-host-services-data.gnoi-shutdown.service
new file mode 100644
index 00000000..76d20ee3
--- /dev/null
+++ b/data/debian/sonic-host-services-data.gnoi-shutdown.service
@@ -0,0 +1,16 @@
+[Unit]
+Description=gNOI based DPU Graceful Shutdown Daemon
+Requires=database.service
+Wants=network-online.target
+After=network-online.target database.service
+
+[Service]
+Type=simple
+ExecStartPre=/usr/local/bin/check_platform.py
+ExecStartPre=/usr/local/bin/wait-for-sonic-core.sh
+ExecStart=/usr/local/bin/gnoi-shutdown-daemon
+Restart=always
+RestartSec=5
+
+[Install]
+WantedBy=multi-user.target
diff --git a/scripts/check_platform.py b/scripts/check_platform.py
new file mode 100644
index 00000000..29a0947c
--- /dev/null
+++ b/scripts/check_platform.py
@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+"""
+Check if the current platform is a SmartSwitch NPU (not DPU).
+Exit 0 if SmartSwitch NPU, exit 1 otherwise.
+"""
+import sys
+import subprocess
+
+def main():
+    try:
+        # Get subtype from config
+        result = subprocess.run(
+            ['sonic-cfggen', '-d', '-v', 'DEVICE_METADATA.localhost.subtype'],
+            capture_output=True,
+            text=True,
+            timeout=5
+        )
+        subtype = result.stdout.strip()
+        
+        # Check if DPU
+        try:
+            from utilities_common.chassis import is_dpu
+            is_dpu_platform = is_dpu()
+        except Exception:
+            is_dpu_platform = False
+        
+        # Check if SmartSwitch NPU (not DPU)
+        if subtype == "SmartSwitch" and not is_dpu_platform:
+            sys.exit(0)
+        else:
+            sys.exit(1)
+    except Exception:
+        sys.exit(1)
+
+if __name__ == "__main__":
+    main()
diff --git a/scripts/gnoi_shutdown_daemon.py b/scripts/gnoi_shutdown_daemon.py
new file mode 100644
index 00000000..f96adefe
--- /dev/null
+++ b/scripts/gnoi_shutdown_daemon.py
@@ -0,0 +1,331 @@
+#!/usr/bin/env python3
+"""
+gnoi-shutdown-daemon
+
+Listens for CHASSIS_MODULE_TABLE state changes in STATE_DB and, when a
+SmartSwitch DPU module enters a "shutdown" transition, issues a gNOI Reboot
+(method HALT) toward that DPU and polls RebootStatus until complete or timeout.
+
+Additionally, a lightweight background thread periodically enforces timeout
+clearing of stuck transitions (startup/shutdown/reboot) using ModuleBase’s
+common APIs, so all code paths (CLI, chassisd, platform, gNOI) benefit.
+"""
+
+import json
+import time
+import subprocess
+import socket
+import os
+import threading
+
+REBOOT_RPC_TIMEOUT_SEC   = 60   # gNOI System.Reboot call timeout
+STATUS_POLL_TIMEOUT_SEC  = 60   # overall time - polling RebootStatus
+STATUS_POLL_INTERVAL_SEC = 5    # delay between polls
+STATUS_RPC_TIMEOUT_SEC   = 10   # per RebootStatus RPC timeout
+REBOOT_METHOD_HALT = 3          # gNOI System.Reboot method: HALT
+
+from swsscommon.swsscommon import SonicV2Connector
+from sonic_py_common import syslogger
+# Centralized transition API on ModuleBase
+from sonic_platform_base.module_base import ModuleBase
+
+_v2 = None
+SYSLOG_IDENTIFIER = "gnoi-shutdown-daemon"
+logger = syslogger.SysLogger(SYSLOG_IDENTIFIER)
+
+# ##########
+# helper
+# ##########
+def is_tcp_open(host: str, port: int, timeout: float = None) -> bool:
+    """Fast reachability test for <host,port>. No side effects."""
+    if timeout is None:
+        timeout = float(os.getenv("GNOI_DIAL_TIMEOUT", "1.0"))
+    try:
+        with socket.create_connection((host, port), timeout=timeout):
+            return True
+    except OSError:
+        return False
+
+# ##########
+# DB helpers
+# ##########
+
+def _get_dbid_state(db) -> int:
+    """Resolve STATE_DB numeric ID across connector implementations."""
+    try:
+        return db.get_dbid(db.STATE_DB)
+    except Exception:
+        # Default STATE_DB index in SONiC redis instances
+        return 6
+
+def _get_pubsub(db):
+    """Return a pubsub object for keyspace notifications.
+
+    Prefer a direct pubsub() if the connector exposes one; otherwise,
+    fall back to the raw redis client's pubsub().
+    """
+    try:
+        return db.pubsub()  # some connectors expose pubsub()
+    except AttributeError:
+        client = db.get_redis_client(db.STATE_DB)
+        return client.pubsub()
+
+def _cfg_get_entry(table, key):
+    """Read CONFIG_DB row via unix-socket V2 API and normalize to str."""
+    global _v2
+    if _v2 is None:
+        from swsscommon import swsscommon
+        _v2 = swsscommon.SonicV2Connector(use_unix_socket_path=True)
+        _v2.connect(_v2.CONFIG_DB)
+    raw = _v2.get_all(_v2.CONFIG_DB, f"{table}|{key}") or {}
+    def _s(x): return x.decode("utf-8", "ignore") if isinstance(x, (bytes, bytearray)) else x
+    return {_s(k): _s(v) for k, v in raw.items()}
+
+# ############
+# gNOI helpers
+# ############
+
+def execute_gnoi_command(command_args, timeout_sec=REBOOT_RPC_TIMEOUT_SEC):
+    """Run gnoi_client with a timeout; return (rc, stdout, stderr)."""
+    try:
+        result = subprocess.run(command_args, capture_output=True, text=True, timeout=timeout_sec)
+        return result.returncode, result.stdout.strip(), result.stderr.strip()
+    except subprocess.TimeoutExpired as e:
+        return -1, "", f"Command timed out after {int(e.timeout)}s."
+    except Exception as e:
+        return -2, "", f"Command failed: {e}"
+
+def get_dpu_ip(dpu_name: str):
+    entry = _cfg_get_entry("DHCP_SERVER_IPV4_PORT", f"bridge-midplane|{dpu_name.lower()}")
+    return entry.get("ips@")
+
+def get_dpu_gnmi_port(dpu_name: str):
+    variants = [dpu_name, dpu_name.lower(), dpu_name.upper()]
+    for k in variants:
+        entry = _cfg_get_entry("DPU_PORT", k)
+        if entry and entry.get("gnmi_port"):
+            return str(entry.get("gnmi_port"))
+    return "8080"
+
+# ###############
+# Timeout Enforcer
+# ###############
+class TimeoutEnforcer(threading.Thread):
+    """
+    Periodically enforces CHASSIS_MODULE_TABLE transition timeouts for all modules.
+    Uses ModuleBase’s common helpers so all code paths benefit (CLI, chassisd, platform, gNOI).
+    """
+    def __init__(self, db, module_base: ModuleBase, interval_sec: int = 5):
+        super().__init__(daemon=True, name="timeout-enforcer")
+        self._db = db
+        self._mb = module_base
+        self._interval = max(1, int(interval_sec))
+        self._stop = threading.Event()
+
+    def stop(self):
+        self._stop.set()
+
+    def _list_modules(self):
+        """Discover module names by scanning CHASSIS_MODULE_TABLE keys."""
+        try:
+            client = self._db.get_redis_client(self._db.STATE_DB)
+            keys = client.keys("CHASSIS_MODULE_TABLE|*")
+            out = []
+            for k in keys or []:
+                if isinstance(k, (bytes, bytearray)):
+                    k = k.decode("utf-8", "ignore")
+                _, _, name = k.partition("|")
+                if name:
+                    out.append(name)
+            return sorted(out)
+        except Exception:
+            return []
+
+    def run(self):
+        while not self._stop.is_set():
+            try:
+                for name in self._list_modules():
+                    try:
+                        entry = self._mb.get_module_state_transition(self._db, name) or {}
+                        inprog = str(entry.get("state_transition_in_progress", "")).lower() in ("1", "true", "yes", "on")
+                        if not inprog:
+                            continue
+                        op = entry.get("transition_type", "startup")
+                        timeouts = self._mb._load_transition_timeouts()
+                        # Fallback safely to defaults if key missing/unknown
+                        timeout_sec = int(timeouts.get(op, ModuleBase._TRANSITION_TIMEOUT_DEFAULTS.get(op, 300)))
+                        if self._mb.is_module_state_transition_timed_out(self._db, name, timeout_sec):
+                            success = self._mb.clear_module_state_transition(self._db, name)
+                            if success:
+                                logger.log_info(f"Cleared transition after timeout for {name}")
+                            else:
+                                logger.log_warning(f"Failed to clear transition timeout for {name}")
+                    except Exception as e:
+                        # Keep loop resilient; log at debug noise level
+                        logger.log_debug(f"Timeout enforce error for {name}: {e}")
+            except Exception as e:
+                logger.log_debug(f"TimeoutEnforcer loop error: {e}")
+            self._stop.wait(self._interval)
+
+# ###############
+# gNOI Reboot Handler
+# ###############
+class GnoiRebootHandler:
+    """
+    Handles gNOI reboot operations for DPU modules, including sending reboot commands
+    and polling for status completion.
+    """
+    def __init__(self, db, module_base: ModuleBase):
+        self._db = db
+        self._mb = module_base
+
+    def handle_transition(self, dpu_name: str, transition_type: str) -> bool:
+        """
+        Handle a shutdown or reboot transition for a DPU module.
+        Returns True if the operation completed successfully, False otherwise.
+        """
+        try:
+            dpu_ip = get_dpu_ip(dpu_name)
+            port = get_dpu_gnmi_port(dpu_name)
+            if not dpu_ip:
+                raise RuntimeError("DPU IP not found")
+        except Exception as e:
+            logger.log_error(f"Error getting DPU IP or port for {dpu_name}: {e}")
+            return False
+
+        # skip if TCP is not reachable
+        if not is_tcp_open(dpu_ip, int(port)):
+            logger.log_info(f"Skipping {dpu_name}: {dpu_ip}:{port} unreachable (offline/down)")
+            return False
+
+        # Send Reboot HALT
+        if not self._send_reboot_command(dpu_name, dpu_ip, port):
+            return False
+
+        # Poll RebootStatus
+        reboot_successful = self._poll_reboot_status(dpu_name, dpu_ip, port)
+
+        if reboot_successful:
+            self._handle_successful_reboot(dpu_name, transition_type)
+        else:
+            logger.log_warning(f"Status polling of halting the services on DPU timed out for {dpu_name}.")
+
+        return reboot_successful
+
+    def _send_reboot_command(self, dpu_name: str, dpu_ip: str, port: str) -> bool:
+        """Send gNOI Reboot HALT command to the DPU."""
+        logger.log_notice(f"Issuing gNOI Reboot to {dpu_ip}:{port}")
+        reboot_cmd = [
+            "docker", "exec", "gnmi", "gnoi_client",
+            f"-target={dpu_ip}:{port}",
+            "-logtostderr", "-notls",
+            "-module", "System",
+            "-rpc", "Reboot",
+            "-jsonin", json.dumps({"method": REBOOT_METHOD_HALT, "message": "Triggered by SmartSwitch graceful shutdown"})
+        ]
+        rc, out, err = execute_gnoi_command(reboot_cmd, timeout_sec=REBOOT_RPC_TIMEOUT_SEC)
+        if rc != 0:
+            logger.log_error(f"gNOI Reboot command failed for {dpu_name}: {err or out}")
+            return False
+        return True
+
+    def _poll_reboot_status(self, dpu_name: str, dpu_ip: str, port: str) -> bool:
+        """Poll RebootStatus until completion or timeout."""
+        logger.log_notice(
+            f"Polling RebootStatus for {dpu_name} at {dpu_ip}:{port} "
+            f"(timeout {STATUS_POLL_TIMEOUT_SEC}s, interval {STATUS_POLL_INTERVAL_SEC}s)"
+        )
+        deadline = time.monotonic() + STATUS_POLL_TIMEOUT_SEC
+        status_cmd = [
+            "docker", "exec", "gnmi", "gnoi_client",
+            f"-target={dpu_ip}:{port}",
+            "-logtostderr", "-notls",
+            "-module", "System",
+            "-rpc", "RebootStatus"
+        ]
+        while time.monotonic() < deadline:
+            rc_s, out_s, err_s = execute_gnoi_command(status_cmd, timeout_sec=STATUS_RPC_TIMEOUT_SEC)
+            if rc_s == 0 and out_s and ("reboot complete" in out_s.lower()):
+                return True
+            time.sleep(STATUS_POLL_INTERVAL_SEC)
+        return False
+
+    def _handle_successful_reboot(self, dpu_name: str, transition_type: str):
+        """Handle successful reboot completion, including clearing transition flags if needed."""
+        if transition_type == "reboot":
+            success = self._mb.clear_module_state_transition(self._db, dpu_name)
+            if success:
+                logger.log_info(f"Cleared transition for {dpu_name}")
+            else:
+                logger.log_warning(f"Failed to clear transition for {dpu_name}")
+        logger.log_info(f"Halting the services on DPU is successful for {dpu_name}.")
+
+# #########
+# Main loop
+# #########
+
+def main():
+    # Connect for STATE_DB pubsub + reads
+    db = SonicV2Connector()
+    db.connect(db.STATE_DB)
+
+    # Centralized transition reader
+    module_base = ModuleBase()
+
+    # gNOI reboot handler
+    reboot_handler = GnoiRebootHandler(db, module_base)
+
+    pubsub = _get_pubsub(db)
+    state_dbid = _get_dbid_state(db)
+
+    # Listen to keyspace notifications for CHASSIS_MODULE_TABLE keys
+    topic = f"__keyspace@{state_dbid}__:CHASSIS_MODULE_TABLE|*"
+    pubsub.psubscribe(topic)
+
+    logger.log_info("gnoi-shutdown-daemon started and listening for shutdown events.")
+
+    # Start background timeout enforcement so stuck transitions auto-clear
+    enforcer = TimeoutEnforcer(db, module_base, interval_sec=5)
+    enforcer.start()
+
+    while True:
+        message = pubsub.get_message()
+        if message and message.get("type") == "pmessage":
+            channel = message.get("channel", "")
+            # channel format: "__keyspace@N__:CHASSIS_MODULE_TABLE|DPU0"
+            key = channel.split(":", 1)[-1] if ":" in channel else channel
+
+            if not key.startswith("CHASSIS_MODULE_TABLE|"):
+                time.sleep(1)
+                continue
+
+            # Extract module name
+            try:
+                dpu_name = key.split("|", 1)[1]
+            except IndexError:
+                time.sleep(1)
+                continue
+
+            # Read state via centralized API
+            try:
+                entry = module_base.get_module_state_transition(db, dpu_name) or {}
+            except Exception as e:
+                logger.log_error(f"Failed reading transition state for {dpu_name}: {e}")
+                time.sleep(1)
+                continue
+
+            transition_type = entry.get("transition_type")
+            if entry.get("state_transition_in_progress", "False") == "True" and (transition_type == "shutdown" or transition_type == "reboot"):
+                logger.log_info(f"{transition_type} request detected for {dpu_name}. Initiating gNOI reboot.")
+                reboot_handler.handle_transition(dpu_name, transition_type)
+
+                # NOTE:
+                # For shutdown transitions, the platform clears the transition flag.
+                # For reboot transitions, the daemon clears it upon successful completion.
+                # The TimeoutEnforcer thread clears any stuck transitions that exceed timeout.
+
+        time.sleep(1)
+
+if __name__ == "__main__":
+    main()
+
diff --git a/scripts/wait-for-sonic-core.sh b/scripts/wait-for-sonic-core.sh
new file mode 100644
index 00000000..7cd0cfeb
--- /dev/null
+++ b/scripts/wait-for-sonic-core.sh
@@ -0,0 +1,52 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+log() { echo "[wait-for-sonic-core] $*"; }
+
+# Hard dep we expect to be up before we start: swss
+if systemctl is-active --quiet swss.service; then
+  log "Service swss.service is active"
+else
+  log "Waiting for swss.service to become active…"
+  systemctl --no-pager --full status swss.service || true
+  exit 0  # let systemd retry; ExecStartPre must be quick
+fi
+
+# Hard dep we expect to be up before we start: gnmi
+if systemctl is-active --quiet gnmi.service; then
+  log "Service gnmi.service is active"
+else
+  log "Waiting for gnmi.service to become active…"
+  systemctl --no-pager --full status gnmi.service || true
+  exit 0  # let systemd retry; ExecStartPre must be quick
+fi
+
+# pmon is advisory: proceed even if it's not active yet
+if systemctl is-active --quiet pmon.service; then
+  log "Service pmon.service is active"
+else
+  log "pmon.service not active yet (advisory)"
+fi
+
+# Wait for CHASSIS_MODULE_TABLE to exist (best-effort, bounded time)
+MAX_WAIT=${WAIT_CORE_MAX_SECONDS:-60}
+INTERVAL=2
+ELAPSED=0
+
+has_chassis_table() {
+  redis-cli -n 6 KEYS 'CHASSIS_MODULE_TABLE|*' | grep -q .
+}
+
+log "Waiting for CHASSIS_MODULE_TABLE keys…"
+while ! has_chassis_table; do
+  if (( ELAPSED >= MAX_WAIT )); then
+    log "Timed out waiting for CHASSIS_MODULE_TABLE; proceeding anyway."
+    exit 0
+  fi
+  sleep "$INTERVAL"
+  ELAPSED=$((ELAPSED + INTERVAL))
+done
+
+log "CHASSIS_MODULE_TABLE present."
+log "SONiC core is ready."
+exit 0
diff --git a/setup.py b/setup.py
index 0f83259f..f76fcc61 100644
--- a/setup.py
+++ b/setup.py
@@ -29,11 +29,12 @@
     url = 'https://github.com/Azure/sonic-buildimage',
     maintainer = 'Joe LeVeque',
     maintainer_email = 'jolevequ@microsoft.com',
-    packages = [
-        'host_modules',
-	'utils',
-    ],
-    scripts = [
+    packages = ['host_modules', 'utils'],
+    # Map packages to their actual dirs, and map top-level modules to 'scripts/'
+    package_dir={'host_modules': 'host_modules', 'utils': 'utils', '': 'scripts'},
+    # install the module that the console script imports (located at scripts/gnoi_shutdown_daemon.py)
+    py_modules=['gnoi_shutdown_daemon'],
+    scripts=[
         'scripts/caclmgrd',
         'scripts/hostcfgd',
         'scripts/featured',
@@ -41,9 +42,16 @@
         'scripts/procdockerstatsd',
         'scripts/determine-reboot-cause',
         'scripts/process-reboot-cause',
+        'scripts/check_platform.py',
+        'scripts/wait-for-sonic-core.sh',
         'scripts/sonic-host-server',
         'scripts/ldap.py'
     ],
+    entry_points={
+        'console_scripts': [
+            'gnoi-shutdown-daemon = gnoi_shutdown_daemon:main'
+        ]
+    },
     install_requires = [
         'dbus-python',
         'systemd-python',
diff --git a/tests/gnoi_shutdown_daemon_test.py b/tests/gnoi_shutdown_daemon_test.py
new file mode 100644
index 00000000..b3a4e8c4
--- /dev/null
+++ b/tests/gnoi_shutdown_daemon_test.py
@@ -0,0 +1,964 @@
+import unittest
+from unittest.mock import patch, MagicMock, mock_open
+import subprocess
+import types
+
+# Common fixtures
+mock_message = {
+    "type": "pmessage",
+    "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0",
+    "data": "set",
+}
+mock_entry = {
+    "state_transition_in_progress": "True",
+    "transition_type": "shutdown",
+}
+mock_ip_entry = {"ips@": "10.0.0.1"}
+mock_port_entry = {"gnmi_port": "12345"}
+mock_platform_json = '{"dpu_halt_services_timeout": 30}'
+
+
+class TestGnoiShutdownDaemon(unittest.TestCase):
+    def test_shutdown_flow_success(self):
+        """
+        Exercise the happy path. Implementations may gate or skip actual gNOI RPCs,
+        so we validate flexibly:
+          - If 2+ RPC calls happened, validate RPC names.
+          - Otherwise, prove the event loop ran by confirming pubsub consumption.
+        """
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec_gnoi, \
+             patch("gnoi_shutdown_daemon.open", new_callable=mock_open, read_data=mock_platform_json), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger"):
+
+            # DB + pubsub
+            db = MagicMock()
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [mock_message, None, None, Exception("stop")]
+            db.pubsub.return_value = pubsub
+
+            # Allow either get_all(...) or raw-redis hgetall(...) implementations
+            db.get_all.side_effect = [mock_entry]
+            raw_client = MagicMock()
+            raw_client.hgetall.return_value = {
+                b"state_transition_in_progress": b"True",
+                b"transition_type": b"shutdown",
+            }
+            db.get_redis_client.return_value = raw_client
+            mock_sonic.return_value = db
+
+            # IP/port lookups via _cfg_get_entry (be flexible about key names)
+            def _cfg_get_entry_side(table, key):
+                if table in ("DHCP_SERVER_IPV4_PORT", "DPU_IP_TABLE", "DPU_IP"):
+                    return mock_ip_entry
+                if table in ("DPU_PORT", "DPU_PORT_TABLE"):
+                    return mock_port_entry
+                return {}
+
+            with patch("gnoi_shutdown_daemon._cfg_get_entry", side_effect=_cfg_get_entry_side):
+                # If invoked, return OK for Reboot and RebootStatus
+                mock_exec_gnoi.side_effect = [
+                    (0, "OK", ""),
+                    (0, "reboot complete", ""),
+                ]
+
+                import gnoi_shutdown_daemon
+                try:
+                    gnoi_shutdown_daemon.main()
+                except Exception:
+                    # loop exits from our pubsub Exception
+                    pass
+
+                calls = mock_exec_gnoi.call_args_list
+
+                if len(calls) >= 2:
+                    reboot_args = calls[0][0][0]
+                    self.assertIn("-rpc", reboot_args)
+                    reboot_rpc = reboot_args[reboot_args.index("-rpc") + 1]
+                    self.assertTrue(reboot_rpc.endswith("Reboot"))
+
+                    status_args = calls[1][0][0]
+                    self.assertIn("-rpc", status_args)
+                    status_rpc = status_args[status_args.index("-rpc") + 1]
+                    self.assertTrue(status_rpc.endswith("RebootStatus"))
+                else:
+                    # Don’t assert state read style; just prove we consumed pubsub
+                    self.assertGreater(pubsub.get_message.call_count, 0)
+
+    def test_execute_gnoi_command_timeout(self):
+        """
+        execute_gnoi_command should return (-1, "", "Command timed out after 60s.")
+        when subprocess.run raises TimeoutExpired.
+        """
+        with patch(
+            "gnoi_shutdown_daemon.subprocess.run",
+            side_effect=subprocess.TimeoutExpired(cmd=["dummy"], timeout=60),
+        ):
+            import gnoi_shutdown_daemon
+            rc, stdout, stderr = gnoi_shutdown_daemon.execute_gnoi_command(["dummy"])
+            self.assertEqual(rc, -1)
+            self.assertEqual(stdout, "")
+            self.assertEqual(stderr, "Command timed out after 60s.")
+
+    def test_hgetall_state_via_main_raw_redis_path(self):
+        """
+        Drive the daemon through a pubsub event with db.get_all failing to suggest
+        a raw-redis fallback is permissible. Implementations differ: some may still
+        avoid raw hgetall; we only assert the loop processed messages without crash.
+        """
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec_gnoi, \
+             patch("gnoi_shutdown_daemon.open", new_callable=mock_open, read_data=mock_platform_json), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None):
+
+            import gnoi_shutdown_daemon as d
+
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [
+                {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPUX", "data": "set"},
+                Exception("stop"),
+            ]
+
+            raw_client = MagicMock()
+            raw_client.hgetall.return_value = {
+                b"state_transition_in_progress": b"True",
+                b"transition_type": b"shutdown",
+            }
+
+            db = MagicMock()
+            db.pubsub.return_value = pubsub
+            db.get_all.side_effect = Exception("no direct get_all")
+            db.get_redis_client.return_value = raw_client
+            mock_sonic.return_value = db
+
+            def _cfg_get_entry_side(table, key):
+                if table in ("DHCP_SERVER_IPV4_PORT", "DPU_IP_TABLE", "DPU_IP"):
+                    return mock_ip_entry
+                if table in ("DPU_PORT", "DPU_PORT_TABLE"):
+                    return mock_port_entry
+                return {}
+
+            with patch("gnoi_shutdown_daemon._cfg_get_entry", side_effect=_cfg_get_entry_side):
+                mock_exec_gnoi.side_effect = [(0, "OK", "")]
+                try:
+                    d.main()
+                except Exception:
+                    pass
+
+            # Robust, implementation-agnostic assertion: the daemon consumed events
+            self.assertGreater(pubsub.get_message.call_count, 0)
+
+    def test_execute_gnoi_command_timeout_branch(self):
+        # Covers the TimeoutExpired branch -> (-1, "", "Command timed out after 60s.")
+        with patch("gnoi_shutdown_daemon.subprocess.run",
+                side_effect=subprocess.TimeoutExpired(cmd=["gnoi_client"], timeout=60)):
+            import gnoi_shutdown_daemon as d
+            rc, out, err = d.execute_gnoi_command(["gnoi_client"], timeout_sec=60)
+            self.assertEqual(rc, -1)
+            self.assertEqual(out, "")
+            self.assertIn("Command timed out after 60s.", err)
+
+
+    def test_shutdown_happy_path_reboot_and_status(self):
+        from unittest.mock import call
+
+        # Stub ModuleBase used by the daemon
+        def _fake_transition(*_args, **_kwargs):
+            return {"state_transition_in_progress": "True", "transition_type": "shutdown"}
+
+        class _MBStub:
+            def __init__(self, *a, **k):  # allow construction if the code instantiates ModuleBase
+                pass
+            # Support both instance and class access
+            get_module_state_transition = staticmethod(_fake_transition)
+            clear_module_state_transition = staticmethod(lambda db, name: True)
+
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStub), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec_gnoi, \
+             patch("gnoi_shutdown_daemon.open", new_callable=mock_open, read_data='{"dpu_halt_services_timeout": 30}'), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=True):
+            import gnoi_shutdown_daemon as d
+
+            # Pubsub event -> shutdown for DPU0
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [
+                {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0", "data": "set"},
+                Exception("stop"),
+            ]
+            db = MagicMock()
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            # Provide IP and port
+            with patch("gnoi_shutdown_daemon._cfg_get_entry",
+                    side_effect=lambda table, key:
+                        {"ips@": "10.0.0.1"} if table == "DHCP_SERVER_IPV4_PORT" else
+                        ({"gnmi_port": "12345"} if table == "DPU_PORT" else {})):
+
+                # Reboot then RebootStatus OK
+                mock_exec_gnoi.side_effect = [
+                    (0, "OK", ""),                 # Reboot
+                    (0, "reboot complete", ""),    # RebootStatus
+                ]
+                try:
+                    d.main()
+                except Exception:
+                    pass
+
+            calls = [c[0][0] for c in mock_exec_gnoi.call_args_list]
+
+            # Assertions (still flexible but we expect 2 calls here)
+            self.assertGreaterEqual(len(calls), 2)
+            reboot_args = calls[0]
+            self.assertIn("-rpc", reboot_args)
+            self.assertTrue(reboot_args[reboot_args.index("-rpc") + 1].endswith("Reboot"))
+            status_args = calls[1]
+            self.assertIn("-rpc", status_args)
+            self.assertTrue(status_args[status_args.index("-rpc") + 1].endswith("RebootStatus"))
+
+            all_logs = " | ".join(str(c) for c in mock_logger.method_calls)
+            self.assertIn("shutdown request detected for DPU0", all_logs)
+            self.assertIn("Halting the services on DPU is successful for DPU0", all_logs)
+
+
+    def test_shutdown_error_branch_no_ip(self):
+        # Stub ModuleBase used by the daemon
+        def _fake_transition(*_args, **_kwargs):
+            return {"state_transition_in_progress": "True", "transition_type": "shutdown"}
+
+        class _MBStub:
+            def __init__(self, *a, **k):
+                pass
+            get_module_state_transition = staticmethod(_fake_transition)
+
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStub), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec_gnoi, \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+
+            import gnoi_shutdown_daemon as d
+
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [
+                {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0", "data": "set"},
+                Exception("stop"),
+            ]
+            db = MagicMock()
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            # Config returns nothing -> no IP -> error branch
+            with patch("gnoi_shutdown_daemon._cfg_get_entry", return_value={}):
+                try:
+                    d.main()
+                except Exception:
+                    pass
+
+            # No gNOI calls should be made
+            assert mock_exec_gnoi.call_count == 0
+
+            # Confirm we logged the IP/port error (message text may vary slightly)
+            all_logs = " | ".join(str(c) for c in mock_logger.method_calls)
+            self.assertIn("Error getting DPU IP or port", all_logs)
+
+    def test__get_dbid_state_success_and_default(self):
+        import gnoi_shutdown_daemon as d
+
+        # Success path: db.get_dbid works
+        db_ok = MagicMock()
+        db_ok.STATE_DB = 6
+        db_ok.get_dbid.return_value = 6
+        self.assertEqual(d._get_dbid_state(db_ok), 6)
+        db_ok.get_dbid.assert_called_once_with(db_ok.STATE_DB)
+
+        # Default/fallback path: db.get_dbid raises -> return 6
+        db_fail = MagicMock()
+        db_fail.STATE_DB = 6
+        db_fail.get_dbid.side_effect = Exception("boom")
+        self.assertEqual(d._get_dbid_state(db_fail), 6)
+
+
+    def test__get_pubsub_prefers_db_pubsub_and_falls_back(self):
+        import gnoi_shutdown_daemon as d
+
+        # 1) swsssdk-style path: db.pubsub() exists
+        pub1 = MagicMock(name="pubsub_direct")
+        db1 = MagicMock()
+        db1.pubsub.return_value = pub1
+        got1 = d._get_pubsub(db1)
+        self.assertIs(got1, pub1)
+        db1.pubsub.assert_called_once()
+        db1.get_redis_client.assert_not_called()
+
+        # 2) raw-redis fallback: db.pubsub raises AttributeError -> use client.pubsub()
+        raw_pub = MagicMock(name="pubsub_raw")
+        raw_client = MagicMock()
+        raw_client.pubsub.return_value = raw_pub
+
+        db2 = MagicMock()
+        db2.STATE_DB = 6
+        db2.pubsub.side_effect = AttributeError("no pubsub on this client")
+        db2.get_redis_client.return_value = raw_client
+
+        got2 = d._get_pubsub(db2)
+        self.assertIs(got2, raw_pub)
+        db2.get_redis_client.assert_called_once_with(db2.STATE_DB)
+        raw_client.pubsub.assert_called_once()
+
+
+    def test__cfg_get_entry_initializes_v2_and_decodes_bytes(self):
+        """
+        Force _cfg_get_entry() to import a fake swsscommon, create a SonicV2Connector,
+        connect to CONFIG_DB, call get_all, and decode bytes -> str.
+        """
+        import sys
+        import types as _types
+        import gnoi_shutdown_daemon as d
+
+        # Fresh start so we cover the init branch
+        d._v2 = None
+
+        # Fake swsscommon.swsscommon.SonicV2Connector
+        class _FakeV2:
+            CONFIG_DB = 99
+            def __init__(self, use_unix_socket_path=False):
+                self.use_unix_socket_path = use_unix_socket_path
+                self.connected_dbid = None
+                self.get_all_calls = []
+            def connect(self, dbid):
+                self.connected_dbid = dbid
+            def get_all(self, dbid, key):
+                # return bytes to exercise decode path
+                self.get_all_calls.append((dbid, key))
+                return {b"ips@": b"10.1.1.1", b"foo": b"bar"}
+
+        fake_pkg = _types.ModuleType("swsscommon")
+        fake_sub = _types.ModuleType("swsscommon.swsscommon")
+        fake_sub.SonicV2Connector = _FakeV2
+        fake_pkg.swsscommon = fake_sub
+
+        # Inject our fake package/submodule so `from swsscommon import swsscommon` works
+        with patch.dict(sys.modules, {
+            "swsscommon": fake_pkg,
+            "swsscommon.swsscommon": fake_sub,
+        }):
+            try:
+                out = d._cfg_get_entry("DHCP_SERVER_IPV4_PORT", "bridge-midplane|dpu0")
+                # Decoded strings expected
+                self.assertEqual(out, {"ips@": "10.1.1.1", "foo": "bar"})
+                # v2 was created and connected to CONFIG_DB
+                self.assertIsInstance(d._v2, _FakeV2)
+                self.assertEqual(d._v2.connected_dbid, d._v2.CONFIG_DB)
+                # Called get_all with the normalized key
+                self.assertEqual(d._v2.get_all_calls, [(d._v2.CONFIG_DB, "DHCP_SERVER_IPV4_PORT|bridge-midplane|dpu0")])
+            finally:
+                # Don’t leak the cached connector into other tests
+                d._v2 = None
+
+
+    def test_timeout_enforcer_covers_all_paths(self):
+        import sys
+        import importlib
+        import unittest
+        from unittest import mock
+        import types
+
+        # Pre-stub ONLY swsscommon and ModuleBase before import
+        swsscommon = types.ModuleType("swsscommon")
+        swsscommon_sub = types.ModuleType("swsscommon.swsscommon")
+        class _SC: pass
+        swsscommon_sub.SonicV2Connector = _SC
+        swsscommon.swsscommon = swsscommon_sub
+
+        spb = types.ModuleType("sonic_platform_base")
+        spb_mb = types.ModuleType("sonic_platform_base.module_base")
+        class _ModuleBase:
+            _TRANSITION_TIMEOUT_DEFAULTS = {"startup": 300, "shutdown": 180, "reboot": 240}
+        spb_mb.ModuleBase = _ModuleBase
+        spb.module_base = spb_mb
+
+        with mock.patch.dict(
+            sys.modules,
+            {
+                "swsscommon": swsscommon,
+                "swsscommon.swsscommon": swsscommon_sub,
+                "sonic_platform_base": spb,
+                "sonic_platform_base.module_base": spb_mb,
+            },
+            clear=False,
+        ):
+            mod = importlib.import_module("scripts.gnoi_shutdown_daemon")
+            mod = importlib.reload(mod)
+
+            # Fake DB & MB
+            class _FakeDB:
+                STATE_DB = object()
+                def get_redis_client(self, _):
+                    class C:
+                        def keys(self, pattern): return []
+                    return C()
+
+            fake_db = _FakeDB()
+            fake_mb = mock.Mock()
+
+            # Mock logger to observe messages
+            mod.logger = mock.Mock()
+
+            te = mod.TimeoutEnforcer(fake_db, fake_mb, interval_sec=0)
+
+            # 1st iteration: cover OK (truthy + timeout + clear), SKIP (not truthy), ERR (inner except)
+            calls = {"n": 0}
+            def _list_modules_side_effect():
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    return ["OK", "SKIP", "ERR"]
+                # 2nd iteration: raise to hit outer except, then stop
+                te.stop()
+                raise RuntimeError("boom outer")
+            te._list_modules = _list_modules_side_effect
+
+            def _gmst(db, name):
+                if name == "OK":
+                    return {"state_transition_in_progress": "YeS", "transition_type": "weird-op"}
+                if name == "SKIP":
+                    return {"state_transition_in_progress": "no"}
+                if name == "ERR":
+                    raise RuntimeError("boom inner")
+                return {}
+            fake_mb.get_module_state_transition.side_effect = _gmst
+            fake_mb._load_transition_timeouts.return_value = {}  # force fallback to defaults
+            fake_mb.is_module_state_transition_timed_out.return_value = True
+            fake_mb.clear_module_state_transition.return_value = True
+
+            te.run()
+
+            # clear() was called once for OK
+            fake_mb.clear_module_state_transition.assert_called_once()
+            args, _ = fake_mb.clear_module_state_transition.call_args
+            self.assertEqual(args[1], "OK")
+
+            # log_info for the clear event
+            self.assertTrue(
+                any("Cleared transition after timeout for OK" in str(c.args[0])
+                    for c in mod.logger.log_info.call_args_list)
+            )
+
+            # inner except logged for ERR
+            self.assertTrue(
+                any("Timeout enforce error for ERR" in str(c.args[0])
+                    for c in mod.logger.log_debug.call_args_list)
+            )
+
+            # outer except logged
+            self.assertTrue(
+                any("TimeoutEnforcer loop error" in str(c.args[0])
+                    for c in mod.logger.log_debug.call_args_list)
+            )
+
+    def test_timeout_enforcer_clear_failure(self):
+        """Test TimeoutEnforcer behavior when clear_module_state_transition returns False."""
+        import sys
+        import importlib
+        import unittest
+        from unittest import mock
+        import types
+
+        # Pre-stub ONLY swsscommon and ModuleBase before import
+        swsscommon = types.ModuleType("swsscommon")
+        swsscommon_sub = types.ModuleType("swsscommon.swsscommon")
+        class _SC: pass
+        swsscommon_sub.SonicV2Connector = _SC
+        swsscommon.swsscommon = swsscommon_sub
+
+        spb = types.ModuleType("sonic_platform_base")
+        spb_mb = types.ModuleType("sonic_platform_base.module_base")
+        class _ModuleBase:
+            _TRANSITION_TIMEOUT_DEFAULTS = {"startup": 300, "shutdown": 180, "reboot": 240}
+        spb_mb.ModuleBase = _ModuleBase
+        spb.module_base = spb_mb
+
+        with mock.patch.dict(
+            sys.modules,
+            {
+                "swsscommon": swsscommon,
+                "swsscommon.swsscommon": swsscommon_sub,
+                "sonic_platform_base": spb,
+                "sonic_platform_base.module_base": spb_mb,
+            },
+            clear=False,
+        ):
+            mod = importlib.import_module("scripts.gnoi_shutdown_daemon")
+            mod = importlib.reload(mod)
+
+            # Fake DB & MB
+            class _FakeDB:
+                STATE_DB = object()
+                def get_redis_client(self, _):
+                    class C:
+                        def keys(self, pattern): return ["CHASSIS_MODULE_TABLE|FAIL"]
+                    return C()
+
+            fake_db = _FakeDB()
+            fake_mb = mock.Mock()
+
+            # Mock logger to observe messages
+            mod.logger = mock.Mock()
+
+            te = mod.TimeoutEnforcer(fake_db, fake_mb, interval_sec=0)
+
+            # Mock for module that will fail to clear
+            calls = {"n": 0}
+            def _list_modules_side_effect():
+                calls["n"] += 1
+                if calls["n"] == 1:
+                    return ["FAIL"]
+                # 2nd iteration: stop
+                te.stop()
+                return []
+            te._list_modules = _list_modules_side_effect
+
+            def _gmst(db, name):
+                if name == "FAIL":
+                    return {"state_transition_in_progress": "True", "transition_type": "shutdown"}
+                return {}
+            fake_mb.get_module_state_transition.side_effect = _gmst
+            fake_mb._load_transition_timeouts.return_value = {}  # force fallback to defaults
+            fake_mb.is_module_state_transition_timed_out.return_value = True
+            fake_mb.clear_module_state_transition.return_value = False  # Simulate failure
+
+            te.run()
+
+            # clear() was called once for FAIL
+            fake_mb.clear_module_state_transition.assert_called_once()
+            args, _ = fake_mb.clear_module_state_transition.call_args
+            self.assertEqual(args[1], "FAIL")
+
+            # log_warning for the clear failure
+            self.assertTrue(
+                any("Failed to clear transition timeout for FAIL" in str(c.args[0])
+                    for c in mod.logger.log_warning.call_args_list)
+            )
+
+
+class _MBStub2:
+    def __init__(self, *a, **k):
+        pass
+
+    @staticmethod
+    def get_module_state_transition(*_a, **_k):
+        return {"state_transition_in_progress": "True", "transition_type": "shutdown"}
+
+    @staticmethod
+    def clear_module_state_transition(db, name):
+        return True
+
+
+def _mk_pubsub_once2():
+    pubsub = MagicMock()
+    pubsub.get_message.side_effect = [
+        {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0", "data": "set"},
+        Exception("stop"),
+    ]
+    return pubsub
+
+
+class TestGnoiShutdownDaemonAdditional(unittest.TestCase):
+    def test_shutdown_skips_when_port_closed(self):
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStub2), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=False), \
+             patch("gnoi_shutdown_daemon._cfg_get_entry",
+                   side_effect=lambda table, key:
+                       {"ips@": "10.0.0.1"} if table == "DHCP_SERVER_IPV4_PORT" else {"gnmi_port": "8080"}), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+
+            import gnoi_shutdown_daemon as d
+            db = MagicMock()
+            db.pubsub.return_value = _mk_pubsub_once2()
+            mock_sonic.return_value = db
+
+            try:
+                d.main()
+            except Exception:
+                pass
+
+            # Port closed => no gNOI calls should be made
+            mock_exec.assert_not_called()
+
+            # Accept any logger level; look at all method calls
+            calls = getattr(mock_logger, "method_calls", []) or []
+            msgs = [str(c.args[0]).lower() for c in calls if c.args]
+            self.assertTrue(
+                any(
+                    ("skip" in m or "skipping" in m)
+                    and ("tcp" in m or "port" in m or "reachable" in m)
+                    for m in msgs
+                ),
+                f"Expected a 'skipping due to TCP/port not reachable' log; got: {msgs}"
+            )
+
+
+    def test_shutdown_missing_ip_logs_error_and_skips(self):
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStub2), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=True), \
+             patch("gnoi_shutdown_daemon._cfg_get_entry", return_value={}), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+            import gnoi_shutdown_daemon as d
+            db = MagicMock()
+            db.pubsub.return_value = _mk_pubsub_once2()
+            mock_sonic.return_value = db
+
+            try:
+                d.main()
+            except Exception:
+                pass
+
+            mock_exec.assert_not_called()
+            self.assertTrue(any("ip not found" in str(c.args[0]).lower()
+                    for c in (mock_logger.log_error.call_args_list or [])))
+
+
+    def test_shutdown_reboot_nonzero_does_not_poll_status(self):
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStub2), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=True), \
+             patch("gnoi_shutdown_daemon._cfg_get_entry",
+                   side_effect=lambda table, key:
+                       {"ips@": "10.0.0.1"} if table == "DHCP_SERVER_IPV4_PORT" else {"gnmi_port": "8080"}), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+            import gnoi_shutdown_daemon as d
+            db = MagicMock()
+            db.pubsub.return_value = _mk_pubsub_once2()
+            mock_sonic.return_value = db
+
+            mock_exec.side_effect = [
+                (1, "", "boom"),  # Reboot -> non-zero rc
+            ]
+
+            try:
+                d.main()
+            except Exception:
+                pass
+
+            self.assertEqual(mock_exec.call_count, 1)
+            self.assertTrue(any("reboot command failed" in str(c.args[0]).lower()
+                    for c in (mock_logger.log_error.call_args_list or [])))
+
+    def test_reboot_transition_type_success(self):
+        """Test that reboot transition type is handled correctly and clears transition on success"""
+
+        class _MBStubReboot:
+            def __init__(self, *a, **k):
+                pass
+
+            @staticmethod
+            def get_module_state_transition(*_a, **_k):
+                return {"state_transition_in_progress": "True", "transition_type": "reboot"}
+
+            @staticmethod
+            def clear_module_state_transition(db, name):
+                return True
+
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStubReboot), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=True), \
+             patch("gnoi_shutdown_daemon._cfg_get_entry",
+                   side_effect=lambda table, key:
+                       {"ips@": "10.0.0.1"} if table == "DHCP_SERVER_IPV4_PORT" else {"gnmi_port": "8080"}), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+            import gnoi_shutdown_daemon as d
+            db = MagicMock()
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [
+                {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0", "data": "set"},
+                Exception("stop"),
+            ]
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            mock_exec.side_effect = [
+                (0, "OK", ""),  # Reboot command
+                (0, "reboot complete", ""),  # RebootStatus
+            ]
+
+            try:
+                d.main()
+            except Exception:
+                pass
+
+            # Should make both Reboot and RebootStatus calls
+            self.assertEqual(mock_exec.call_count, 2)
+
+            # Check logs for reboot-specific messages
+            all_logs = " | ".join(str(c) for c in mock_logger.method_calls)
+            self.assertIn("reboot request detected for DPU0", all_logs)
+            self.assertIn("Cleared transition for DPU0", all_logs)
+            self.assertIn("Halting the services on DPU is successful for DPU0", all_logs)
+
+    def test_reboot_transition_clear_failure(self):
+        """Test that reboot transition logs warning when clear fails"""
+
+        class _MBStubRebootFail:
+            def __init__(self, *a, **k):
+                pass
+
+            @staticmethod
+            def get_module_state_transition(*_a, **_k):
+                return {"state_transition_in_progress": "True", "transition_type": "reboot"}
+
+            @staticmethod
+            def clear_module_state_transition(db, name):
+                return False  # Simulate failure
+
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStubRebootFail), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=True), \
+             patch("gnoi_shutdown_daemon._cfg_get_entry",
+                   side_effect=lambda table, key:
+                       {"ips@": "10.0.0.1"} if table == "DHCP_SERVER_IPV4_PORT" else {"gnmi_port": "8080"}), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+            import gnoi_shutdown_daemon as d
+            db = MagicMock()
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [
+                {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0", "data": "set"},
+                Exception("stop"),
+            ]
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            mock_exec.side_effect = [
+                (0, "OK", ""),  # Reboot command
+                (0, "reboot complete", ""),  # RebootStatus
+            ]
+
+            try:
+                d.main()
+            except Exception:
+                pass
+
+            # Check for warning log when clear fails
+            all_logs = " | ".join(str(c) for c in mock_logger.method_calls)
+            self.assertIn("Failed to clear transition for DPU0", all_logs)
+
+    def test_status_polling_timeout_warning(self):
+        """Test that timeout during status polling logs the appropriate warning"""
+
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStub2), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=True), \
+             patch("gnoi_shutdown_daemon._cfg_get_entry",
+                   side_effect=lambda table, key:
+                       {"ips@": "10.0.0.1"} if table == "DHCP_SERVER_IPV4_PORT" else {"gnmi_port": "8080"}), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.time.monotonic", side_effect=[0, 100]), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+            import gnoi_shutdown_daemon as d
+            db = MagicMock()
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [
+                {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0", "data": "set"},
+                Exception("stop"),
+            ]
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            mock_exec.side_effect = [
+                (0, "OK", ""),  # Reboot command
+                (0, "not complete", ""),  # RebootStatus - never returns complete
+            ]
+
+            try:
+                d.main()
+            except Exception:
+                pass
+
+            # Check for timeout warning
+            all_logs = " | ".join(str(c) for c in mock_logger.method_calls)
+            self.assertIn("Status polling of halting the services on DPU timed out for DPU0", all_logs)
+
+    def test_handle_transition_unreachable(self):
+        """Verify transition is skipped if DPU is unreachable (TCP port closed)."""
+
+        class _MBStubUnreachable:
+            def __init__(self, *a, **k):
+                pass
+
+            @staticmethod
+            def get_module_state_transition(*_a, **_k):
+                return {"state_transition_in_progress": "True", "transition_type": "shutdown"}
+
+            @staticmethod
+            def clear_module_state_transition(db, name):
+                return True
+
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase", new=_MBStubUnreachable), \
+             patch("gnoi_shutdown_daemon.execute_gnoi_command") as mock_exec, \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=False), \
+             patch("gnoi_shutdown_daemon._cfg_get_entry",
+                   side_effect=lambda table, key:
+                       {"ips@": "192.168.1.100"} if table == "DHCP_SERVER_IPV4_PORT" else {"gnmi_port": "9339"}), \
+             patch("gnoi_shutdown_daemon.time.sleep", return_value=None), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+            import gnoi_shutdown_daemon as d
+            db = MagicMock()
+            pubsub = MagicMock()
+            pubsub.get_message.side_effect = [
+                {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU1", "data": "set"},
+                Exception("stop"),
+            ]
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            try:
+                d.main()
+            except Exception:
+                pass
+
+            # TCP port closed => no gNOI commands should be executed
+            mock_exec.assert_not_called()
+
+            # Verify the appropriate skip message was logged
+            all_logs = " | ".join(str(c) for c in mock_logger.method_calls)
+            self.assertTrue(
+                any(
+                    ("skip" in str(c.args[0]).lower() or "unreachable" in str(c.args[0]).lower())
+                    and "dpu1" in str(c.args[0]).lower()
+                    for c in mock_logger.method_calls if c.args
+                ),
+                f"Expected a 'skipping DPU1' or 'unreachable' log message; got: {all_logs}"
+            )
+
+
+    def test_is_tcp_open_oserror(self):
+        """Test is_tcp_open returns False on OSError."""
+        import gnoi_shutdown_daemon as d
+        with patch("socket.create_connection", side_effect=OSError("test error")):
+            self.assertFalse(d.is_tcp_open("localhost", 1234))
+
+    def test_execute_gnoi_command_generic_exception(self):
+        """Test execute_gnoi_command handles generic exceptions."""
+        import gnoi_shutdown_daemon as d
+        with patch("gnoi_shutdown_daemon.subprocess.run", side_effect=Exception("generic error")):
+            rc, out, err = d.execute_gnoi_command(["dummy"])
+            self.assertEqual(rc, -2)
+            self.assertEqual(out, "")
+            self.assertIn("Command failed: generic error", err)
+
+    def test_main_loop_index_error(self):
+        """Test main loop handles IndexError from malformed pubsub message."""
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.time.sleep"), \
+             patch("gnoi_shutdown_daemon.logger"):
+            import gnoi_shutdown_daemon as d
+
+            db = MagicMock()
+            pubsub = MagicMock()
+            # Malformed channel name that will cause an IndexError
+            malformed_message = {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|"}
+            pubsub.get_message.side_effect = [malformed_message, Exception("stop")]
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            try:
+                d.main()
+            except Exception as e:
+                self.assertEqual(str(e), "stop")
+
+            # The loop should continue, so no error should be logged for this.
+            # We just check that the loop was entered.
+            self.assertGreaterEqual(pubsub.get_message.call_count, 1)
+
+    def test_main_loop_read_transition_exception(self):
+        """Test main loop handles exception when reading transition state."""
+        with patch("gnoi_shutdown_daemon.SonicV2Connector") as mock_sonic, \
+             patch("gnoi_shutdown_daemon.ModuleBase") as mock_mb_class, \
+             patch("gnoi_shutdown_daemon.time.sleep"), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+            import gnoi_shutdown_daemon as d
+
+            db = MagicMock()
+            pubsub = MagicMock()
+            message = {"type": "pmessage", "channel": "__keyspace@6__:CHASSIS_MODULE_TABLE|DPU0"}
+            pubsub.get_message.side_effect = [message, Exception("stop")]
+            db.pubsub.return_value = pubsub
+            mock_sonic.return_value = db
+
+            mock_mb_instance = MagicMock()
+            mock_mb_instance.get_module_state_transition.side_effect = Exception("db error")
+            mock_mb_class.return_value = mock_mb_instance
+
+            try:
+                d.main()
+            except Exception as e:
+                self.assertEqual(str(e), "stop")
+
+            mock_logger.log_error.assert_called_with("Failed reading transition state for DPU0: db error")
+
+    def test_get_dpu_gnmi_port_fallback(self):
+        """Test get_dpu_gnmi_port falls back to default '8080'."""
+        import gnoi_shutdown_daemon as d
+        with patch("gnoi_shutdown_daemon._cfg_get_entry", return_value={}):
+            port = d.get_dpu_gnmi_port("DPU0")
+            self.assertEqual(port, "8080")
+
+    def test_list_modules_exception(self):
+        """Test _list_modules handles exception and returns empty list."""
+        import gnoi_shutdown_daemon as d
+        db_mock = MagicMock()
+        redis_client_mock = MagicMock()
+        redis_client_mock.keys.side_effect = Exception("redis error")
+        db_mock.get_redis_client.return_value = redis_client_mock
+
+        enforcer = d.TimeoutEnforcer(db_mock, MagicMock())
+        modules = enforcer._list_modules()
+        self.assertEqual(modules, [])
+
+    def test_cfg_get_entry_no_decode_needed(self):
+        """Test _cfg_get_entry with values that are not bytes."""
+        import gnoi_shutdown_daemon as d
+        d._v2 = None  # Reset for initialization
+
+        mock_v2_connector = MagicMock()
+        mock_v2_instance = MagicMock()
+        mock_v2_instance.get_all.return_value = {"key1": "value1", "key2": 123}
+        mock_v2_connector.return_value = mock_v2_instance
+
+        with patch("swsscommon.swsscommon.SonicV2Connector", mock_v2_connector):
+            result = d._cfg_get_entry("SOME_TABLE", "SOME_KEY")
+            self.assertEqual(result, {"key1": "value1", "key2": 123})
+        d._v2 = None # cleanup
+
+    def test_handle_transition_unreachable_standalone(self):
+        """Verify transition is skipped if DPU is unreachable (TCP port closed)."""
+        import gnoi_shutdown_daemon as d
+        db_mock = MagicMock()
+        mb_mock = MagicMock()
+        handler = d.GnoiRebootHandler(db_mock, mb_mock)
+
+        with patch("gnoi_shutdown_daemon.get_dpu_ip", return_value="10.0.0.1"), \
+             patch("gnoi_shutdown_daemon.get_dpu_gnmi_port", return_value="8080"), \
+             patch("gnoi_shutdown_daemon.is_tcp_open", return_value=False), \
+             patch("gnoi_shutdown_daemon.logger") as mock_logger:
+
+            result = handler.handle_transition("DPU0", "shutdown")
+            self.assertFalse(result)
+            mock_logger.log_info.assert_called_with("Skipping DPU0: 10.0.0.1:8080 unreachable (offline/down)")