diff --git a/CHANGELOG.md b/CHANGELOG.md
index ee2b2451b6e7877efd7590ac89e827a95d5ac953..a945fc68c0d3749974bd573075678b15b2a742ee 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -10,6 +10,16 @@ breaks something, or changes the observable behavior of existing surfaces
 broadly; **patch** for fixes, and for additive opt-in capability — a new key,
 flag, or page that no existing user can encounter without opting into it.
 
+## [Unreleased]
+
+### Fixed
+
+<!-- [doc->REQ-IO-MSG-EDGES] -->
+- Messages delivered by `spt api listen` now reach the receiver's incoming-message
+  history and last-message state, for both queued backlog and live TCP delivery.
+  A later hook poll does not record those deliveries again; filtered notifications
+  are not recorded as delivered messages.
+
 ## [0.67.1] - 2026-09-06
 
 Documentation-only release. The manifest reference in the developer docs now
diff --git a/crates/spt/src/api/startup.rs b/crates/spt/src/api/startup.rs
index 5b98ac68e2dae523b58e4f25d5fd916bfc2910ac..3d77d9faf7d9607941fe085585ac1b6745f29f24 100644
--- a/crates/spt/src/api/startup.rs
+++ b/crates/spt/src/api/startup.rs
@@ -1058,6 +1058,10 @@ pub fn cmd_listen(
         emit(id, from, body);
         let event = spt_msg::emit::render_event_whole_for(id, from, body);
         spt_live::record_context_injection(id, spt_live::KIND_OWL_MESSAGE, &event);
+        // Both backlog (after notif validity) and live TCP reach this edge.
+        // Observe only delivered content, through the same bus as hook polling.
+        // [impl->REQ-IO-MSG-EDGES]
+        crate::cli::publish_msg_io(id, spt_proto::ioevent::IO_KIND_MSG_IN, from, body);
     };
 
     if once {
diff --git a/crates/spt/tests/io_events_undriven_kinds_e2e.rs b/crates/spt/tests/io_events_undriven_kinds_e2e.rs
index 5f23372ec23eb1824515f2f2bb70a97361fcd824..9eeec25952ada86c46efb3fdbacc1cb040729139 100644
--- a/crates/spt/tests/io_events_undriven_kinds_e2e.rs
+++ b/crates/spt/tests/io_events_undriven_kinds_e2e.rs
@@ -177,11 +177,14 @@ fn send_bounded(spt_bin: &Path, home: &Path, args: &[&str], body: &str) -> Outpu
         .expect("spt send output")
 }
 
