diff --git a/amqtt_client/README.md b/amqtt_client/README.md
index f4e7450..2421455 100644
--- a/amqtt_client/README.md
+++ b/amqtt_client/README.md
@@ -196,7 +196,7 @@ Erlang:
 {mqtt, Client, disconnected, #{}}
 
 %% Transport error, protocol error, or other fatal condition. The
-%% client process stops immediately afterwards.
+%% client process stops `normal' immediately afterwards.
 {mqtt, Client, error, #{reason := term()}}
 ```
 
@@ -223,7 +223,7 @@ Elixir:
 {:mqtt, client, :disconnected, %{}}
 
 # Transport error, protocol error, or other fatal condition. The
-# client process stops immediately afterwards.
+# client process stops `:normal` immediately afterwards.
 {:mqtt, client, :error, %{reason: term()}}
 ```
 
diff --git a/amqtt_client/src/amqtt_client.erl b/amqtt_client/src/amqtt_client.erl
index 8125499..9709666 100644
--- a/amqtt_client/src/amqtt_client.erl
+++ b/amqtt_client/src/amqtt_client.erl
@@ -62,7 +62,9 @@
 %%       connection.</li>
 %%   <li>`{mqtt, Pid, error, #{reason := term()}}' on transport error,
 %%       protocol error, or other fatal condition. The gen_server stops
-%%       immediately afterwards.</li>
+%%       with reason `normal' immediately afterwards: the owner observes
+%%       the failure via this message, so a linked non-trap_exit owner
+%%       survives.</li>
 %% </ul>
 %%
 %% == Manual ACK ==
@@ -77,6 +79,17 @@
 %% delivery and persistence will trigger a broker retransmit on
 %% reconnect.
 %%
+%% == Keep-alive ==
+%%
+%% When `keep_alive_seconds' is non-zero a PINGREQ is sent after
+%% `keep_alive_seconds * 0.75' seconds of outbound silence and arms a
+%% PINGRESP watchdog that fires after `keep_alive_seconds' seconds.
+%% Any complete inbound MQTT packet, not only PINGRESP, clears the
+%% watchdog: inbound PUBLISH, SUBACK, or PUBACK already prove the
+%% broker is responding. The trade-off is that a broker dribbling
+%% non-PINGRESP traffic can mask a dropped PINGRESP indefinitely.
+%% Setting `keep_alive_seconds = 0' disables keep-alive entirely.
+%%
 %% @end
 -module(amqtt_client).
 -behaviour(gen_server).
@@ -166,8 +179,9 @@
     owner :: pid() | undefined,
     owner_monitor :: reference() | undefined,
     keep_alive_seconds :: 0..65535 | undefined,
-    ping_timer :: reference() | undefined,
+    keepalive_timer :: reference() | undefined,
     pingresp_timer :: reference() | undefined,
+    last_outbound_ms :: integer() | undefined,
     connected = false :: boolean()
 }).
 
@@ -337,7 +351,11 @@ disconnect(Pid) ->
 %%
 %% The keep-alive timer already pings automatically; this is for
 %% applications that want to probe the link explicitly. Asynchronous:
-%% the PINGRESP is consumed silently by the gen_server.
+%% the PINGRESP is consumed silently by the gen_server. If a PINGREQ
+%% is already in flight the call is a no-op (at most one PINGREQ
+%% outstanding). With `keep_alive_seconds = 0' no PINGRESP watchdog is
+%% armed, so the call is best-effort: a missing PINGRESP is not
+%% reported as a timeout.
 %%
 %% @param Pid client returned by {@link connect/1}.
 %% @end
@@ -488,6 +506,11 @@ handle_cast({publish, Topic, Message, 0, Opts}, State) ->
         {ok, State1} -> {noreply, State1};
         {error, Reason, State1} -> stop_with_error(Reason, State1)
     end;
+handle_cast(ping, #state{pingresp_timer = Ref} = State) when Ref =/= undefined ->
+    %% PINGREQ already in flight: keep the existing watchdog deadline,
+    %% do not emit redundant bytes that arm_pingresp_timer/1 would refuse
+    %% to renew anyway.
+    {noreply, State};
 handle_cast(ping, State) ->
     case send_packet(amqtt_proto:encode_pingreq(), State) of
         {ok, State1} -> {noreply, arm_pingresp_timer(State1)};
@@ -533,13 +556,14 @@ handle_info(
     } = State
 ) ->
     stop_with_error(Reason, State);
-handle_info(send_ping, State) ->
-    case send_packet(amqtt_proto:encode_pingreq(), State#state{ping_timer = undefined}) of
-        {ok, State1} -> {noreply, arm_pingresp_timer(State1)};
-        {error, Reason, State1} -> stop_with_error(Reason, State1)
-    end;
-handle_info(pingresp_timeout, State) ->
+handle_info({timeout, Ref, keepalive_check}, #state{keepalive_timer = Ref} = State) ->
+    handle_keepalive_check(State#state{keepalive_timer = undefined});
+handle_info({timeout, _Stale, keepalive_check}, State) ->
+    {noreply, State};
+handle_info({timeout, Ref, pingresp_timeout}, #state{pingresp_timer = Ref} = State) ->
     stop_with_error(pingresp_timeout, State#state{pingresp_timer = undefined});
+handle_info({timeout, _Stale, pingresp_timeout}, State) ->
+    {noreply, State};
 handle_info({request_timeout, PacketId}, State) ->
     case maps:get(PacketId, State#state.pending, undefined) of
         {_AckType, From, _TRef} ->
@@ -640,8 +664,10 @@ raw_send(Packet, #state{transport = T, connection = C}) ->
 
 send_packet(Packet, State) ->
     case raw_send(Packet, State) of
-        ok -> {ok, restart_ping_timer(State)};
-        {error, Reason} -> {error, Reason, State}
+        ok ->
+            {ok, State#state{last_outbound_ms = erlang:monotonic_time(millisecond)}};
+        {error, Reason} ->
+            {error, Reason, State}
     end.
 
 send_and_pend(Packet, PacketId, AckType, From, State) ->
@@ -659,8 +685,7 @@ stop_with_error(Reason, #state{owner = Owner} = State) ->
     Owner ! {mqtt, self(), error, #{reason => Reason}},
     State1 = reply_pending_with_error(Reason, State),
     close_connection(State1),
-    {stop, {transport_error, Reason},
-        cancel_all_timers(State1#state{connection = undefined, connected = false})}.
+    {stop, normal, cancel_all_timers(State1#state{connection = undefined, connected = false})}.
 
 reply_pending_with_error(Reason, #state{pending = Pending} = State) ->
     maps:foreach(
@@ -697,9 +722,10 @@ alloc_packet_id(#state{next_packet_id = Id, pending = Pending} = State, N) ->
 process_buffer(#state{buffer = Buffer} = State) ->
     case amqtt_proto:decode(Buffer) of
         {ok, {Type, Data}, Rest} ->
-            case handle_packet(Type, Data, State#state{buffer = Rest}) of
-                {ok, State1} -> process_buffer(State1);
-                {stop_normal, State1} -> {stop_normal, State1};
+            State1 = observe_inbound(State#state{buffer = Rest}),
+            case handle_packet(Type, Data, State1) of
+                {ok, State2} -> process_buffer(State2);
+                {stop_normal, State2} -> {stop_normal, State2};
                 {error, _, _} = Err -> Err
             end;
         {error, incomplete} ->
@@ -710,7 +736,7 @@ process_buffer(#state{buffer = Buffer} = State) ->
 
 handle_packet(connack, #{return_code := 0} = Data, #state{owner = Owner} = State) ->
     Owner ! {mqtt, self(), connack, Data},
-    {ok, State#state{connected = true}};
+    {ok, schedule_keepalive_timer(State#state{connected = true})};
 handle_packet(connack, Data, #state{owner = Owner} = State) ->
     %% MQTT 3.1.1 §3.2: a non-zero return code requires the broker to close.
     Owner ! {mqtt, self(), connack, Data},
@@ -827,7 +853,8 @@ handle_packet(suback, #{packet_id := PacketId, return_codes := RCs}, State) ->
 handle_packet(unsuback, #{packet_id := PacketId}, State) ->
     {ok, complete_pending(PacketId, unsuback, ok, State)};
 handle_packet(pingresp, _Data, State) ->
-    {ok, cancel_pingresp_timer(State)};
+    %% pingresp_timer already cleared by observe_inbound.
+    {ok, State};
 handle_packet(_Type, _Data, State) ->
     {ok, State}.
 
@@ -845,35 +872,83 @@ complete_pending(PacketId, ExpectedAckType, Reply, #state{pending = Pending} = S
 %% Internal: Timers
 %% -------------------------------------------------------------------
 
-restart_ping_timer(#state{keep_alive_seconds = 0} = State) ->
+cancel_timer_safe(undefined) ->
+    ok;
+cancel_timer_safe(Ref) ->
+    erlang:cancel_timer(Ref),
+    ok.
+
+schedule_keepalive_timer(#state{last_outbound_ms = undefined} = State) ->
     State;
-restart_ping_timer(State) ->
-    State1 = cancel_ping_timer(State),
-    Interval = State1#state.keep_alive_seconds * 750,
-    Ref = erlang:send_after(Interval, self(), send_ping),
-    State1#state{ping_timer = Ref}.
+schedule_keepalive_timer(#state{keep_alive_seconds = 0} = State) ->
+    State;
+schedule_keepalive_timer(
+    #state{
+        keep_alive_seconds = KA,
+        last_outbound_ms = Last,
+        keepalive_timer = OldRef
+    } = State
+) ->
+    cancel_timer_safe(OldRef),
+    Now = erlang:monotonic_time(millisecond),
+    DueIn = max(0, Last + KA * 750 - Now),
+    Ref = erlang:start_timer(DueIn, self(), keepalive_check),
+    State#state{keepalive_timer = Ref}.
+
+handle_keepalive_check(#state{pingresp_timer = Ref} = State) when Ref =/= undefined ->
+    %% PINGREQ in flight: pingresp_timer adjudicates, do not send another.
+    {noreply, State};
+handle_keepalive_check(#state{keep_alive_seconds = 0} = State) ->
+    {noreply, State};
+handle_keepalive_check(
+    #state{keep_alive_seconds = KA, last_outbound_ms = Last} = State
+) ->
+    Idle = KA * 750,
+    Now = erlang:monotonic_time(millisecond),
+    Elapsed = Now - Last,
+    if
+        Elapsed >= Idle ->
+            send_keepalive_pingreq(State);
+        true ->
+            Ref = erlang:start_timer(Idle - Elapsed, self(), keepalive_check),
+            {noreply, State#state{keepalive_timer = Ref}}
+    end.
 
-cancel_ping_timer(#state{ping_timer = undefined} = State) ->
+send_keepalive_pingreq(State) ->
+    case send_packet(amqtt_proto:encode_pingreq(), State) of
+        {ok, State1} ->
+            {noreply, arm_pingresp_timer(State1)};
+        {error, Reason, State1} ->
+            stop_with_error(Reason, State1)
+    end.
+
+%% Any complete inbound MQTT packet clears the watchdog: PINGRESP has
+%% no privileged status. Called per decoded packet, not per TCP chunk.
+observe_inbound(#state{pingresp_timer = undefined} = State) ->
     State;
-cancel_ping_timer(#state{ping_timer = Ref} = State) ->
-    erlang:cancel_timer(Ref),
-    State#state{ping_timer = undefined}.
+observe_inbound(State) ->
+    schedule_keepalive_timer(cancel_pingresp_timer(State)).
 
 arm_pingresp_timer(#state{keep_alive_seconds = 0} = State) ->
     State;
-arm_pingresp_timer(#state{pingresp_timer = OldRef, keep_alive_seconds = KA} = State) ->
-    case OldRef of
-        undefined -> ok;
-        _ -> erlang:cancel_timer(OldRef)
-    end,
-    Ref = erlang:send_after(KA * 1000, self(), pingresp_timeout),
+arm_pingresp_timer(#state{pingresp_timer = Ref} = State) when Ref =/= undefined ->
+    %% Do not renew: a dropped PINGRESP must not be hidden by the next PINGREQ.
+    State;
+arm_pingresp_timer(#state{keep_alive_seconds = KA} = State) ->
+    Ref = erlang:start_timer(KA * 1000, self(), pingresp_timeout),
     State#state{pingresp_timer = Ref}.
 
 cancel_pingresp_timer(#state{pingresp_timer = undefined} = State) ->
     State;
 cancel_pingresp_timer(#state{pingresp_timer = Ref} = State) ->
-    erlang:cancel_timer(Ref),
+    cancel_timer_safe(Ref),
     State#state{pingresp_timer = undefined}.
 
+cancel_keepalive_timer(#state{keepalive_timer = undefined} = State) ->
+    State;
+cancel_keepalive_timer(#state{keepalive_timer = Ref} = State) ->
+    cancel_timer_safe(Ref),
+    State#state{keepalive_timer = undefined}.
+
 cancel_all_timers(State) ->
-    cancel_pingresp_timer(cancel_ping_timer(State)).
+    cancel_keepalive_timer(cancel_pingresp_timer(State)).
diff --git a/amqtt_client/test/amqtt_client_tests.erl b/amqtt_client/test/amqtt_client_tests.erl
index 0a031dc..cfca5e3 100644
--- a/amqtt_client/test/amqtt_client_tests.erl
+++ b/amqtt_client/test/amqtt_client_tests.erl
@@ -308,6 +308,216 @@ manual_ping_test_() ->
         cleanup(Listener, Client)
     end}.
 
+manual_ping_suppressed_while_watchdog_armed_test_() ->
+    {timeout, ?TEST_TIMEOUT, fun() ->
+        {Listener, Port} = start_mock_broker(),
+        Client = connect_and_handshake(Port, <<"c">>, 1),
+        BrokerSocket = accept(Listener),
+        _ConnectData = recv_packet(BrokerSocket),
+        send_connack(BrokerSocket),
+        ?ASSERT_RECEIVE({mqtt, Client, connack, _}, ?RECV_TIMEOUT),
+
+        %% Auto PINGREQ at ~750 ms arms the pingresp_timer watchdog.
+        PingData = recv_packet(BrokerSocket),
+        ?assertMatch({ok, {pingreq, _}, <<>>}, amqtt_proto:decode(PingData)),
+
+        %% A manual ping/1 while the watchdog is armed must not put more
+        %% bytes on the wire: arm_pingresp_timer/1 refuses to renew, so
+        %% any extra PINGREQ would be redundant.
+        amqtt_client:ping(Client),
+        ?assertEqual({error, timeout}, gen_tcp:recv(BrokerSocket, 0, 300)),
+
+        %% Clear the watchdog so cleanup runs against a healthy client.
+        send_broker(BrokerSocket, <<16#D0, 0>>),
+
+        cleanup(Listener, Client)
+    end}.
+
+no_false_timeout_under_publish_cadence_test_() ->
+    {timeout, ?TEST_TIMEOUT, fun() ->
+        {Listener, Port} = start_mock_broker(),
+        process_flag(trap_exit, true),
+        Client = connect_and_handshake(Port, <<"c">>, 1),
+        BrokerSocket = accept(Listener),
+        _ConnectData = recv_packet(BrokerSocket),
+        send_connack(BrokerSocket),
+        ?ASSERT_RECEIVE({mqtt, Client, connack, _}, ?RECV_TIMEOUT),
+
+        PingData = recv_packet(BrokerSocket),
+        ?assertMatch({ok, {pingreq, _}, <<>>}, amqtt_proto:decode(PingData)),
+        timer:sleep(300),
+
+        PubAsync = async_call(fun() ->
+            amqtt_client:publish(Client, <<"t">>, <<"m">>, 1)
+        end),
+        PubData = recv_packet(BrokerSocket),
+        {ok, {publish, PubInfo}, <<>>} = amqtt_proto:decode(PubData),
+        PacketId = maps:get(packet_id, PubInfo),
+        send_broker(BrokerSocket, amqtt_proto:encode_puback(PacketId)),
+        ?assertEqual({ok, PacketId}, await(PubAsync, 2000)),
+
+        receive
+            {mqtt, Client, error, _} = E ->
+                erlang:error({unexpected_error, E});
+            {'EXIT', Client, _} = X ->
+                erlang:error({unexpected_exit, X})
+        after 1200 ->
+            ok
+        end,
+        ?assert(is_process_alive(Client)),
+        process_flag(trap_exit, false),
+
+        cleanup(Listener, Client)
+    end}.
+
+pingresp_timeout_fires_when_broker_silent_test_() ->
+    {timeout, ?TEST_TIMEOUT, fun() ->
+        {Listener, Port} = start_mock_broker(),
+        process_flag(trap_exit, true),
+        Client = connect_and_handshake(Port, <<"c">>, 1),
+        BrokerSocket = accept(Listener),
+        _ConnectData = recv_packet(BrokerSocket),
+        send_connack(BrokerSocket),
+        ?ASSERT_RECEIVE({mqtt, Client, connack, _}, ?RECV_TIMEOUT),
+
+        PingData = recv_packet(BrokerSocket),
+        ?assertMatch({ok, {pingreq, _}, <<>>}, amqtt_proto:decode(PingData)),
+
+        ?ASSERT_RECEIVE({mqtt, Client, error, #{reason := pingresp_timeout}}, 1500),
+        ?ASSERT_RECEIVE({'EXIT', Client, normal}, 1500),
+        process_flag(trap_exit, false),
+
+        try
+            gen_tcp:close(BrokerSocket)
+        catch
+            _:_ -> ok
+        end,
+        try
+            gen_tcp:close(Listener)
+        catch
+            _:_ -> ok
+        end,
+        ok
+    end}.
+
+inbound_publish_cancels_pingresp_watchdog_test_() ->
+    {timeout, ?TEST_TIMEOUT, fun() ->
+        {Listener, Port} = start_mock_broker(),
+        process_flag(trap_exit, true),
+        Client = connect_and_handshake(Port, <<"c">>, 1),
+        BrokerSocket = accept(Listener),
+        _ConnectData = recv_packet(BrokerSocket),
+        send_connack(BrokerSocket),
+        ?ASSERT_RECEIVE({mqtt, Client, connack, _}, ?RECV_TIMEOUT),
+
+        PingData = recv_packet(BrokerSocket),
+        ?assertMatch({ok, {pingreq, _}, <<>>}, amqtt_proto:decode(PingData)),
+
+        InboundPublish = iolist_to_binary(
+            amqtt_proto:encode_publish(#{
+                topic => <<"t">>,
+                message => <<"hi">>,
+                qos => 0
+            })
+        ),
+        send_broker(BrokerSocket, InboundPublish),
+        ?ASSERT_RECEIVE({mqtt, Client, publish, #{topic := <<"t">>}}, ?RECV_TIMEOUT),
+
+        receive
+            {mqtt, Client, error, _} = E ->
+                erlang:error({unexpected_error, E});
+            {'EXIT', Client, _} = X ->
+                erlang:error({unexpected_exit, X})
+        after 1200 ->
+            ok
+        end,
+        ?assert(is_process_alive(Client)),
+        process_flag(trap_exit, false),
+
+        cleanup(Listener, Client)
+    end}.
+
+owner_without_trap_exit_survives_transport_error_test_() ->
+    {timeout, ?TEST_TIMEOUT, fun() ->
+        {Listener, Port} = start_mock_broker(),
+        Tester = self(),
+        Owner = spawn(fun() ->
+            case
+                amqtt_client:connect(#{
+                    host => "127.0.0.1",
+                    port => Port,
+                    client_id => <<"o">>,
+                    keep_alive_seconds => 0,
+                    owner => self()
+                })
+            of
+                {ok, C} ->
+                    Tester ! {client, C},
+                    owner_forwarding_loop(Tester);
+                Err ->
+                    Tester ! {connect_failed, Err}
+            end
+        end),
+        OwnerRef = erlang:monitor(process, Owner),
+        BrokerSocket = accept(Listener),
+        _ConnectData = recv_packet(BrokerSocket),
+        send_connack(BrokerSocket),
+
+        Client =
+            receive
+                {client, C} -> C;
+                {connect_failed, E} -> erlang:error({connect_failed, E})
+            after 3000 ->
+                erlang:error(no_client_msg)
+            end,
+
+        %% PUBACK with packet_id 0: rejected by the hardened decoder. The
+        %% explicit assertion on the forwarded error reason keeps this
+        %% test from silently turning into a no-op if decoder validation
+        %% ever loosens.
+        send_broker(BrokerSocket, <<16#40, 2, 0, 0>>),
+
+        receive
+            {forwarded, {mqtt, Client, error, #{reason := Reason}}} ->
+                ?assertMatch({protocol_error, _}, Reason)
+        after 1000 ->
+            erlang:error(no_error_event)
+        end,
+
+        timer:sleep(50),
+
+        ?assert(is_process_alive(Owner)),
+        ?assertNot(is_process_alive(Client)),
+
+        Owner ! stop,
+        receive
+            {'DOWN', OwnerRef, process, Owner, _} -> ok
+        after 1000 ->
+            erlang:demonitor(OwnerRef, [flush])
+        end,
+
+        try
+            gen_tcp:close(BrokerSocket)
+        catch
+            _:_ -> ok
+        end,
+        try
+            gen_tcp:close(Listener)
+        catch
+            _:_ -> ok
+        end,
+        ok
+    end}.
+
+owner_forwarding_loop(Tester) ->
+    receive
+        stop ->
+            ok;
+        Msg ->
+            Tester ! {forwarded, Msg},
+            owner_forwarding_loop(Tester)
+    end.
+
 %% -------------------------------------------------------------------
 %% QoS 1
 %% -------------------------------------------------------------------
@@ -829,7 +1039,7 @@ buffer_overflow_stops_with_error_test_() ->
         send_broker(BrokerSocket, <<Header/binary, Garbage/binary>>),
 
         ?ASSERT_RECEIVE({mqtt, Client, error, #{reason := buffer_overflow}}, 3000),
-        ?ASSERT_RECEIVE({'EXIT', Client, {transport_error, buffer_overflow}}, 3000),
+        ?ASSERT_RECEIVE({'EXIT', Client, normal}, 3000),
         process_flag(trap_exit, false),
 
         try
@@ -859,7 +1069,7 @@ protocol_error_stops_connection_test_() ->
         send_broker(BrokerSocket, <<16#40, 2, 0, 0>>),
 
         ?ASSERT_RECEIVE({mqtt, Client, error, #{reason := {protocol_error, _}}}, ?RECV_TIMEOUT),
-        ?ASSERT_RECEIVE({'EXIT', Client, {transport_error, {protocol_error, _}}}, ?RECV_TIMEOUT),
+        ?ASSERT_RECEIVE({'EXIT', Client, normal}, ?RECV_TIMEOUT),
         process_flag(trap_exit, false),
 
         try