diff --git a/crates/khive-pack-comm/src/handlers.rs b/crates/khive-pack-comm/src/handlers.rs
index 335374aa..254935c0 100644
--- a/crates/khive-pack-comm/src/handlers.rs
+++ b/crates/khive-pack-comm/src/handlers.rs
@@ -115,6 +115,7 @@ pub(crate) async fn handle_send(
         Some(&to_actor),
         None,
         None,
+        p.tags.as_deref(),
     )
     .await?;
 
@@ -146,6 +147,13 @@ pub(crate) async fn handle_inbox(
     }
     let limit = raw_limit.clamp(1, 200) as usize;
 
+    // #493: from_actor / from_prefix sender filter — mutually exclusive.
+    if p.from_actor.is_some() && p.from_prefix.is_some() {
+        return Err(RuntimeError::InvalidInput(
+            "inbox: `from_actor` and `from_prefix` are mutually exclusive".into(),
+        ));
+    }
+
     let status = match p.status.as_deref().unwrap_or("unread") {
         s @ ("unread" | "read" | "all") => s,
         other => {
@@ -201,18 +209,66 @@ pub(crate) async fn handle_inbox(
         order_by: None, // preserves existing created_at DESC ordering
         ..Default::default()
     };
-    let page = runtime
-        .notes(token)?
-        .query_notes_filtered(
-            token.namespace().as_str(),
-            &filter,
-            PageRequest {
-                limit: limit as u32,
-                offset: 0,
-            },
-        )
-        .await?;
-    let messages: Vec<Value> = page.items.iter().map(note_to_message_json).collect();
+    let store = runtime.notes(token)?;
+
+    // #493: when a sender filter is supplied, apply it in Rust after the standard
+    // direction/status/to_actor filters (which stay pushed into SQL for index usage) —
+    // `FilterOp` has no prefix-match operator, so from_prefix cannot be pushed down.
+    // Pages beyond the first are scanned (same unbounded-page-loop shape `handle_thread`
+    // uses) until `limit` matches are collected or the store is exhausted.
+    let messages: Vec<Value> = if p.from_actor.is_some() || p.from_prefix.is_some() {
+        const PAGE_SIZE: u32 = 200;
+        let mut collected: Vec<Value> = Vec::new();
+        let mut db_offset: u32 = 0;
+        loop {
+            let page = store
+                .query_notes_filtered(
+                    token.namespace().as_str(),
+                    &filter,
+                    PageRequest {
+                        limit: PAGE_SIZE,
+                        offset: db_offset.into(),
+                    },
+                )
+                .await?;
+            let fetched = page.items.len() as u32;
+            for n in &page.items {
+                let sender = n
+                    .properties
+                    .as_ref()
+                    .and_then(|props| props.get("from_actor"))
+                    .and_then(Value::as_str);
+                let matches = match (p.from_actor.as_deref(), p.from_prefix.as_deref()) {
+                    (Some(exact), None) => sender == Some(exact),
+                    (None, Some(prefix)) => sender.map(|s| s.starts_with(prefix)).unwrap_or(false),
+                    _ => unreachable!("mutual exclusion already validated above"),
+                };
+                if matches {
+                    collected.push(note_to_message_json(n));
+                    if collected.len() >= limit {
+                        break;
+                    }
+                }
+            }
+            if collected.len() >= limit || fetched < PAGE_SIZE {
+                break;
+            }
+            db_offset += PAGE_SIZE;
+        }
+        collected
+    } else {
+        let page = store
+            .query_notes_filtered(
+                token.namespace().as_str(),
+                &filter,
+                PageRequest {
+                    limit: limit as u32,
+                    offset: 0,
+                },
+            )
+            .await?;
+        page.items.iter().map(note_to_message_json).collect()
+    };
     let count = messages.len();
     Ok(json!({ "messages": messages, "count": count }))
 }
@@ -419,6 +475,7 @@ pub(crate) async fn handle_reply(
         Some(&reply_to_actor),
         in_reply_to_message_id.as_deref(),
         references_chain.as_deref(),
+        p.tags.as_deref(),
     )
     .await?;
 
@@ -458,6 +515,16 @@ pub(crate) async fn handle_thread(
     let p: ThreadParams = deser(params)?;
     let limit = p.limit.unwrap_or(100).clamp(1, 500) as usize;
 
+    // #494: order — "asc" (default, unchanged) | "desc". Closed set.
+    let order = match p.order.as_deref().unwrap_or("asc") {
+        o @ ("asc" | "desc") => o,
+        other => {
+            return Err(RuntimeError::InvalidInput(format!(
+                "thread: invalid order {other:?}; expected one of: asc, desc"
+            )));
+        }
+    };
+
     // Resolve and validate the passed ID.
     let passed_uuid = resolve_id(runtime, token, &p.id, "thread").await?;
 
@@ -524,7 +591,7 @@ pub(crate) async fn handle_thread(
         ..Default::default()
     };
     const PAGE_SIZE: u32 = 200;
-    let mut messages: Vec<Value> = Vec::new();
+    let mut rows: Vec<ThreadRow> = Vec::new();
     let mut db_offset: u32 = 0;
 
     loop {
@@ -540,7 +607,11 @@ pub(crate) async fn handle_thread(
             .await?;
         let fetched = page.items.len() as u32;
         for n in &page.items {
-            messages.push(note_to_message_json(n));
+            rows.push(ThreadRow {
+                created_at: n.created_at,
+                full_id: n.id,
+                json: note_to_message_json(n),
+            });
         }
         if fetched < PAGE_SIZE {
             break;
@@ -555,24 +626,102 @@ pub(crate) async fn handle_thread(
     // already-validated root note when the query didn't already return it, so
     // `comm.thread(id=root)` never reports an empty/incomplete thread for a
     // root that exists but predates the canonical `thread_id` field.
-    let root_full_id = root_note.id.as_hyphenated().to_string();
-    let root_already_present = messages
-        .iter()
-        .any(|m| m.get("full_id").and_then(Value::as_str) == Some(root_full_id.as_str()));
+    let root_already_present = rows.iter().any(|r| r.full_id == root_note.id);
     if !root_already_present {
-        messages.push(note_to_message_json(&root_note));
+        rows.push(ThreadRow {
+            created_at: root_note.created_at,
+            full_id: root_note.id,
+            json: note_to_message_json(&root_note),
+        });
     }
 
-    // Sort chronologically ascending (earliest first).
-    // ISO 8601 timestamps (e.g. "2026-05-27T10:30:00.000000Z") are lexicographically
-    // ordered, so string comparison is correct and cheaper than parsing.
-    messages.sort_by(|a, b| {
-        let a_ts = a.get("created_at").and_then(Value::as_str).unwrap_or("");
-        let b_ts = b.get("created_at").and_then(Value::as_str).unwrap_or("");
-        a_ts.cmp(b_ts)
+    // #494 / codex r1: `after` cursor — either a message id (short prefix or full
+    // UUID, resolved the same way `id` is) or an RFC 3339 timestamp. An id cursor
+    // resolves to the full `(created_at, full_id)` tuple of the referenced note so
+    // ties on equal microsecond timestamps are broken deterministically instead of
+    // being skipped or duplicated. A timestamp cursor is parsed to microseconds via
+    // chrono (matching the pattern in khive-pack-brain/src/handlers.rs and
+    // khive-vcs/src/sync.rs) rather than compared as a raw string, so non-canonical
+    // but valid RFC 3339 forms (whole-second `Z`, `+00:00` offsets, ...) compare
+    // correctly against khive's canonical microsecond timestamps. An `after` value
+    // that is neither a resolvable id nor a parseable RFC 3339 timestamp is a hard
+    // error — never silently coerced or treated as "no cursor".
+    let after_cursor: Option<AfterCursor> = match p.after.as_deref() {
+        None => None,
+        Some(raw) => {
+            let looks_like_id = raw.parse::<Uuid>().is_ok()
+                || (raw.len() >= 8 && raw.chars().all(|c| c.is_ascii_hexdigit()));
+            if looks_like_id {
+                let cursor_uuid = resolve_id(runtime, token, raw, "thread").await?;
+                let cursor_store = runtime.notes(token)?;
+                let cursor_note = cursor_store
+                    .get_note(cursor_uuid)
+                    .await
+                    .map_err(|e| RuntimeError::Internal(format!("thread: get_note (after): {e}")))?
+                    .ok_or_else(|| {
+                        RuntimeError::InvalidInput(format!(
+                            "thread: `after` cursor {raw:?} does not resolve to a message"
+                        ))
+                    })?;
+                Some(AfterCursor::Id {
+                    created_at: cursor_note.created_at,
+                    full_id: cursor_note.id,
+                })
+            } else {
+                let micros = chrono::DateTime::parse_from_rfc3339(raw.trim())
+                    .map(|dt| dt.with_timezone(&Utc).timestamp_micros())
+                    .map_err(|e| {
+                        RuntimeError::InvalidInput(format!(
+                            "thread: `after` cursor {raw:?} is neither a resolvable message id \
+                             nor a valid RFC 3339 timestamp: {e}"
+                        ))
+                    })?;
+                Some(AfterCursor::Timestamp { micros })
+            }
+        }
+    };
+    if let Some(cursor) = &after_cursor {
+        rows.retain(|r| match cursor {
+            // Tuple compare, not timestamp-only: two rows sharing a microsecond
+            // `created_at` (e.g. ADR-057 dual-write self-send copies) are ordered
+            // deterministically by `full_id`, so an id cursor sitting on one of them
+            // never skips or re-includes its tie.
+            AfterCursor::Id {
+                created_at,
+                full_id,
+            } => {
+                let row_key = (r.created_at, r.full_id);
+                let cursor_key = (*created_at, *full_id);
+                match order {
+                    // "after" in desc order means further along the desc sequence,
+                    // i.e. strictly older (smaller key) than the cursor.
+                    "desc" => row_key < cursor_key,
+                    _ => row_key > cursor_key,
+                }
+            }
+            AfterCursor::Timestamp { micros } => match order {
+                "desc" => r.created_at < *micros,
+                _ => r.created_at > *micros,
+            },
+        });
+    }
+
+    // Total order: sort by `(created_at, full_id)` — the same tuple the cursor
+    // filter above compares against — ascending for order="asc", reversed for
+    // "desc". Sorting on timestamp alone (prior behavior) left ties among
+    // same-microsecond rows in query-return order, which is not stable across
+    // pages/backends.
+    rows.sort_by(|a, b| {
+        let a_key = (a.created_at, a.full_id);
+        let b_key = (b.created_at, b.full_id);
+        match order {
+            "desc" => b_key.cmp(&a_key),
+            _ => a_key.cmp(&b_key),
+        }
     });
-    messages.truncate(limit);
-    let count = messages.len();
+    rows.truncate(limit);
+    let count = rows.len();
+    let messages: Vec<Value> = rows.into_iter().map(|r| r.json).collect();
 
     Ok(json!({
         "thread_id": canonical_thread_id,
@@ -581,6 +730,25 @@ pub(crate) async fn handle_thread(
     }))
 }
 
+/// A thread row carries the sort/cursor key (`created_at`, `full_id`) alongside
+/// the already-rendered message JSON, so the total-order sort and cursor filter
+/// in `handle_thread` compare exact `(i64, Uuid)` tuples instead of re-parsing
+/// the ISO string embedded in the JSON.
+struct ThreadRow {
+    created_at: i64,
+    full_id: Uuid,
+    json: Value,
+}
+
+/// `after` cursor resolved to a comparable key. An id cursor carries the full
+/// `(created_at, full_id)` tuple of the referenced message for tie-breaking; a
+/// timestamp cursor carries only the parsed microsecond value (there is no
+/// specific row to break ties against).
+enum AfterCursor {
+    Id { created_at: i64, full_id: Uuid },
+    Timestamp { micros: i64 },
+}
+
 /// `ingest` — write a single inbound message note from a channel adapter.
 ///
 /// This is a `Visibility::Subhandler` verb: it is not accessible via the MCP
diff --git a/crates/khive-pack-comm/src/message.rs b/crates/khive-pack-comm/src/message.rs
index a1edf025..bb7e40fa 100644
--- a/crates/khive-pack-comm/src/message.rs
+++ b/crates/khive-pack-comm/src/message.rs
@@ -171,6 +171,7 @@ pub(crate) async fn dual_write_message(
     to_actor: Option<&str>,
     in_reply_to_message_id: Option<&str>,
     references_chain: Option<&str>,
+    tags: Option<&[String]>,
 ) -> Result<Note, RuntimeError> {
     let recipient_ns_str = to.trim();
     if from != recipient_ns_str {
@@ -231,6 +232,11 @@ pub(crate) async fn dual_write_message(
     if let Some(refs) = references_chain {
         outbound_props["references_chain"] = json!(refs);
     }
+    if let Some(t) = tags {
+        if !t.is_empty() {
+            outbound_props["tags"] = json!(t);
+        }
+    }
 
     let outbound_note = runtime
         .create_note(
@@ -321,6 +327,11 @@ pub(crate) async fn dual_write_message(
         if let Some(refs) = references_chain {
             inbound_props["references_chain"] = json!(refs);
         }
+        if let Some(t) = tags {
+            if !t.is_empty() {
+                inbound_props["tags"] = json!(t);
+            }
+        }
 
         let inbound_result = runtime
             .create_note(
@@ -412,6 +423,7 @@ mod tests {
             None,
             None,
             None,
+            None,
         )
         .await;
 
diff --git a/crates/khive-pack-comm/src/params.rs b/crates/khive-pack-comm/src/params.rs
index 8a07d5cb..57055352 100644
--- a/crates/khive-pack-comm/src/params.rs
+++ b/crates/khive-pack-comm/src/params.rs
@@ -15,6 +15,11 @@ pub(crate) struct SendParams {
     pub subject: Option<String>,
     #[serde(default)]
     pub thread_id: Option<String>,
+    /// Structured provenance tags (e.g. run id, job id, traffic class), persisted
+    /// verbatim to `properties["tags"]` on both the outbound and inbound copies.
+    /// Mirrors the shipped `memory.remember` `tags` precedent (issue #495).
+    #[serde(default)]
+    pub tags: Option<Vec<String>>,
 }
 
 #[derive(Deserialize)]
@@ -24,6 +29,13 @@ pub(crate) struct InboxParams {
     pub limit: Option<u32>,
     #[serde(default)]
     pub status: Option<String>,
+    /// Exact match on `properties.from_actor`. Mutually exclusive with `from_prefix`.
+    #[serde(default)]
+    pub from_actor: Option<String>,
+    /// Prefix match on `properties.from_actor` (e.g. `"agent:khive:"` selects all
+    /// agents under one namespace). Mutually exclusive with `from_actor`.
+    #[serde(default)]
+    pub from_prefix: Option<String>,
 }
 
 #[derive(Deserialize)]
@@ -37,6 +49,10 @@ pub(crate) struct ReadParams {
 pub(crate) struct ReplyParams {
     pub id: String,
     pub content: String,
+    /// Structured provenance tags, persisted verbatim to `properties["tags"]` on
+    /// both the outbound and inbound copies of the reply (issue #495).
+    #[serde(default)]
+    pub tags: Option<Vec<String>>,
 }
 
 #[derive(Deserialize)]
@@ -48,6 +64,21 @@ pub(crate) struct ThreadParams {
     pub id: String,
     #[serde(default)]
     pub limit: Option<u32>,
+    /// Ordering of the returned messages: `"asc"` (default, chronological) |
+    /// `"desc"` (newest first). Truncation to `limit` applies after ordering,
+    /// so `order="desc"` returns the newest `limit` messages instead of the
+    /// oldest (issue #494 — long threads previously lost the tail).
+    #[serde(default)]
+    pub order: Option<String>,
+    /// Cursor: a message id (short prefix or full UUID) or an RFC 3339
+    /// timestamp. Id cursors compare on the total order `(created_at,
+    /// full_id)` so equal-timestamp messages are never skipped or duplicated;
+    /// timestamp cursors are parsed (any valid RFC 3339 form) rather than
+    /// string-compared. "After" is relative to the chosen `order` (for
+    /// `desc`, strictly older in the descending sequence). An unparseable or
+    /// unresolvable cursor is an error, never silently ignored.
+    #[serde(default)]
+    pub after: Option<String>,
 }
 
 /// Parameters for `comm.ingest` — ingests a single inbound message from a channel adapter.
diff --git a/crates/khive-pack-comm/src/vocab.rs b/crates/khive-pack-comm/src/vocab.rs
index 8c433ed5..6687bfa0 100644
--- a/crates/khive-pack-comm/src/vocab.rs
+++ b/crates/khive-pack-comm/src/vocab.rs
@@ -65,6 +65,12 @@ pub(crate) static COMM_HANDLERS: [HandlerDef; 9] = [
                 required: false,
                 description: "Optional UUID to group messages into a thread.",
             },
+            ParamDef {
+                name: "tags",
+                param_type: "array of string",
+                required: false,
+                description: "Structured provenance tags (e.g. run id, job id, traffic class), persisted verbatim to `properties[\"tags\"]` on both the outbound and inbound copies.",
+            },
         ],
     },
     HandlerDef {
@@ -85,6 +91,18 @@ pub(crate) static COMM_HANDLERS: [HandlerDef; 9] = [
                 required: false,
                 description: "Filter by read status: \"unread\" (default) | \"read\" | \"all\".",
             },
+            ParamDef {
+                name: "from_actor",
+                param_type: "string",
+                required: false,
+                description: "Exact match on the sender's actor label (`properties.from_actor`). Mutually exclusive with `from_prefix`.",
+            },
+            ParamDef {
+                name: "from_prefix",
+                param_type: "string",
+                required: false,
+                description: "Prefix match on the sender's actor label (e.g. `\"agent:khive:\"` selects all agents under one namespace). Mutually exclusive with `from_actor`.",
+            },
         ],
     },
     HandlerDef {
@@ -117,6 +135,12 @@ pub(crate) static COMM_HANDLERS: [HandlerDef; 9] = [
                 required: true,
                 description: "Reply body. Must not be empty.",
             },
+            ParamDef {
+                name: "tags",
+                param_type: "array of string",
+                required: false,
+                description: "Structured provenance tags, persisted verbatim to `properties[\"tags\"]` on both the outbound and inbound copies of the reply.",
+            },
         ],
     },
     HandlerDef {
@@ -135,7 +159,19 @@ pub(crate) static COMM_HANDLERS: [HandlerDef; 9] = [
                 name: "limit",
                 param_type: "integer",
                 required: false,
-                description: "Max messages to return. Default 100, max 500.",
+                description: "Max messages to return. Default 100, max 500. Truncation applies after ordering, so order=\"desc\" + limit returns the newest `limit` messages.",
+            },
+            ParamDef {
+                name: "order",
+                param_type: "string",
+                required: false,
+                description: "Ordering of returned messages: \"asc\" (default, chronological) | \"desc\" (newest first).",
+            },
+            ParamDef {
+                name: "after",
+                param_type: "string",
+                required: false,
+                description: "Cursor: a message id (short prefix or full UUID) or an RFC 3339 timestamp (any valid form, e.g. whole-second `Z` or `+00:00` offset). An id cursor ties-break on (created_at, full_id) so equal-timestamp messages are never skipped or duplicated. Only messages strictly after that point in the chosen `order` are returned; an unparseable value is a hard error.",
             },
         ],
     },
diff --git a/crates/khive-pack-comm/tests/integration.rs b/crates/khive-pack-comm/tests/integration.rs
index f6a071d5..478fea05 100644
--- a/crates/khive-pack-comm/tests/integration.rs
+++ b/crates/khive-pack-comm/tests/integration.rs
@@ -5565,3 +5565,828 @@ async fn health_channel_entry_never_carries_a_healthy_bool() {
         "channel entry must never carry a computed healthy bool: {ch:?}"
     );
 }
+
+// ── #493: comm.inbox from_actor / from_prefix sender filter ─────────────────
+
+/// A single actor namespace receives messages from two distinct senders;
+/// `from_actor` (exact match) selects only the messages from the named sender.
+#[tokio::test]
+async fn t493_inbox_from_actor_filters_to_exact_sender() {
+    let backend = shared_backend();
+    let (registry_a, _rt_a) = build_actor_registry(backend.clone(), "lambda:a");
+    let (registry_b, _rt_b) = build_actor_registry(backend.clone(), "lambda:b");
+    let (registry_c, _rt_c) = build_actor_registry(backend.clone(), "lambda:c");
+
+    registry_a
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:c", "content": "hi from A" }),
+        )
+        .await
+        .expect("A send succeeds");
+    registry_b
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:c", "content": "hi from B" }),
+        )
+        .await
+        .expect("B send succeeds");
+
+    let filtered = registry_c
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({ "status": "all", "limit": 50, "from_actor": "lambda:a" }),
+        )
+        .await
+        .expect("filtered inbox succeeds");
+    let messages = filtered["messages"].as_array().expect("messages array");
+    assert_eq!(
+        messages.len(),
+        1,
+        "from_actor=lambda:a must return exactly 1 message; got {messages:?}"
+    );
+    assert_eq!(
+        messages[0]["properties"]["from_actor"].as_str(),
+        Some("lambda:a")
+    );
+}
+
+/// `from_prefix` selects all senders whose actor label starts with the given prefix,
+/// e.g. `"agent:khive:"` selects every spawned agent under that namespace.
+#[tokio::test]
+async fn t493_inbox_from_prefix_filters_to_matching_senders() {
+    let backend = shared_backend();
+    let (registry_a1, _rt_a1) = build_actor_registry(backend.clone(), "agent:khive:role-1");
+    let (registry_a2, _rt_a2) = build_actor_registry(backend.clone(), "agent:khive:role-2");
+    let (registry_other, _rt_other) = build_actor_registry(backend.clone(), "lambda:other");
+    let (registry_c, _rt_c) = build_actor_registry(backend.clone(), "lambda:c");
+
+    registry_a1
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:c", "content": "status from role-1" }),
+        )
+        .await
+        .expect("a1 send succeeds");
+    registry_a2
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:c", "content": "status from role-2" }),
+        )
+        .await
+        .expect("a2 send succeeds");
+    registry_other
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:c", "content": "unrelated message" }),
+        )
+        .await
+        .expect("other send succeeds");
+
+    let filtered = registry_c
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({ "status": "all", "limit": 50, "from_prefix": "agent:khive:" }),
+        )
+        .await
+        .expect("filtered inbox succeeds");
+    let messages = filtered["messages"].as_array().expect("messages array");
+    assert_eq!(
+        messages.len(),
+        2,
+        "from_prefix=agent:khive: must return the 2 agent messages, excluding lambda:other; got {messages:?}"
+    );
+    for m in messages {
+        let from_actor = m["properties"]["from_actor"].as_str().unwrap_or("");
+        assert!(
+            from_actor.starts_with("agent:khive:"),
+            "every returned message must have a from_actor matching the prefix; got {from_actor:?}"
+        );
+    }
+}
+
+/// Supplying both `from_actor` and `from_prefix` is a per-op error naming the conflict.
+#[tokio::test]
+async fn t493_inbox_from_actor_and_from_prefix_mutually_exclusive() {
+    let backend = shared_backend();
+    let (registry, _rt) = build_actor_registry(backend, "lambda:c");
+
+    let result = registry
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({
+                "from_actor": "lambda:a",
+                "from_prefix": "agent:khive:",
+            }),
+        )
+        .await;
+    assert!(
+        result.is_err(),
+        "from_actor + from_prefix together must be rejected; got {result:?}"
+    );
+    let err = result.unwrap_err().to_string();
+    assert!(
+        err.contains("mutually exclusive"),
+        "error must name the conflict; got: {err}"
+    );
+}
+
+/// Absent from_actor/from_prefix preserves today's behavior exactly: no sender filter
+/// is applied and both senders' messages are returned.
+#[tokio::test]
+async fn t493_inbox_without_sender_filter_returns_all_senders() {
+    let backend = shared_backend();
+    let (registry_a, _rt_a) = build_actor_registry(backend.clone(), "lambda:a");
+    let (registry_b, _rt_b) = build_actor_registry(backend.clone(), "lambda:b");
+    let (registry_c, _rt_c) = build_actor_registry(backend.clone(), "lambda:c");
+
+    registry_a
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:c", "content": "hi from A" }),
+        )
+        .await
+        .expect("A send succeeds");
+    registry_b
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:c", "content": "hi from B" }),
+        )
+        .await
+        .expect("B send succeeds");
+
+    let unfiltered = registry_c
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({ "status": "all", "limit": 50 }),
+        )
+        .await
+        .expect("unfiltered inbox succeeds");
+    let messages = unfiltered["messages"].as_array().expect("messages array");
+    assert_eq!(
+        messages.len(),
+        2,
+        "no sender filter must return both senders' messages unchanged; got {messages:?}"
+    );
+}
+
+// ── #494: comm.thread tail pagination (order + after cursor) ────────────────
+//
+// NOTE: `comm.send`/`comm.reply` targeting the caller's own namespace ("local")
+// write BOTH an outbound and an inbound copy of every logical message into that
+// same namespace (dual_write_message, ADR-057) — so each `content` string below
+// appears TWICE in an unfiltered thread(), consecutively (outbound then inbound),
+// since the inbound copy is always written a moment after the outbound copy in
+// the same call. Tests account for this pairing explicitly rather than assuming
+// one physical note per logical send (matches the existing #485/H3 tests' use of
+// tolerant `>=` counts for the same reason).
+
+/// Default order ("asc") truncates from the tail — this is the pre-existing
+/// (buggy, per #494) behavior that must stay byte-identical: a thread longer
+/// than `limit` returns the HEAD (oldest messages), not the newest.
+#[tokio::test]
+async fn t494_thread_default_order_truncates_head_unchanged() {
+    let (registry, _rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+
+    for i in 1..=4 {
+        registry
+            .dispatch(
+                "comm.reply",
+                serde_json::json!({ "id": root_full_id, "content": format!("reply-{i}") }),
+            )
+            .await
+            .unwrap_or_else(|e| panic!("reply-{i} succeeds: {e:?}"));
+    }
+
+    // 5 logical messages (root + 4 replies) = 10 physical notes (outbound+inbound
+    // pairs). limit=2 with no order= must return the OLDEST 2 physical notes —
+    // both copies of "root" — matching pre-#494 truncate-from-head behavior.
+    let result = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "limit": 2 }),
+        )
+        .await
+        .expect("thread succeeds");
+    let msgs = result["messages"].as_array().expect("messages array");
+    assert_eq!(msgs.len(), 2, "limit=2 must return exactly 2 messages");
+    let contents: Vec<&str> = msgs
+        .iter()
+        .map(|m| m["content"].as_str().unwrap_or(""))
+        .collect();
+    assert_eq!(
+        contents,
+        vec!["root", "root"],
+        "default order must truncate from the tail (keep the head), unchanged from before #494"
+    );
+}
+
+/// `order="desc"` returns the newest `limit` messages instead of the oldest — the
+/// #494 fix: long threads can now reach their tail.
+#[tokio::test]
+async fn t494_thread_order_desc_returns_newest_messages() {
+    let (registry, _rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+
+    for i in 1..=4 {
+        registry
+            .dispatch(
+                "comm.reply",
+                serde_json::json!({ "id": root_full_id, "content": format!("reply-{i}") }),
+            )
+            .await
+            .unwrap_or_else(|e| panic!("reply-{i} succeeds: {e:?}"));
+    }
+
+    let result = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "limit": 2, "order": "desc" }),
+        )
+        .await
+        .expect("thread succeeds");
+    let msgs = result["messages"].as_array().expect("messages array");
+    assert_eq!(msgs.len(), 2, "limit=2 must return exactly 2 messages");
+    let contents: Vec<&str> = msgs
+        .iter()
+        .map(|m| m["content"].as_str().unwrap_or(""))
+        .collect();
+    assert_eq!(
+        contents,
+        vec!["reply-4", "reply-4"],
+        "order=desc + limit=2 must return the newest 2 physical notes — both copies \
+         of the last reply — not the oldest (#494 fix: the tail is now reachable)"
+    );
+}
+
+/// An invalid `order` value is rejected, naming the valid set (ADR-084 Rule 2).
+#[tokio::test]
+async fn t494_thread_invalid_order_rejected() {
+    let (registry, _rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+
+    let result = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "order": "banana" }),
+        )
+        .await;
+    assert!(
+        result.is_err(),
+        "order=banana must be rejected; got {result:?}"
+    );
+    let err = result.unwrap_err().to_string();
+    assert!(
+        err.contains("asc") && err.contains("desc"),
+        "error must name the valid order values; got: {err}"
+    );
+}
+
+/// `after` accepts a message id cursor and returns only messages strictly after it
+/// (enables incremental polling without re-fetching history). The cursor resolves
+/// to the OUTBOUND copy's `full_id` (what `comm.reply` returns); its own inbound
+/// copy — created a moment later in the same dual-write call — is strictly after
+/// it and so is included.
+#[tokio::test]
+async fn t494_thread_after_id_cursor_returns_strictly_later_messages() {
+    let (registry, _rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+
+    let reply1 = registry
+        .dispatch(
+            "comm.reply",
+            serde_json::json!({ "id": root_full_id, "content": "reply-1" }),
+        )
+        .await
+        .expect("reply-1 succeeds");
+    let reply1_full_id = reply1["full_id"]
+        .as_str()
+        .expect("reply1 full_id")
+        .to_string();
+
+    let result = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "after": reply1_full_id }),
+        )
+        .await
+        .expect("thread succeeds");
+    let msgs = result["messages"].as_array().expect("messages array");
+    let contents: Vec<&str> = msgs
+        .iter()
+        .map(|m| m["content"].as_str().unwrap_or(""))
+        .collect();
+    assert_eq!(
+        contents,
+        vec!["reply-1"],
+        "after=reply-1's outbound id must return only its own inbound copy \
+         (strictly later), excluding root and reply-1's own outbound copy; got {contents:?}"
+    );
+}
+
+/// Insert a `message` note directly into the store with an explicit `created_at`,
+/// bypassing `comm.send`/`comm.reply`. Cursor/tie-break/ordering tests need exact
+/// control over timestamps (including two rows sharing the same microsecond) that
+/// racing the wall clock through the normal dispatch path cannot guarantee.
+async fn insert_thread_message(
+    rt: &KhiveRuntime,
+    ns: &str,
+    id: uuid::Uuid,
+    thread_id: uuid::Uuid,
+    created_at: i64,
+    content: &str,
+) {
+    let token = rt
+        .authorize(Namespace::parse(ns).expect("valid namespace"))
+        .expect("authorize");
+    let store = rt.notes(&token).expect("notes store");
+    store
+        .upsert_note(khive_storage::note::Note {
+            id,
+            namespace: ns.to_string(),
+            kind: "message".into(),
+            status: "active".into(),
+            name: None,
+            content: content.to_string(),
+            salience: None,
+            decay_factor: None,
+            expires_at: None,
+            properties: Some(serde_json::json!({
+                "direction": "inbound",
+                "from": "x",
+                "to": ns,
+                "read": false,
+                "thread_id": thread_id.as_hyphenated().to_string(),
+            })),
+            created_at,
+            updated_at: created_at,
+            deleted_at: None,
+        })
+        .await
+        .expect("insert message");
+}
+
+/// codex r1 (#494 re-review): two physical messages that share the exact same
+/// microsecond `created_at` (e.g. what an ADR-057 dual-write self-send can
+/// produce) must not be skipped or duplicated around an id cursor — the cursor
+/// filter and sort must compare the full `(created_at, full_id)` tuple, not
+/// timestamp alone.
+#[tokio::test]
+async fn t494_thread_after_id_cursor_ties_on_equal_created_at_no_skip_no_dup() {
+    let (registry, rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+    let root_uuid = uuid::Uuid::parse_str(&root_full_id).unwrap();
+
+    let tied_at = chrono::Utc::now().timestamp_micros();
+    let uuid_lo = uuid::Uuid::parse_str("00000000-0000-4000-8000-000000000001").unwrap();
+    let uuid_hi = uuid::Uuid::parse_str("ffffffff-0000-4000-8000-000000000002").unwrap();
+    insert_thread_message(&rt, "local", uuid_lo, root_uuid, tied_at, "tied-lo").await;
+    insert_thread_message(&rt, "local", uuid_hi, root_uuid, tied_at, "tied-hi").await;
+
+    let after_lo = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "after": uuid_lo.to_string() }),
+        )
+        .await
+        .expect("thread succeeds");
+    let contents_lo: Vec<&str> = after_lo["messages"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|m| m["content"].as_str().unwrap_or(""))
+        .collect();
+    assert_eq!(
+        contents_lo,
+        vec!["tied-hi"],
+        "after=lo must return exactly the higher-uuid tied row once, not skip or \
+         duplicate it; got {contents_lo:?}"
+    );
+
+    let after_hi = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "after": uuid_hi.to_string() }),
+        )
+        .await
+        .expect("thread succeeds");
+    let contents_hi: Vec<&str> = after_hi["messages"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|m| m["content"].as_str().unwrap_or(""))
+        .collect();
+    assert!(
+        contents_hi.is_empty(),
+        "after=hi must return nothing — hi is the greatest key among the tied rows; \
+         got {contents_hi:?}"
+    );
+}
+
+/// codex r1 (#494 re-review): an `after` timestamp cursor must be parsed to
+/// microseconds (not compared as a raw string) so non-canonical but valid RFC
+/// 3339 forms — whole-second `Z`, or an explicit `+00:00` offset — compare
+/// correctly against khive's canonical microsecond timestamps.
+#[tokio::test]
+async fn t494_thread_after_timestamp_cursor_accepts_noncanonical_rfc3339() {
+    let (registry, rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+    let root_uuid = uuid::Uuid::parse_str(&root_full_id).unwrap();
+
+    // Far in the future so the real-clock root note (created "now") is always
+    // strictly before the cursor and never leaks into the filtered result.
+    let ts1 = chrono::DateTime::parse_from_rfc3339("2099-01-01T00:00:00Z")
+        .unwrap()
+        .timestamp_micros();
+    let ts2 = ts1 + 1_000_000;
+    let id1 = uuid::Uuid::parse_str("11111111-0000-4000-8000-000000000001").unwrap();
+    let id2 = uuid::Uuid::parse_str("22222222-0000-4000-8000-000000000002").unwrap();
+    insert_thread_message(&rt, "local", id1, root_uuid, ts1, "at-ts1").await;
+    insert_thread_message(&rt, "local", id2, root_uuid, ts2, "at-ts2").await;
+
+    for cursor in ["2099-01-01T00:00:00Z", "2099-01-01T00:00:00+00:00"] {
+        let result = registry
+            .dispatch(
+                "comm.thread",
+                serde_json::json!({ "id": root_full_id, "after": cursor }),
+            )
+            .await
+            .unwrap_or_else(|e| panic!("cursor {cursor:?} must parse: {e:?}"));
+        let contents: Vec<&str> = result["messages"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|m| m["content"].as_str().unwrap_or(""))
+            .collect();
+        assert_eq!(
+            contents,
+            vec!["at-ts2"],
+            "whole-second/offset RFC3339 cursor {cursor:?} must exclude the note at \
+             exactly that instant and include only the strictly-later one; got {contents:?}"
+        );
+    }
+}
+
+/// codex r1 (#494 re-review): an `after` value that is neither a resolvable
+/// message id nor a parseable RFC 3339 timestamp must fail loudly, never be
+/// silently coerced into "no cursor" (which would return the whole thread).
+#[tokio::test]
+async fn t494_thread_after_invalid_string_is_hard_error() {
+    let (registry, _rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+
+    let result = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "after": "not-a-valid-cursor" }),
+        )
+        .await;
+    assert!(
+        result.is_err(),
+        "an `after` value that is neither a resolvable id nor a valid RFC 3339 \
+         timestamp must be a hard error, not silently treated as no-cursor; got {result:?}"
+    );
+    let err = result.unwrap_err().to_string();
+    assert!(
+        err.contains("neither a resolvable message id nor a valid RFC 3339 timestamp"),
+        "error must name why the cursor was rejected; got: {err}"
+    );
+}
+
+/// codex r1 (#494 re-review): `order="desc"` combined with an id `after` cursor
+/// must filter against the DESC sequence, not always `created_at >`. "After" in
+/// desc order means further along the desc traversal, i.e. strictly older.
+#[tokio::test]
+async fn t494_thread_order_desc_with_after_id_cursor_returns_strictly_older_in_desc_sequence() {
+    let (registry, rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+    let root_uuid = uuid::Uuid::parse_str(&root_full_id).unwrap();
+
+    let base = chrono::DateTime::parse_from_rfc3339("2099-01-01T00:00:00Z")
+        .unwrap()
+        .timestamp_micros();
+    let id_a = uuid::Uuid::parse_str("aaaaaaaa-0000-4000-8000-000000000001").unwrap();
+    let id_b = uuid::Uuid::parse_str("bbbbbbbb-0000-4000-8000-000000000002").unwrap();
+    let id_c = uuid::Uuid::parse_str("cccccccc-0000-4000-8000-000000000003").unwrap();
+    insert_thread_message(&rt, "local", id_a, root_uuid, base, "msg-a").await;
+    insert_thread_message(&rt, "local", id_b, root_uuid, base + 1_000_000, "msg-b").await;
+    insert_thread_message(&rt, "local", id_c, root_uuid, base + 2_000_000, "msg-c").await;
+
+    // `comm.send(to="local", ...)` is a self-send: ADR-057 dual-write stores
+    // both an outbound and an inbound copy of "root", both real-clock (and so
+    // both strictly older than the synthetic 2099 timestamps). Full desc
+    // sequence is therefore [msg-c, msg-b, msg-a, root, root]. `after=msg-b`
+    // must return only what comes strictly after it in THAT sequence —
+    // msg-a and both root copies — never msg-c, even though msg-c is also
+    // `>` msg-b in wall-clock terms.
+    let result = registry
+        .dispatch(
+            "comm.thread",
+            serde_json::json!({ "id": root_full_id, "order": "desc", "after": id_b.to_string() }),
+        )
+        .await
+        .expect("thread succeeds");
+    let contents: Vec<&str> = result["messages"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|m| m["content"].as_str().unwrap_or(""))
+        .collect();
+    assert_eq!(
+        contents,
+        vec!["msg-a", "root", "root"],
+        "order=desc + after=msg-b must return only rows strictly older than msg-b \
+         (further along the desc sequence), in desc order — both self-send root \
+         copies included, msg-c excluded; got {contents:?}"
+    );
+}
+
+/// Absent `order`/`after` preserves today's behavior exactly: same messages, same order,
+/// same truncation as before #494 (regression guard alongside the existing #485 test).
+#[tokio::test]
+async fn t494_thread_without_new_params_unchanged() {
+    let (registry, _rt) = build_registry_for_ns("local");
+
+    let root = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "local", "content": "root" }),
+        )
+        .await
+        .expect("root send succeeds");
+    let root_full_id = root["full_id"].as_str().expect("root full_id").to_string();
+
+    registry
+        .dispatch(
+            "comm.reply",
+            serde_json::json!({ "id": root_full_id, "content": "reply-1" }),
+        )
+        .await
+        .expect("reply-1 succeeds");
+
+    let result = registry
+        .dispatch("comm.thread", serde_json::json!({ "id": root_full_id }))
+        .await
+        .expect("thread succeeds");
+    let msgs = result["messages"].as_array().expect("messages array");
+    assert_eq!(
+        msgs.len(),
+        4,
+        "root (outbound+inbound) + reply-1 (outbound+inbound) = 4 physical notes"
+    );
+    let contents: Vec<&str> = msgs
+        .iter()
+        .map(|m| m["content"].as_str().unwrap_or(""))
+        .collect();
+    assert_eq!(contents, vec!["root", "root", "reply-1", "reply-1"]);
+}
+
+// ── #495: comm.send / comm.reply metadata (tags) passthrough ────────────────
+
+/// `comm.send(tags=[...])` persists the tags into `properties["tags"]` on the
+/// inbound copy, round-tripped via `comm.inbox`.
+#[tokio::test]
+async fn t495_send_tags_roundtrip_via_inbox() {
+    let backend = shared_backend();
+    let (registry_a, _rt_a) = build_actor_registry(backend.clone(), "lambda:a");
+    let (registry_b, _rt_b) = build_actor_registry(backend, "lambda:b");
+
+    registry_a
+        .dispatch(
+            "comm.send",
+            serde_json::json!({
+                "to": "lambda:b",
+                "content": "tagged message",
+                "tags": ["run:abc123", "traffic:agent"],
+            }),
+        )
+        .await
+        .expect("tagged send succeeds");
+
+    let inbox = registry_b
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({ "status": "all", "limit": 10 }),
+        )
+        .await
+        .expect("inbox succeeds");
+    let messages = inbox["messages"].as_array().expect("messages array");
+    assert_eq!(messages.len(), 1);
+    let tags = messages[0]["properties"]["tags"]
+        .as_array()
+        .expect("tags array present on inbound copy");
+    let tag_strs: Vec<&str> = tags.iter().map(|t| t.as_str().unwrap_or("")).collect();
+    assert_eq!(tag_strs, vec!["run:abc123", "traffic:agent"]);
+}
+
+/// `comm.send(tags=[...])` also persists on the outbound copy, round-tripped
+/// via `comm.read` after resolving the sender's own outbound note.
+#[tokio::test]
+async fn t495_send_tags_present_on_outbound_copy() {
+    let backend = shared_backend();
+    let (registry_a, rt_a) = build_actor_registry(backend, "lambda:a");
+
+    let send_result = registry_a
+        .dispatch(
+            "comm.send",
+            serde_json::json!({
+                "to": "lambda:a",
+                "content": "self-tagged",
+                "tags": ["job:42"],
+            }),
+        )
+        .await
+        .expect("tagged self-send succeeds");
+    let outbound_full_id = send_result["full_id"].as_str().expect("full_id");
+    let outbound_uuid: uuid::Uuid = outbound_full_id.parse().expect("valid uuid");
+
+    let tok = rt_a.authorize(Namespace::parse("local").unwrap()).unwrap();
+    let store = rt_a.notes(&tok).expect("notes store");
+    let note = store
+        .get_note(outbound_uuid)
+        .await
+        .expect("get_note succeeds")
+        .expect("outbound note exists");
+    let tags = note.properties.as_ref().and_then(|p| p.get("tags"));
+    assert_eq!(
+        tags,
+        Some(&serde_json::json!(["job:42"])),
+        "outbound copy must also carry tags"
+    );
+}
+
+/// `comm.reply(tags=[...])` persists tags on the reply's inbound copy.
+#[tokio::test]
+async fn t495_reply_tags_roundtrip_via_inbox() {
+    let backend = shared_backend();
+    let (registry_a, _rt_a) = build_actor_registry(backend.clone(), "lambda:a");
+    let (registry_b, _rt_b) = build_actor_registry(backend, "lambda:b");
+
+    registry_a
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:b", "content": "hello" }),
+        )
+        .await
+        .expect("send succeeds");
+
+    let inbox_b = registry_b
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({ "status": "all", "limit": 10 }),
+        )
+        .await
+        .expect("B inbox succeeds");
+    let b_inbound_id = inbox_b["messages"][0]["full_id"]
+        .as_str()
+        .expect("B inbound full_id");
+
+    registry_b
+        .dispatch(
+            "comm.reply",
+            serde_json::json!({
+                "id": b_inbound_id,
+                "content": "reply with tags",
+                "tags": ["job:reply-1"],
+            }),
+        )
+        .await
+        .expect("tagged reply succeeds");
+
+    let inbox_a = registry_a
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({ "status": "all", "limit": 10 }),
+        )
+        .await
+        .expect("A inbox succeeds");
+    let a_messages = inbox_a["messages"].as_array().expect("messages array");
+    let reply_msg = a_messages
+        .iter()
+        .find(|m| m["content"] == "reply with tags")
+        .expect("A's inbox contains the tagged reply");
+    let tags = reply_msg["properties"]["tags"]
+        .as_array()
+        .expect("tags array present on reply's inbound copy");
+    let tag_strs: Vec<&str> = tags.iter().map(|t| t.as_str().unwrap_or("")).collect();
+    assert_eq!(tag_strs, vec!["job:reply-1"]);
+}
+
+/// Absent `tags` preserves today's behavior exactly: no `properties["tags"]` key at all.
+#[tokio::test]
+async fn t495_send_without_tags_omits_tags_property() {
+    let backend = shared_backend();
+    let (registry_a, _rt_a) = build_actor_registry(backend.clone(), "lambda:a");
+    let (registry_b, _rt_b) = build_actor_registry(backend, "lambda:b");
+
+    registry_a
+        .dispatch(
+            "comm.send",
+            serde_json::json!({ "to": "lambda:b", "content": "no tags here" }),
+        )
+        .await
+        .expect("send succeeds");
+
+    let inbox = registry_b
+        .dispatch(
+            "comm.inbox",
+            serde_json::json!({ "status": "all", "limit": 10 }),
+        )
+        .await
+        .expect("inbox succeeds");
+    let messages = inbox["messages"].as_array().expect("messages array");
+    assert_eq!(messages.len(), 1);
+    assert!(
+        messages[0]["properties"].get("tags").is_none(),
+        "absent tags must not add a properties.tags key; got {:?}",
+        messages[0]["properties"]
+    );
+}
+
+/// `comm.send` with an unknown top-level field (typo) is still rejected —
+/// `tags` addition must not have loosened `deny_unknown_fields`.
+#[tokio::test]
+async fn t495_send_rejects_unknown_field_alongside_tags() {
+    let (registry, _rt) = build_registry_for_ns("local");
+
+    let result = registry
+        .dispatch(
+            "comm.send",
+            serde_json::json!({
+                "to": "local",
+                "content": "hi",
+                "tags": ["a"],
+                "bogus_field": "typo",
+            }),
+        )
+        .await;
+    assert!(
+        result.is_err(),
+        "unknown field alongside tags must still be rejected; got {result:?}"
+    );
+}
diff --git a/crates/khive-pack-gtd/src/handlers.rs b/crates/khive-pack-gtd/src/handlers.rs
index b420c0bf..dcc4b3bf 100644
--- a/crates/khive-pack-gtd/src/handlers.rs
+++ b/crates/khive-pack-gtd/src/handlers.rs
@@ -776,6 +776,12 @@ impl GtdPack {
         params: Value,
     ) -> Result<Value, RuntimeError> {
         let p: NextParams = deser(params)?;
+        // #744: this clamp is silent by design here — the response shape is a bare
+        // JSON array (`Value::Array`), consumed directly via `.as_array()` by every
+        // caller in this crate and beyond (kkernel, li surfaces). Adding a sibling
+        // `truncated` field would require wrapping the response in an object, which
+        // is a breaking shape change, not an additive one. The cap is documented on
+        // the `limit` ParamDef instead (issue #744 fallback ask 1).
         let limit = p.limit.unwrap_or(10).clamp(1, 200);
 
         // Pull a broad window of recent tasks, filter in-memory by GTD status.
@@ -987,6 +993,9 @@ impl GtdPack {
         params: Value,
     ) -> Result<Value, RuntimeError> {
         let p: TasksParams = deser(params)?;
+        // #744: silent clamp, documented rather than signaled — see the identical
+        // note in `handle_next` above (bare-array response shape rules out an
+        // additive `truncated` field).
         let limit = p.limit.unwrap_or(50).clamp(1, 200);
         let offset = p.offset.unwrap_or(0) as usize;
 
diff --git a/crates/khive-pack-gtd/src/vocab.rs b/crates/khive-pack-gtd/src/vocab.rs
index 52346762..4d64f484 100644
--- a/crates/khive-pack-gtd/src/vocab.rs
+++ b/crates/khive-pack-gtd/src/vocab.rs
@@ -161,7 +161,7 @@ pub(crate) static GTD_HANDLERS: [HandlerDef; 5] = [
                 name: "limit",
                 param_type: "integer",
                 required: false,
-                description: "Maximum tasks to return (default 10).",
+                description: "Maximum tasks to return (default 10, silently clamped to 200 — issue #744: a `limit` above 200 is capped without a separate signal in the response).",
             },
             ParamDef {
                 name: "assignee",
@@ -229,7 +229,7 @@ pub(crate) static GTD_HANDLERS: [HandlerDef; 5] = [
                 name: "limit",
                 param_type: "integer",
                 required: false,
-                description: "Maximum results (default 20).",
+                description: "Maximum results (default 20, silently clamped to 200 — issue #744: a `limit` above 200 is capped without a separate signal in the response).",
             },
             ParamDef {
                 name: "offset",
diff --git a/crates/khive-pack-gtd/tests/query.rs b/crates/khive-pack-gtd/tests/query.rs
index eacc7bb1..37ef68b6 100644
--- a/crates/khive-pack-gtd/tests/query.rs
+++ b/crates/khive-pack-gtd/tests/query.rs
@@ -235,3 +235,130 @@ async fn next_ordering_is_deterministic_on_equal_priority_and_timestamp() {
         "gtd.next must return identical ordering on repeated calls with the same task set"
     );
 }
+
+// ── #744: gtd.tasks / gtd.next silent 200-row clamp ──────────────────────────
+
+/// Under-limit: `limit` below the 200 cap is unaffected — no change from before #744.
+#[tokio::test]
+async fn next_limit_under_cap_returns_requested_count_unaffected() {
+    let pack = pack(rt());
+    for i in 0..5 {
+        assign(
+            &pack,
+            json!({"title": format!("task-{i}"), "status": "next"}),
+        )
+        .await;
+    }
+
+    let resp = pack
+        .dispatch("gtd.next", json!({"limit": 10}))
+        .await
+        .unwrap();
+    let arr = resp.as_array().unwrap();
+    assert_eq!(
+        arr.len(),
+        5,
+        "limit=10 with only 5 actionable tasks must return all 5, unaffected by the cap"
+    );
+}
+
+/// Over-limit: `gtd.next(limit=500)` is silently clamped to 200 (issue #744 — the cap
+/// itself is intentional and unchanged; #744 is about the *silence*, addressed here by
+/// documenting the cap on the `limit` ParamDef rather than a response-shape change,
+/// since the response is a bare JSON array consumed via `.as_array()` throughout the
+/// codebase and a sibling `truncated` field would require a breaking wrap-in-object
+/// change).
+#[tokio::test]
+async fn next_limit_over_cap_clamps_to_200() {
+    let pack = pack(rt());
+    for i in 0..205 {
+        assign(
+            &pack,
+            json!({"title": format!("task-{i}"), "status": "next"}),
+        )
+        .await;
+    }
+
+    let resp = pack
+        .dispatch("gtd.next", json!({"limit": 500}))
+        .await
+        .unwrap();
+    let arr = resp.as_array().unwrap();
+    assert_eq!(
+        arr.len(),
+        200,
+        "limit=500 over 205 actionable tasks must clamp to exactly 200"
+    );
+}
+
+/// Under-limit: `gtd.tasks(limit=...)` below the cap is unaffected.
+#[tokio::test]
+async fn tasks_limit_under_cap_returns_requested_count_unaffected() {
+    let pack = pack(rt());
+    for i in 0..5 {
+        assign(
+            &pack,
+            json!({"title": format!("task-{i}"), "status": "next"}),
+        )
+        .await;
+    }
+
+    let resp = pack
+        .dispatch("gtd.tasks", json!({"limit": 10}))
+        .await
+        .unwrap();
+    let arr = resp.as_array().unwrap();
+    assert_eq!(arr.len(), 5, "limit=10 with only 5 tasks must return all 5");
+}
+
+/// Over-limit: `gtd.tasks(limit=500)` is silently clamped to 200, mirroring `gtd.next`.
+#[tokio::test]
+async fn tasks_limit_over_cap_clamps_to_200() {
+    let pack = pack(rt());
+    for i in 0..205 {
+        assign(
+            &pack,
+            json!({"title": format!("task-{i}"), "status": "next"}),
+        )
+        .await;
+    }
+
+    let resp = pack
+        .dispatch("gtd.tasks", json!({"limit": 500}))
+        .await
+        .unwrap();
+    let arr = resp.as_array().unwrap();
+    assert_eq!(
+        arr.len(),
+        200,
+        "limit=500 over 205 tasks must clamp to exactly 200"
+    );
+}
+
+/// The 200 cap is documented on both verbs' `limit` ParamDef (issue #744 fallback
+/// ask 1), so `help=true`/verb introspection surfaces it even though the response
+/// itself carries no truncation signal.
+#[tokio::test]
+async fn next_and_tasks_limit_param_documents_the_200_cap() {
+    use khive_pack_gtd::GtdPack;
+    use khive_runtime::pack::HandlerDef;
+    use khive_types::Pack;
+
+    let handlers: &[HandlerDef] = GtdPack::HANDLERS;
+    for verb in ["gtd.next", "gtd.tasks"] {
+        let h = handlers
+            .iter()
+            .find(|h| h.name == verb)
+            .unwrap_or_else(|| panic!("{verb} must be declared"));
+        let limit = h
+            .params
+            .iter()
+            .find(|p| p.name == "limit")
+            .unwrap_or_else(|| panic!("{verb} must declare a limit param"));
+        assert!(
+            limit.description.contains("200"),
+            "{verb}.limit description must document the 200 cap; got: {:?}",
+            limit.description
+        );
+    }
+}