+// The API also carries lifecycle boundaries (listen emits boot). These cells
+// assert the IO vocabulary; boundary_events_e2e covers the other event family.
 fn rows(v: &serde_json::Value) -> Vec<(String, String, Option<String>)> {
     v["events"]
         .as_array()
         .expect("events array")
         .iter()
+        .filter(|e| e["kind"].as_str().is_some_and(spt_proto::ioevent::is_io_kind))
         .map(|e| {
             (
                 e["kind"].as_str().unwrap_or_default().to_string(),
@@ -266,9 +269,40 @@ impl Rig {
         );
     }
 
+    fn listen_once(&self) -> Command {
+        // Keep bob on its DISTINCT live owner, rather than aliasing alice's pid.
+        let anchor = reap::breadcrumb_daemon_pid(self.home.path())
+            .expect("receiver's live owner")
+            .to_string();
+        let mut cmd = Command::new(&self.spt_bin);
+        cmd.no_window()
+            .args([
+                "api",
+                "--adapter",
+                "mock",
+                "listen",
+                PEER,
+                "--parent-pid",
+                &anchor,
+                "--session-id",
+                SID_B,
+                "--once",
+            ])
+            .env("SPT_HOME", self.home.path());
+        cmd
+    }
+
+    fn hook_poll_receiver(&self) -> Output {
+        let mut cmd = Command::new(&self.spt_bin);
+        cmd.no_window()
+            .args(["api", "poll", PEER, "--session-id", SID_B])
+            .env("SPT_HOME", self.home.path());
+        common::output_bounded(cmd, Duration::from_secs(60))
+    }
+
     fn reap(&self, label: &str) {
-        let observed = reap::breadcrumb_daemon_pid(self.home.path())
-            .and_then(|pid| reap::observe(label, pid));
+        let observed =
+            reap::breadcrumb_daemon_pid(self.home.path()).and_then(|pid| reap::observe(label, pid));
         let mut stop = Command::new(&self.spt_bin);
         stop.no_window()
             .args(["daemon", "stop"])
@@ -309,7 +343,11 @@ fn a_consumed_commune_reaches_the_log_and_polls_back() {
             .expect("live host");
         std::fs::write(r.drop_path(), COMMUNE_BODY).unwrap();
         let report = host.pulse_tick(Some(SID_A)).expect("tick");
-        assert_eq!(report.ingested.len(), 1, "PRECONDITION: the drop was ingested");
+        assert_eq!(
+            report.ingested.len(),
+            1,
+            "PRECONDITION: the drop was ingested"
+        );
         assert!(
             !r.drop_path().exists(),
             "PRECONDITION: a consumed drop is unlinked, which is why the event must carry its bytes"
@@ -426,3 +464,186 @@ fn the_delivery_edge_polls_back_as_msg_in_on_the_receiver() {
         );
     });
 }
+
+// [int->REQ-IO-MSG-EDGES]
+// [int->REQ-NOW-SIGNAL-CATEGORIES-V1]
+// releases#280: the actual listen callback, not a synthetic bus publication.
+#[test]
+fn relay_backlog_and_live_deliveries_stamp_once_even_after_hook_poll() {
+    with_rig("io_events_undriven/relay_msg_in", |r| {
+        const BACKLOG: &str = "the offline backlog message";
+        const LIVE: &str = "the live TCP message";
+        const HOOK: &str = "the later hook-only message";
+        r.seed(PEER, SID_B);
+        let receiver = perch::resolve_perch_path(PEER, ParentHint::Infer);
+        let owlery = perch::owlery_dir();
+        assert!(spt_store::registry::resolve_address(PEER, &owlery).is_none());
+        let queued = send_bounded(
+            &r.spt_bin,
+            r.home.path(),
+            &["send", PEER, "--from", AUTHOR],
+            BACKLOG,
+        );
+        assert!(
+            queued.status.success(),
+            "PRECONDITION: backlog send committed"
+        );
+        let pending = spt_store::spool::peek_all_at(&receiver).unwrap();
+        assert_eq!(pending.len(), 1, "PRECONDITION: one actual backlog row");
+        assert_eq!(pending[0].2, BACKLOG);
+        let backlog_id = pending[0].0;
+
+        // The real listener registers its bound TCP address before the first
+        // 200ms pump. Send directly through the public TCP leg as soon as that
+        // address exists: launching another CLI after READY could miss the pump.
+        // This leg has NO spool fallback, so it cannot quietly retest backlog.
+        let sender = std::thread::spawn(move || {
+            let deadline = std::time::Instant::now() + Duration::from_secs(60);
+            while spt_store::registry::resolve_address(PEER, &owlery).is_none() {
+                assert!(
+                    std::time::Instant::now() < deadline,
+                    "listener never registered"
+                );
+                std::thread::sleep(Duration::from_millis(2));
+            }
+            spt_msg::deliver::deliver_tcp(PEER, AUTHOR, LIVE, &owlery)
+        });
+        let listen = common::output_bounded(r.listen_once(), Duration::from_secs(60));
+        assert!(
+            sender.join().expect("live sender thread"),
+            "PRECONDITION: live TCP write succeeded"
+        );
+        assert!(
+            listen.status.success(),
+            "listen failed: {}",
+            String::from_utf8_lossy(&listen.stderr)
+        );
+        let presented = String::from_utf8_lossy(&listen.stdout);
+        assert!(
+            presented.contains(BACKLOG),
+            "backlog reached the receiver: {presented}"
+        );
+        assert!(
+            presented.contains(LIVE),
+            "live message reached the receiver: {presented}"
+        );
+        let expected = vec![
+            (
+                "MSG_IN".to_string(),
+                BACKLOG.to_string(),
+                Some(AUTHOR.to_string()),
+            ),
+            (
+                "MSG_IN".to_string(),
+                LIVE.to_string(),
+                Some(AUTHOR.to_string()),
+            ),
+        ];
+        assert_eq!(
+            rows(&poll(
+                &r.spt_bin,
+                r.home.path(),
+                PEER,
+                SID_B,
+                &["--after", "0"]
+            )),
+            expected,
+            "both relay arms publish exactly once, into the receiver's log"
+        );
+        let last = spt_store::lastmsg::read_at(&receiver, spt_store::lastmsg::Direction::In)
+            .expect("the relay updates LAST_MSGS through the default bus");
+        assert_eq!(last.peer.as_deref(), Some(AUTHOR));
+        assert_eq!(last.excerpt, LIVE);
+
+        // Exercise the SAME durable spool through the later hook drain. A new
+        // hook-only row is the positive control that this drain really delivers
+        // and publishes, while the relay-consumed rows must not stamp again.
+        let retained = spt_store::spool::audit_rows_at(&receiver).unwrap();
+        assert!(
+            retained.iter().any(|row| row.id == backlog_id && row.delivered),
+            "PRECONDITION: the consumed backlog row still exists for the hook's later scan"
+        );
+        spt_store::spool::spool_message_at(&receiver, AUTHOR, HOOK).unwrap();
+        let hook = r.hook_poll_receiver();
+        assert!(
+            hook.status.success(),
+            "hook poll failed: {}",
+            String::from_utf8_lossy(&hook.stderr)
+        );
+        let hook_body = String::from_utf8_lossy(&hook.stdout);
+        assert!(
+            hook_body.contains(HOOK),
+            "hook drain delivered its fresh row: {hook_body}"
+        );
+        assert!(
+            !hook_body.contains(BACKLOG) && !hook_body.contains(LIVE),
+            "relay rows replayed: {hook_body}"
+        );
+        let mut expected_after_hook = expected;
+        expected_after_hook.push((
+            "MSG_IN".to_string(),
+            HOOK.to_string(),
+            Some(AUTHOR.to_string()),
+        ));
+        assert_eq!(
+            rows(&poll(
+                &r.spt_bin,
+                r.home.path(),
+                PEER,
+                SID_B,
+                &["--after", "0"]
+            )),
+            expected_after_hook,
+            "a later hook poll must not double-stamp either relay-delivered message"
+        );
+    });
+}
+
+// [int->REQ-IO-MSG-EDGES]
+// [int->REQ-NOTIF-DRAIN-ROW-VALIDITY]
+#[test]
+fn a_notification_filtered_by_the_relay_publishes_no_msg_in() {
+    with_rig("io_events_undriven/relay_filtered_notify", |r| {
+        r.seed(PEER, SID_B);
+        let receiver = perch::resolve_perch_path(PEER, ParentHint::Infer);
+        let body = spt_proto::event::compose_typed_event(
+            spt_proto::event::EVENT_TYPE_NOTIFY,
+            &[
+                ("from", AUTHOR),
+                ("notif_id", "gone:relay-280"),
+                ("subnet", "home"),
+            ],
+            "an obsolete notification that must never arrive",
+        );
+        // No row with this notif_id exists in the isolated home. A non-deferred
+        // copy reaches the relay drain but fails its source-row validity gate.
+        spt_store::spool::spool_message_at(&receiver, AUTHOR, &body).unwrap();
+        let listen = common::output_bounded(r.listen_once(), Duration::from_secs(60));
+        assert!(
+            listen.status.success(),
+            "listen failed: {}",
+            String::from_utf8_lossy(&listen.stderr)
+        );
+        assert!(
+            !String::from_utf8_lossy(&listen.stdout).contains("an obsolete notification"),
+            "PRECONDITION: the relay did not present the filtered notification"
+        );
+        let hook = r.hook_poll_receiver();
+        assert!(hook.status.success(), "subsequent hook poll failed");
+        assert!(
+            rows(&poll(
+                &r.spt_bin,
+                r.home.path(),
+                PEER,
+                SID_B,
+                &["--after", "0"]
+            ))
+            .is_empty(),
+            "filtered content must not publish through either delivery edge"
+        );
+        assert!(
+            spt_store::lastmsg::read_at(&receiver, spt_store::lastmsg::Direction::In).is_none(),
+            "a filtered notification must not become the receiver's last message"
+        );
+    });
+}