From 094be758b14aadd9dadb9066973da2a92f618b06 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Wed, 29 Apr 2026 21:10:56 +0000
Subject: [PATCH 1/3] Initial plan


From 3a0929123ebc68ac3d309f8399af0334976790cd Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Wed, 29 Apr 2026 21:41:49 +0000
Subject: [PATCH 2/3] Implement Phase 1a and Phase 2 of name/ID size limits
 (size-limits proposal)

Agent-Logs-Url: https://github.com/microsoft/duroxide/sessions/e00db919-62c8-4245-a786-c4e35f8d5fcc

Co-authored-by: pinodeca <32303022+pinodeca@users.noreply.github.com>
---
 src/client/mod.rs                        |  90 ++++-
 src/lib.rs                               |  14 +
 src/runtime/dispatchers/orchestration.rs | 205 ++++++++++-
 src/runtime/limits.rs                    | 358 ++++++++++++++++++-
 src/runtime/mod.rs                       |  11 +
 src/runtime/registry.rs                  |  27 ++
 tests/scenarios.rs                       |   4 +
 tests/scenarios/limits.rs                | 419 +++++++++++++++++++++++
 tests/tag_serde_tests.rs                 |  17 +-
 9 files changed, 1116 insertions(+), 29 deletions(-)
 create mode 100644 tests/scenarios/limits.rs

diff --git a/src/client/mod.rs b/src/client/mod.rs
index d24ebce0..c1c5d186 100644
--- a/src/client/mod.rs
+++ b/src/client/mod.rs
@@ -166,6 +166,12 @@ const POLL_DELAY_MULTIPLIER: u64 = 2;
 /// ```
 pub struct Client {
     store: Arc<dyn Provider>,
+    /// Limits applied to Tier-1 (client-side) name and identifier checks.
+    ///
+    /// Defaults to [`crate::runtime::limits::Limits::permissive()`], preserving
+    /// pre-Phase-2 behavior. Set to [`crate::runtime::limits::Limits::recommended()`]
+    /// (or tighter) to enable enforcement.
+    limits: crate::runtime::limits::Limits,
 }
 
 impl Client {
@@ -184,7 +190,61 @@ impl Client {
     /// let client2 = client.clone();
     /// ```
     pub fn new(store: Arc<dyn Provider>) -> Self {
-        Self { store }
+        Self { store, limits: crate::runtime::limits::Limits::permissive() }
+    }
+
+    /// Create a client with custom size limits applied to Tier-1 (client-side) checks.
+    ///
+    /// Use [`crate::runtime::limits::Limits::recommended()`] to enable the documented
+    /// production defaults, or supply custom limits to tighten or loosen for your
+    /// deployment.
+    ///
+    /// # Parameters
+    ///
+    /// * `store` - Arc-wrapped Provider (same instance used by Runtime)
+    /// * `limits` - Size and shape limits for client-side validation
+    pub fn new_with_limits(store: Arc<dyn Provider>, limits: crate::runtime::limits::Limits) -> Self {
+        Self { store, limits }
+    }
+
+    /// Validate a *name* (orchestration, activity, event, queue, tag, session ID)
+    /// against the configured name byte limit.
+    fn check_name(
+        &self,
+        name: &str,
+        kind: crate::runtime::limits::NameKind,
+    ) -> Result<(), ClientError> {
+        let len = crate::runtime::limits::measured_len(name);
+        if len > self.limits.max_name_bytes {
+            let violation = crate::runtime::limits::LimitViolation::NameTooLong {
+                kind,
+                name: name.to_string(),
+                size: len,
+                limit: self.limits.max_name_bytes,
+            };
+            return Err(ClientError::InvalidInput { message: violation.display_message() });
+        }
+        Ok(())
+    }
+
+    /// Validate an *identifier* (instance ID, KV key) against the configured
+    /// identifier byte limit.
+    fn check_identifier(
+        &self,
+        id: &str,
+        kind: crate::runtime::limits::NameKind,
+    ) -> Result<(), ClientError> {
+        let len = crate::runtime::limits::measured_len(id);
+        if len > self.limits.max_identifier_bytes {
+            let violation = crate::runtime::limits::LimitViolation::NameTooLong {
+                kind,
+                name: id.to_string(),
+                size: len,
+                limit: self.limits.max_identifier_bytes,
+            };
+            return Err(ClientError::InvalidInput { message: violation.display_message() });
+        }
+        Ok(())
     }
 
     /// Start an orchestration instance with string input.
@@ -237,9 +297,13 @@ impl Client {
         orchestration: impl Into<String>,
         input: impl Into<String>,
     ) -> Result<(), ClientError> {
+        let instance = instance.into();
+        let orchestration = orchestration.into();
+        self.check_identifier(&instance, crate::runtime::limits::NameKind::InstanceId)?;
+        self.check_name(&orchestration, crate::runtime::limits::NameKind::OrchestrationName)?;
         let item = WorkItem::StartOrchestration {
-            instance: instance.into(),
-            orchestration: orchestration.into(),
+            instance,
+            orchestration,
             input: input.into(),
             version: None,
             parent_instance: None,
@@ -264,11 +328,17 @@ impl Client {
         version: impl Into<String>,
         input: impl Into<String>,
     ) -> Result<(), ClientError> {
+        let instance = instance.into();
+        let orchestration = orchestration.into();
+        let version = version.into();
+        self.check_identifier(&instance, crate::runtime::limits::NameKind::InstanceId)?;
+        self.check_name(&orchestration, crate::runtime::limits::NameKind::OrchestrationName)?;
+        self.check_name(&version, crate::runtime::limits::NameKind::PinnedVersion)?;
         let item = WorkItem::StartOrchestration {
-            instance: instance.into(),
-            orchestration: orchestration.into(),
+            instance,
+            orchestration,
             input: input.into(),
-            version: Some(version.into()),
+            version: Some(version),
             parent_instance: None,
             parent_id: None,
             execution_id: crate::INITIAL_EXECUTION_ID,
@@ -378,9 +448,11 @@ impl Client {
         event_name: impl Into<String>,
         data: impl Into<String>,
     ) -> Result<(), ClientError> {
+        let event_name = event_name.into();
+        self.check_name(&event_name, crate::runtime::limits::NameKind::EventName)?;
         let item = WorkItem::ExternalRaised {
             instance: instance.into(),
-            name: event_name.into(),
+            name: event_name,
             data: data.into(),
         };
         self.store
@@ -423,9 +495,11 @@ impl Client {
         queue: impl Into<String>,
         data: impl Into<String>,
     ) -> Result<(), ClientError> {
+        let queue = queue.into();
+        self.check_name(&queue, crate::runtime::limits::NameKind::QueueName)?;
         let item = WorkItem::QueueMessage {
             instance: instance.into(),
-            name: queue.into(),
+            name: queue,
             data: data.into(),
         };
         self.store
diff --git a/src/lib.rs b/src/lib.rs
index 75417ffe..387dd7d6 100644
--- a/src/lib.rs
+++ b/src/lib.rs
@@ -961,6 +961,16 @@ pub enum PoisonMessageType {
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
 pub enum ConfigErrorKind {
     Nondeterminism,
+    /// A runtime size or shape limit was exceeded.
+    ///
+    /// The structured [`crate::runtime::limits::LimitViolation`] details are
+    /// serialized into the `message` field of `ErrorDetails::Configuration`
+    /// using [`LimitViolation::encode_into_message()`] so older runtimes
+    /// still surface a human-readable error string.
+    ///
+    /// First producers land in Phase 2 (name/identifier limits) and Phase 4
+    /// (error-model migration for existing limits).
+    LimitExceeded,
 }
 
 /// Application error kinds.
@@ -1008,6 +1018,10 @@ impl ErrorDetails {
                     .as_ref()
                     .map(|m| format!("nondeterministic: {m}"))
                     .unwrap_or_else(|| format!("nondeterministic in {resource}")),
+                ConfigErrorKind::LimitExceeded => message
+                    .as_ref()
+                    .map(|m| format!("limit exceeded: {m}"))
+                    .unwrap_or_else(|| format!("limit exceeded in {resource}")),
             },
             ErrorDetails::Application { kind, message, .. } => match kind {
                 AppErrorKind::Cancelled { reason } => format!("canceled: {reason}"),
diff --git a/src/runtime/dispatchers/orchestration.rs b/src/runtime/dispatchers/orchestration.rs
index 72488abc..910a0752 100644
--- a/src/runtime/dispatchers/orchestration.rs
+++ b/src/runtime/dispatchers/orchestration.rs
@@ -25,7 +25,11 @@ use super::super::{HistoryManager, Runtime, WorkItemReader};
 ///
 /// Currently enforces:
 /// - Custom status size limit ([`crate::runtime::limits::MAX_CUSTOM_STATUS_BYTES`])
-/// - Tag name size limit ([`crate::runtime::limits::MAX_TAG_NAME_BYTES`])
+/// - Tag name size limit (via [`crate::runtime::limits::Limits::max_name_bytes`])
+/// - Activity name size limit (via [`crate::runtime::limits::Limits::max_name_bytes`])
+/// - Sub-orchestration name size limit (via [`crate::runtime::limits::Limits::max_name_bytes`])
+/// - Session ID size limit (via [`crate::runtime::limits::Limits::max_name_bytes`])
+/// - KV key size limit (via [`crate::runtime::limits::Limits::max_identifier_bytes`])
 /// - KV value size limit ([`crate::runtime::limits::MAX_KV_VALUE_BYTES`])
 /// - KV key count limit ([`crate::runtime::limits::MAX_KV_KEYS`])
 ///
@@ -40,6 +44,7 @@ fn validate_limits(
     instance: &str,
     execution_id: u64,
     kv_snapshot: &std::collections::HashMap<String, crate::providers::KvEntry>,
+    limits: &crate::runtime::limits::Limits,
 ) -> bool {
     // --- Custom status size ---
     let last_custom_status = history_delta.iter().rev().find_map(|e| {
@@ -77,7 +82,7 @@ fn validate_limits(
         return true;
     }
 
-    // --- Tag name size ---
+    // --- Tag name size (rewired through limits.max_name_bytes) ---
     let oversized_tag = history_delta.iter().find_map(|e| {
         if let EventKind::ActivityScheduled {
             tag: Some(ref t),
@@ -85,8 +90,8 @@ fn validate_limits(
             ..
         } = e.kind
         {
-            if t.len() > crate::runtime::limits::MAX_TAG_NAME_BYTES {
-                Some((name.clone(), t.len()))
+            if crate::runtime::limits::measured_len(t) > limits.max_name_bytes {
+                Some((name.clone(), crate::runtime::limits::measured_len(t)))
             } else {
                 None
             }
@@ -102,16 +107,164 @@ fn validate_limits(
             execution_id = %execution_id,
             activity_name = %activity_name,
             tag_bytes = tag_len,
-            max_bytes = crate::runtime::limits::MAX_TAG_NAME_BYTES,
+            max_bytes = limits.max_name_bytes,
             "Activity tag exceeds size limit, failing orchestration"
         );
-        fail_orchestration_for_limit(
-            format!(
-                "Activity '{}' tag size ({} bytes) exceeds limit ({} bytes)",
-                activity_name,
-                tag_len,
-                crate::runtime::limits::MAX_TAG_NAME_BYTES,
-            ),
+        fail_orchestration_for_name_limit(
+            crate::runtime::limits::LimitViolation::NameTooLong {
+                kind: crate::runtime::limits::NameKind::TagName,
+                name: activity_name.clone(),
+                size: tag_len,
+                limit: limits.max_name_bytes,
+            },
+            history_mgr,
+            metadata,
+            worker_items,
+            orchestrator_items,
+            instance,
+            execution_id,
+        );
+        return true;
+    }
+
+    // --- Activity name size ---
+    let oversized_activity_name = history_delta.iter().find_map(|e| {
+        if let EventKind::ActivityScheduled { ref name, .. } = e.kind {
+            let len = crate::runtime::limits::measured_len(name);
+            if len > limits.max_name_bytes { Some((name.clone(), len)) } else { None }
+        } else {
+            None
+        }
+    });
+
+    if let Some((ref name, name_len)) = oversized_activity_name {
+        tracing::error!(
+            target: "duroxide::runtime",
+            instance_id = %instance,
+            execution_id = %execution_id,
+            activity_name = %name,
+            name_bytes = name_len,
+            max_bytes = limits.max_name_bytes,
+            "Activity name exceeds size limit, failing orchestration"
+        );
+        fail_orchestration_for_name_limit(
+            crate::runtime::limits::LimitViolation::NameTooLong {
+                kind: crate::runtime::limits::NameKind::ActivityName,
+                name: name.clone(),
+                size: name_len,
+                limit: limits.max_name_bytes,
+            },
+            history_mgr,
+            metadata,
+            worker_items,
+            orchestrator_items,
+            instance,
+            execution_id,
+        );
+        return true;
+    }
+
+    // --- Session ID size ---
+    let oversized_session_id = history_delta.iter().find_map(|e| {
+        if let EventKind::ActivityScheduled { session_id: Some(ref sid), ref name, .. } = e.kind {
+            let len = crate::runtime::limits::measured_len(sid);
+            if len > limits.max_name_bytes { Some((name.clone(), sid.clone(), len)) } else { None }
+        } else {
+            None
+        }
+    });
+
+    if let Some((ref activity_name, ref sid, sid_len)) = oversized_session_id {
+        tracing::error!(
+            target: "duroxide::runtime",
+            instance_id = %instance,
+            execution_id = %execution_id,
+            activity_name = %activity_name,
+            session_id_bytes = sid_len,
+            max_bytes = limits.max_name_bytes,
+            "Activity session ID exceeds size limit, failing orchestration"
+        );
+        fail_orchestration_for_name_limit(
+            crate::runtime::limits::LimitViolation::NameTooLong {
+                kind: crate::runtime::limits::NameKind::SessionId,
+                name: sid.clone(),
+                size: sid_len,
+                limit: limits.max_name_bytes,
+            },
+            history_mgr,
+            metadata,
+            worker_items,
+            orchestrator_items,
+            instance,
+            execution_id,
+        );
+        return true;
+    }
+
+    // --- Sub-orchestration name size ---
+    let oversized_sub_orch_name = history_delta.iter().find_map(|e| {
+        if let EventKind::SubOrchestrationScheduled { ref name, .. } = e.kind {
+            let len = crate::runtime::limits::measured_len(name);
+            if len > limits.max_name_bytes { Some((name.clone(), len)) } else { None }
+        } else {
+            None
+        }
+    });
+
+    if let Some((ref name, name_len)) = oversized_sub_orch_name {
+        tracing::error!(
+            target: "duroxide::runtime",
+            instance_id = %instance,
+            execution_id = %execution_id,
+            sub_orchestration_name = %name,
+            name_bytes = name_len,
+            max_bytes = limits.max_name_bytes,
+            "Sub-orchestration name exceeds size limit, failing orchestration"
+        );
+        fail_orchestration_for_name_limit(
+            crate::runtime::limits::LimitViolation::NameTooLong {
+                kind: crate::runtime::limits::NameKind::SubOrchestrationName,
+                name: name.clone(),
+                size: name_len,
+                limit: limits.max_name_bytes,
+            },
+            history_mgr,
+            metadata,
+            worker_items,
+            orchestrator_items,
+            instance,
+            execution_id,
+        );
+        return true;
+    }
+
+    // --- KV key size (identifier limit) ---
+    let oversized_kv_key = history_delta.iter().find_map(|e| {
+        if let EventKind::KeyValueSet { ref key, .. } = e.kind {
+            let len = crate::runtime::limits::measured_len(key);
+            if len > limits.max_identifier_bytes { Some((key.clone(), len)) } else { None }
+        } else {
+            None
+        }
+    });
+
+    if let Some((ref key, key_len)) = oversized_kv_key {
+        tracing::error!(
+            target: "duroxide::runtime",
+            instance_id = %instance,
+            execution_id = %execution_id,
+            kv_key = %key,
+            key_bytes = key_len,
+            max_bytes = limits.max_identifier_bytes,
+            "KV key exceeds size limit, failing orchestration"
+        );
+        fail_orchestration_for_name_limit(
+            crate::runtime::limits::LimitViolation::NameTooLong {
+                kind: crate::runtime::limits::NameKind::KvKey,
+                name: key.clone(),
+                size: key_len,
+                limit: limits.max_identifier_bytes,
+            },
             history_mgr,
             metadata,
             worker_items,
@@ -211,6 +364,9 @@ fn validate_limits(
 }
 
 /// Shared helper to fail an orchestration due to a limit violation.
+///
+/// Produces an `Application` error (Phase 4 will migrate to `Configuration { LimitExceeded }`).
+/// The `message` is stored verbatim in the failure event.
 fn fail_orchestration_for_limit(
     message: String,
     history_mgr: &mut HistoryManager,
@@ -242,6 +398,30 @@ fn fail_orchestration_for_limit(
     orchestrator_items.clear();
 }
 
+/// Convenience wrapper: fail an orchestration due to a structured [`LimitViolation`].
+///
+/// Encodes the violation into the message field so it is machine-parseable on
+/// retrieval while remaining human-readable to older runtimes.
+fn fail_orchestration_for_name_limit(
+    violation: crate::runtime::limits::LimitViolation,
+    history_mgr: &mut HistoryManager,
+    metadata: &mut ExecutionMetadata,
+    worker_items: &mut Vec<WorkItem>,
+    orchestrator_items: &mut Vec<WorkItem>,
+    instance: &str,
+    execution_id: u64,
+) {
+    fail_orchestration_for_limit(
+        violation.encode_into_message(),
+        history_mgr,
+        metadata,
+        worker_items,
+        orchestrator_items,
+        instance,
+        execution_id,
+    );
+}
+
 /// Error returned when orchestration processing fails before execution
 #[derive(Debug)]
 pub(crate) enum OrchestrationProcessingError {
@@ -723,6 +903,7 @@ impl Runtime {
             instance,
             execution_id_for_ack,
             &item.kv_snapshot,
+            &self.options.limits,
         );
 
         // Calculate metrics
diff --git a/src/runtime/limits.rs b/src/runtime/limits.rs
index 4faf121a..78560b9c 100644
--- a/src/runtime/limits.rs
+++ b/src/runtime/limits.rs
@@ -2,6 +2,273 @@
 //!
 //! Collect all hard limits in one place so they're easy to find, document,
 //! and reference from both runtime code and provider validators.
+//!
+//! # Two layers of limit constants
+//!
+//! - **Struct-based (`Limits`)**: runtime-configurable via `RuntimeOptions::limits`.
+//!   Use `Limits::recommended()` for the documented defaults and
+//!   `Limits::permissive()` to disable all checks (the current `Default`).
+//! - **Legacy `pub const`s**: kept for backward compatibility. Each is a
+//!   `#[deprecated]` alias pointing at the corresponding `Limits::recommended()`
+//!   value. External code that references them continues to compile unchanged.
+
+// ============================================================================
+// Runtime-configurable limits struct
+// ============================================================================
+
+/// Runtime-configurable size and shape limits.
+///
+/// Override via [`crate::runtime::RuntimeOptions::limits`] to tighten or
+/// loosen for a deployment.
+///
+/// # Phase defaults
+///
+/// [`Limits::default()`] currently returns [`Limits::permissive()`] (all
+/// checks effectively disabled). This will flip to [`Limits::recommended()`]
+/// in a future "Phase 7" release. See the size-limits proposal for details.
+#[derive(Clone, Debug)]
+pub struct Limits {
+    /// Maximum UTF-8 byte length for a *name*: orchestration name, activity
+    /// name, sub-orchestration name, event name, queue name, tag name, or
+    /// session ID.
+    ///
+    /// Default (`recommended`): 256 bytes.
+    pub max_name_bytes: usize,
+
+    /// Maximum UTF-8 byte length for an *identifier*: instance ID or KV key.
+    ///
+    /// These values end up as primary keys / index columns in provider storage
+    /// and benefit from being independently tunable.
+    ///
+    /// Default (`recommended`): 256 bytes.
+    pub max_identifier_bytes: usize,
+
+    // ---- payload/aggregate fields reserved for future phases ----
+    // (payloads: max_payload_bytes, max_message_bytes, max_diagnostic_bytes)
+    // (shape: max_fanout_per_turn, max_history_delta_events, …)
+}
+
+impl Limits {
+    /// The recommended production limits.
+    ///
+    /// These values are the documented defaults for all limit constants and
+    /// are appropriate for most deployments. `Default` will return these
+    /// values from Phase 7 onward.
+    pub fn recommended() -> Self {
+        Self {
+            max_name_bytes: 256,
+            max_identifier_bytes: 256,
+        }
+    }
+
+    /// Permissive limits — all size/shape checks effectively disabled.
+    ///
+    /// Used as [`Default`] in Phases 1–6 so no existing orchestration can
+    /// regress on upgrade. Tests that exercise specific limits should
+    /// override individual fields explicitly rather than relying on the
+    /// default.
+    pub fn permissive() -> Self {
+        Self {
+            max_name_bytes: usize::MAX,
+            max_identifier_bytes: usize::MAX,
+        }
+    }
+}
+
+impl Default for Limits {
+    /// Returns [`Limits::permissive()`].
+    ///
+    /// Will flip to [`Limits::recommended()`] in Phase 7 (a future minor
+    /// version with explicit release notes). Until then, all new limit checks
+    /// are inert unless the operator explicitly sets `RuntimeOptions::limits`.
+    fn default() -> Self {
+        Self::permissive()
+    }
+}
+
+// ============================================================================
+// Name / identifier validation types (used by Phase 2 enforcement)
+// ============================================================================
+
+/// Identifies which call site produced a name that violated a limit.
+///
+/// Carried inside [`LimitViolation::NameTooLong`] to produce per-call-site
+/// error messages without requiring a separate constant per call site.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum NameKind {
+    /// Orchestration type name (passed to `start_orchestration*` or registered
+    /// in `OrchestrationRegistry`).
+    OrchestrationName,
+    /// Activity type name (passed to `schedule_activity*` or registered in
+    /// `ActivityRegistry`).
+    ActivityName,
+    /// Sub-orchestration type name (passed to `schedule_sub_orchestration*`).
+    SubOrchestrationName,
+    /// External event name (passed to `raise_event`, `schedule_wait`).
+    EventName,
+    /// Named queue identifier (passed to `enqueue_event*`,
+    /// `dequeue_event*`).
+    QueueName,
+    /// Activity routing tag (passed via `.with_tag()`).
+    TagName,
+    /// Session identifier (passed to `schedule_activity_on_session*`).
+    SessionId,
+    /// Orchestration instance ID (passed to `start_orchestration*`,
+    /// `schedule_sub_orchestration*`).
+    InstanceId,
+    /// KV store key.
+    KvKey,
+    /// Pinned version string (passed to `start_orchestration_versioned*`).
+    PinnedVersion,
+}
+
+impl NameKind {
+    /// Human-readable label used in error messages and metrics.
+    pub fn label(&self) -> &'static str {
+        match self {
+            NameKind::OrchestrationName => "orchestration_name",
+            NameKind::ActivityName => "activity_name",
+            NameKind::SubOrchestrationName => "sub_orchestration_name",
+            NameKind::EventName => "event_name",
+            NameKind::QueueName => "queue_name",
+            NameKind::TagName => "tag_name",
+            NameKind::SessionId => "session_id",
+            NameKind::InstanceId => "instance_id",
+            NameKind::KvKey => "kv_key",
+            NameKind::PinnedVersion => "pinned_version",
+        }
+    }
+}
+
+/// A structured description of a limit violation.
+///
+/// Carried inside [`crate::ConfigErrorKind::LimitExceeded`] error details and
+/// used by registry-time panics. Serialize/deserialize support is provided so
+/// the payload can be embedded in the `message` field of
+/// `ErrorDetails::Configuration`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum LimitViolation {
+    /// A name or identifier exceeded its configured byte-length limit.
+    NameTooLong {
+        /// Which call site / field was checked.
+        kind: NameKind,
+        /// The offending name (truncated to 128 bytes for log safety).
+        name: String,
+        /// Actual UTF-8 byte length of the name.
+        size: usize,
+        /// Configured limit that was exceeded.
+        limit: usize,
+    },
+    // Future variants: InvalidName, PayloadTooLarge, TooManyEvents, …
+}
+
+impl LimitViolation {
+    /// Stable marker prefix used when embedding a `LimitViolation` inside the
+    /// `message: Option<String>` field of `ErrorDetails::Configuration`.
+    ///
+    /// Older runtimes that do not understand this format will still surface
+    /// the string as a human-readable error message.
+    pub const MESSAGE_PREFIX: &'static str = "__duroxide.limit_violation:";
+
+    /// Serialize this violation into a `message` string suitable for storing
+    /// in `ErrorDetails::Configuration { message }`.
+    pub fn encode_into_message(&self) -> String {
+        let json = match self {
+            LimitViolation::NameTooLong { kind, name, size, limit } => {
+                let safe_name = if name.len() > 128 { &name[..128] } else { name.as_str() };
+                format!(
+                    r#"{{"v":"NameTooLong","kind":"{}","name":{},"size":{},"limit":{}}}"#,
+                    kind.label(),
+                    serde_json::Value::String(safe_name.to_string()),
+                    size,
+                    limit,
+                )
+            }
+        };
+        format!("{}{}", Self::MESSAGE_PREFIX, json)
+    }
+
+    /// Try to parse a `LimitViolation` from a message string produced by
+    /// [`encode_into_message`]. Returns `None` if the string was not produced
+    /// by this method (e.g., a legacy plain-text error message).
+    pub fn parse_from_message(msg: &str) -> Option<Self> {
+        let json_str = msg.strip_prefix(Self::MESSAGE_PREFIX)?;
+        let v: serde_json::Value = serde_json::from_str(json_str).ok()?;
+        match v["v"].as_str()? {
+            "NameTooLong" => {
+                let kind_str = v["kind"].as_str()?;
+                let kind = NameKind::from_label(kind_str)?;
+                let name = v["name"].as_str()?.to_string();
+                let size = v["size"].as_u64()? as usize;
+                let limit = v["limit"].as_u64()? as usize;
+                Some(LimitViolation::NameTooLong { kind, name, size, limit })
+            }
+            _ => None,
+        }
+    }
+
+    /// Human-readable summary for error messages and log fields.
+    pub fn display_message(&self) -> String {
+        match self {
+            LimitViolation::NameTooLong { kind, name, size, limit } => {
+                let safe_name = if name.len() > 64 {
+                    format!("{}…", &name[..64])
+                } else {
+                    name.clone()
+                };
+                format!(
+                    "{} '{}' is {} bytes, exceeds limit of {} bytes",
+                    kind.label(),
+                    safe_name,
+                    size,
+                    limit,
+                )
+            }
+        }
+    }
+}
+
+impl NameKind {
+    fn from_label(label: &str) -> Option<Self> {
+        match label {
+            "orchestration_name" => Some(NameKind::OrchestrationName),
+            "activity_name" => Some(NameKind::ActivityName),
+            "sub_orchestration_name" => Some(NameKind::SubOrchestrationName),
+            "event_name" => Some(NameKind::EventName),
+            "queue_name" => Some(NameKind::QueueName),
+            "tag_name" => Some(NameKind::TagName),
+            "session_id" => Some(NameKind::SessionId),
+            "instance_id" => Some(NameKind::InstanceId),
+            "kv_key" => Some(NameKind::KvKey),
+            "pinned_version" => Some(NameKind::PinnedVersion),
+            _ => None,
+        }
+    }
+}
+
+// ============================================================================
+// Byte-counting helper
+// ============================================================================
+
+/// Measure the byte length of a string for limit checks.
+///
+/// Returns the raw UTF-8 byte length (`str::len()`). All limit comparisons go
+/// through this helper so the measurement strategy is easy to change in one
+/// place.
+///
+/// Note: This measures the *raw* string, not the JSON-escaped form. The
+/// catalog defaults include ~2× headroom over typical provider envelopes to
+/// absorb JSON escaping overhead.
+#[inline]
+pub fn measured_len(s: &str) -> usize {
+    s.len()
+}
+
+// ============================================================================
+// Legacy `pub const` aliases (kept for backward compatibility)
+// ============================================================================
 
 /// Maximum number of unmatched persistent events that can be carried forward
 /// across a `continue_as_new()` boundary.
@@ -28,9 +295,12 @@ pub const MAX_WORKER_TAGS: usize = 5;
 
 /// Maximum size in bytes for a single activity tag name.
 ///
-/// Enforced at the orchestration dispatcher level (before ack) following
-/// the same pattern as [`MAX_CUSTOM_STATUS_BYTES`]. If exceeded, the
-/// orchestration is failed with an Infrastructure error.
+/// # Deprecation
+///
+/// This constant is a legacy alias for [`Limits::recommended().max_name_bytes`](Limits::recommended).
+/// Prefer accessing the limit through [`RuntimeOptions::limits`](crate::runtime::RuntimeOptions::limits)
+/// at runtime, or through [`Limits::recommended()`] for static defaults.
+#[deprecated(since = "0.1.29", note = "use Limits::recommended().max_name_bytes or RuntimeOptions::limits.max_name_bytes")]
 pub const MAX_TAG_NAME_BYTES: usize = 256;
 
 /// Maximum number of **user** KV keys per orchestration instance.
@@ -43,3 +313,85 @@ pub const MAX_KV_KEYS: usize = 150;
 ///
 /// Enforced in `validate_limits()` by scanning `KeyValueSet` events in the history delta.
 pub const MAX_KV_VALUE_BYTES: usize = 64 * 1024;
+
+// ============================================================================
+// Unit tests
+// ============================================================================
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn permissive_is_effectively_unlimited() {
+        let l = Limits::permissive();
+        assert_eq!(l.max_name_bytes, usize::MAX);
+        assert_eq!(l.max_identifier_bytes, usize::MAX);
+    }
+
+    #[test]
+    fn recommended_has_256_byte_limits() {
+        let l = Limits::recommended();
+        assert_eq!(l.max_name_bytes, 256);
+        assert_eq!(l.max_identifier_bytes, 256);
+    }
+
+    #[test]
+    fn default_is_permissive() {
+        let l = Limits::default();
+        assert_eq!(l.max_name_bytes, usize::MAX);
+    }
+
+    #[test]
+    fn measured_len_counts_utf8_bytes_not_chars() {
+        // Each emoji is 4 UTF-8 bytes
+        let emoji = "🦀".repeat(64); // 64 * 4 = 256 bytes, 64 chars
+        assert_eq!(measured_len(&emoji), 256);
+        assert_eq!(emoji.chars().count(), 64);
+
+        let one_more = "🦀".repeat(65); // 260 bytes
+        assert_eq!(measured_len(&one_more), 260);
+    }
+
+    #[test]
+    fn limit_violation_encode_decode_roundtrip() {
+        let v = LimitViolation::NameTooLong {
+            kind: NameKind::ActivityName,
+            name: "my_activity".to_string(),
+            size: 512,
+            limit: 256,
+        };
+        let encoded = v.encode_into_message();
+        assert!(encoded.starts_with(LimitViolation::MESSAGE_PREFIX));
+        let decoded = LimitViolation::parse_from_message(&encoded).expect("should decode");
+        assert_eq!(decoded, v);
+    }
+
+    #[test]
+    fn limit_violation_display_truncates_long_names() {
+        let long_name = "a".repeat(200);
+        let v = LimitViolation::NameTooLong {
+            kind: NameKind::InstanceId,
+            name: long_name,
+            size: 200,
+            limit: 256,
+        };
+        let msg = v.display_message();
+        // Should not panic, and should contain the kind label
+        assert!(msg.contains("instance_id"));
+    }
+
+    #[test]
+    fn parse_from_message_ignores_non_prefixed_strings() {
+        assert!(LimitViolation::parse_from_message("plain error message").is_none());
+        assert!(LimitViolation::parse_from_message("").is_none());
+    }
+
+    #[test]
+    fn name_kind_labels_are_stable() {
+        assert_eq!(NameKind::OrchestrationName.label(), "orchestration_name");
+        assert_eq!(NameKind::ActivityName.label(), "activity_name");
+        assert_eq!(NameKind::InstanceId.label(), "instance_id");
+        assert_eq!(NameKind::KvKey.label(), "kv_key");
+    }
+}
diff --git a/src/runtime/mod.rs b/src/runtime/mod.rs
index fef0b093..93efc542 100644
--- a/src/runtime/mod.rs
+++ b/src/runtime/mod.rs
@@ -342,6 +342,16 @@ pub struct RuntimeOptions {
     ///
     /// Default: `TagFilter::DefaultOnly` (untagged activities only)
     pub worker_tag_filter: crate::providers::TagFilter,
+
+    /// Runtime-configurable size and shape limits.
+    ///
+    /// Override individual fields to tighten limits for a deployment.
+    /// The default ([`limits::Limits::permissive()`]) disables all new
+    /// limit checks so existing orchestrations are unaffected on upgrade.
+    ///
+    /// Set to [`limits::Limits::recommended()`] to enable the documented
+    /// production defaults.
+    pub limits: limits::Limits,
 }
 
 impl Default for RuntimeOptions {
@@ -367,6 +377,7 @@ impl Default for RuntimeOptions {
             max_sessions_per_runtime: 10,
             worker_node_id: None,
             worker_tag_filter: crate::providers::TagFilter::default(),
+            limits: limits::Limits::default(),
         }
     }
 }
diff --git a/src/runtime/registry.rs b/src/runtime/registry.rs
index 94ff1603..8811c4e0 100644
--- a/src/runtime/registry.rs
+++ b/src/runtime/registry.rs
@@ -278,6 +278,27 @@ impl<H: ?Sized> RegistryBuilder<H> {
     }
 }
 
+/// Validate a handler name against [`crate::runtime::limits::Limits::recommended()`].
+///
+/// Panics if the name exceeds the recommended byte limit, matching the existing
+/// behavior for duplicate registrations (which also panic). Uses static defaults
+/// rather than `RuntimeOptions::limits` because registry builders have no access
+/// to `RuntimeOptions` at registration time; the authoritative runtime check runs
+/// separately in `validate_limits()`.
+fn check_registry_name_limit(name: &str, kind: crate::runtime::limits::NameKind) {
+    let recommended = crate::runtime::limits::Limits::recommended();
+    let len = crate::runtime::limits::measured_len(name);
+    if len > recommended.max_name_bytes {
+        let violation = crate::runtime::limits::LimitViolation::NameTooLong {
+            kind,
+            name: name.to_string(),
+            size: len,
+            limit: recommended.max_name_bytes,
+        };
+        panic!("handler registration rejected: {}", violation.display_message());
+    }
+}
+
 // ============================================================================
 // Orchestration Builder - Specialized Methods
 // ============================================================================
@@ -289,6 +310,7 @@ impl OrchestrationRegistryBuilder {
         Fut: std::future::Future<Output = Result<String, String>> + Send + 'static,
     {
         let name = name.into();
+        check_registry_name_limit(&name, crate::runtime::limits::NameKind::OrchestrationName);
         if self.check_duplicate(&name, &DEFAULT_VERSION, "orchestration") {
             return self;
         }
@@ -317,6 +339,7 @@ impl OrchestrationRegistryBuilder {
             }
         };
         let name = name.into();
+        check_registry_name_limit(&name, crate::runtime::limits::NameKind::OrchestrationName);
         self.map
             .entry(name)
             .or_default()
@@ -330,6 +353,7 @@ impl OrchestrationRegistryBuilder {
         Fut: std::future::Future<Output = Result<String, String>> + Send + 'static,
     {
         let name = name.into();
+        check_registry_name_limit(&name, crate::runtime::limits::NameKind::OrchestrationName);
         // Version parsing should never fail for valid semver strings from registry
         let v = Version::parse(version.as_ref()).expect("Version should be valid semver");
         if self.check_duplicate(&name, &v, "orchestration") {
@@ -359,6 +383,7 @@ impl OrchestrationRegistryBuilder {
     {
         use super::FnOrchestration;
         let name = name.into();
+        check_registry_name_limit(&name, crate::runtime::limits::NameKind::OrchestrationName);
         // Version parsing should never fail for valid semver strings from registry
         let v = Version::parse(version.as_ref()).expect("Version should be valid semver");
         if self.check_duplicate(&name, &v, "orchestration") {
@@ -438,6 +463,7 @@ impl ActivityRegistryBuilder {
         if self.check_reserved_prefix(&name) {
             return self;
         }
+        check_registry_name_limit(&name, crate::runtime::limits::NameKind::ActivityName);
         if self.check_duplicate(&name, &DEFAULT_VERSION, "activity") {
             return self;
         }
@@ -471,6 +497,7 @@ impl ActivityRegistryBuilder {
         if self.check_reserved_prefix(&name) {
             return self;
         }
+        check_registry_name_limit(&name, crate::runtime::limits::NameKind::ActivityName);
         if self.check_duplicate(&name, &DEFAULT_VERSION, "activity") {
             return self;
         }
diff --git a/tests/scenarios.rs b/tests/scenarios.rs
index 86437cea..447329b9 100644
--- a/tests/scenarios.rs
+++ b/tests/scenarios.rs
@@ -12,6 +12,7 @@
 //! - Single-thread runtime mode (for embedded hosts)
 //! - Rolling deployment scenarios (multi-node with version upgrades)
 //! - Unobserved future cancellation (select losers, dropped futures)
+//! - Name and identifier size limits (Phase 2)
 #![allow(clippy::unwrap_used)]
 #![allow(clippy::clone_on_ref_ptr)]
 #![allow(clippy::expect_used)]
@@ -39,3 +40,6 @@ mod replay_versioning;
 
 #[path = "scenarios/copilot_chat.rs"]
 mod copilot_chat;
+
+#[path = "scenarios/limits.rs"]
+mod limits;
diff --git a/tests/scenarios/limits.rs b/tests/scenarios/limits.rs
new file mode 100644
index 00000000..5e925023
--- /dev/null
+++ b/tests/scenarios/limits.rs
@@ -0,0 +1,419 @@
+//! Scenario tests for name and identifier size limits (Phase 2).
+//!
+//! These tests validate that `RuntimeOptions::limits` is respected at both the
+//! Tier-1 (client-side) and Tier-2 (`validate_limits()`) enforcement points.
+//!
+//! Defaults remain permissive (`Limits::permissive()`), so every test that
+//! exercises a tightened limit must create its own runtime/client with
+//! explicit `RuntimeOptions::limits` overrides.
+
+#![allow(clippy::unwrap_used)]
+#![allow(clippy::clone_on_ref_ptr)]
+#![allow(clippy::expect_used)]
+
+#[path = "../common/mod.rs"]
+mod common;
+
+use duroxide::runtime::limits::{LimitViolation, Limits, NameKind, measured_len};
+use duroxide::runtime::{RuntimeOptions, registry::ActivityRegistry};
+use duroxide::{ActivityContext, Client, OrchestrationContext, OrchestrationRegistry};
+use std::time::Duration;
+
+// ============================================================================
+// Unit-level tests (no runtime needed)
+// ============================================================================
+
+#[test]
+fn measured_len_byte_length_not_char_length() {
+    // Each crab emoji is 4 UTF-8 bytes
+    let emoji_64 = "🦀".repeat(64); // 256 bytes, 64 chars
+    assert_eq!(measured_len(&emoji_64), 256);
+    assert_eq!(emoji_64.chars().count(), 64);
+
+    let emoji_65 = "🦀".repeat(65); // 260 bytes, 65 chars
+    assert_eq!(measured_len(&emoji_65), 260);
+}
+
+#[test]
+fn limits_permissive_allows_everything() {
+    let l = Limits::permissive();
+    assert_eq!(l.max_name_bytes, usize::MAX);
+    assert_eq!(l.max_identifier_bytes, usize::MAX);
+}
+
+#[test]
+fn limits_recommended_has_256_byte_values() {
+    let l = Limits::recommended();
+    assert_eq!(l.max_name_bytes, 256);
+    assert_eq!(l.max_identifier_bytes, 256);
+}
+
+#[test]
+fn limits_default_is_permissive() {
+    // Phases 1–6: default must be permissive to avoid regressions.
+    let l = Limits::default();
+    assert_eq!(l.max_name_bytes, usize::MAX);
+}
+
+#[test]
+fn runtime_options_includes_limits_field() {
+    // Smoke-test that RuntimeOptions::limits is accessible and defaults to permissive.
+    let opts = RuntimeOptions::default();
+    assert_eq!(opts.limits.max_name_bytes, usize::MAX);
+}
+
+// ============================================================================
+// LimitViolation encode/decode round-trip
+// ============================================================================
+
+#[test]
+fn limit_violation_roundtrip_activity_name() {
+    let v = LimitViolation::NameTooLong {
+        kind: NameKind::ActivityName,
+        name: "my_activity".to_string(),
+        size: 512,
+        limit: 256,
+    };
+    let encoded = v.encode_into_message();
+    assert!(encoded.starts_with(LimitViolation::MESSAGE_PREFIX));
+    let decoded = LimitViolation::parse_from_message(&encoded).expect("should round-trip");
+    assert_eq!(decoded, v);
+}
+
+#[test]
+fn limit_violation_roundtrip_instance_id() {
+    let v = LimitViolation::NameTooLong {
+        kind: NameKind::InstanceId,
+        name: "very-long-instance-id".to_string(),
+        size: 300,
+        limit: 256,
+    };
+    let encoded = v.encode_into_message();
+    let decoded = LimitViolation::parse_from_message(&encoded).expect("should round-trip");
+    assert_eq!(decoded, v);
+}
+
+#[test]
+fn limit_violation_parse_ignores_plain_strings() {
+    assert!(LimitViolation::parse_from_message("plain error").is_none());
+    assert!(LimitViolation::parse_from_message("").is_none());
+    // Wrong prefix
+    assert!(LimitViolation::parse_from_message("__other_prefix:{\"v\":\"NameTooLong\"}").is_none());
+}
+
+#[test]
+fn limit_violation_display_message_is_human_readable() {
+    let v = LimitViolation::NameTooLong {
+        kind: NameKind::OrchestrationName,
+        name: "short".to_string(),
+        size: 300,
+        limit: 256,
+    };
+    let msg = v.display_message();
+    assert!(msg.contains("orchestration_name"));
+    assert!(msg.contains("300"));
+    assert!(msg.contains("256"));
+}
+
+// ============================================================================
+// Tier-1 client checks (with tightened limits)
+// ============================================================================
+
+#[tokio::test]
+async fn client_rejects_oversized_instance_id() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+    let client = Client::new_with_limits(store, Limits::recommended());
+
+    let oversized_instance = "i".repeat(Limits::recommended().max_identifier_bytes + 1);
+    let result = client
+        .start_orchestration(&oversized_instance, "MyOrch", "input")
+        .await;
+
+    assert!(result.is_err(), "should reject oversized instance ID");
+    let err = result.unwrap_err();
+    assert!(
+        err.to_string().contains("instance_id"),
+        "error should mention instance_id: {err}"
+    );
+}
+
+#[tokio::test]
+async fn client_accepts_max_size_instance_id() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+    let client = Client::new_with_limits(store, Limits::recommended());
+
+    let max_instance = "i".repeat(Limits::recommended().max_identifier_bytes);
+    // Should not fail with InvalidInput (may fail for other reasons like no running runtime)
+    let result = client
+        .start_orchestration(&max_instance, "MyOrch", "input")
+        .await;
+
+    // The limit check itself should pass (no InvalidInput error)
+    match result {
+        Err(duroxide::ClientError::InvalidInput { .. }) => {
+            panic!("should not reject max-size instance ID: {result:?}")
+        }
+        _ => {} // Provider or other error is fine — we only care about no limit rejection
+    }
+}
+
+#[tokio::test]
+async fn client_rejects_oversized_orchestration_name() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+    let client = Client::new_with_limits(store, Limits::recommended());
+
+    let oversized_name = "n".repeat(Limits::recommended().max_name_bytes + 1);
+    let result = client
+        .start_orchestration("my-instance", &oversized_name, "input")
+        .await;
+
+    assert!(result.is_err());
+    assert!(result.unwrap_err().to_string().contains("orchestration_name"));
+}
+
+#[tokio::test]
+async fn client_rejects_oversized_event_name() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+    let client = Client::new_with_limits(store, Limits::recommended());
+
+    let oversized_name = "e".repeat(Limits::recommended().max_name_bytes + 1);
+    let result = client
+        .raise_event("my-instance", &oversized_name, "data")
+        .await;
+
+    assert!(result.is_err());
+    assert!(result.unwrap_err().to_string().contains("event_name"));
+}
+
+#[tokio::test]
+async fn client_rejects_oversized_queue_name() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+    let client = Client::new_with_limits(store, Limits::recommended());
+
+    let oversized_name = "q".repeat(Limits::recommended().max_name_bytes + 1);
+    let result = client
+        .enqueue_event("my-instance", &oversized_name, "data")
+        .await;
+
+    assert!(result.is_err());
+    assert!(result.unwrap_err().to_string().contains("queue_name"));
+}
+
+#[tokio::test]
+async fn client_rejects_oversized_pinned_version() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+    let client = Client::new_with_limits(store, Limits::recommended());
+
+    let oversized_version = "v".repeat(Limits::recommended().max_name_bytes + 1);
+    let result = client
+        .start_orchestration_versioned("my-instance", "MyOrch", &oversized_version, "input")
+        .await;
+
+    assert!(result.is_err());
+    assert!(result.unwrap_err().to_string().contains("pinned_version"));
+}
+
+#[tokio::test]
+async fn client_with_default_limits_allows_long_names() {
+    // Default (permissive) limits must not reject anything.
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+    let client = Client::new(store); // default = permissive
+
+    let very_long = "x".repeat(1024);
+    let result = client
+        .start_orchestration(&very_long, &very_long, "input")
+        .await;
+
+    // Should not get InvalidInput — may get a provider error but not a limit rejection.
+    match result {
+        Err(duroxide::ClientError::InvalidInput { .. }) => {
+            panic!("permissive client should not reject long names")
+        }
+        _ => {}
+    }
+}
+
+// ============================================================================
+// Tier-2 validate_limits() checks (requires running runtime)
+// ============================================================================
+
+#[tokio::test]
+async fn tier2_oversized_activity_name_fails_orchestration() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+
+    let oversized_name = "a".repeat(Limits::recommended().max_name_bytes + 1);
+    let orch_name = oversized_name.clone();
+
+    let activity_registry = ActivityRegistry::builder()
+        .register("dummy", |_ctx: ActivityContext, input: String| async move { Ok(input) })
+        .build();
+
+    let orchestration = move |ctx: OrchestrationContext, _input: String| {
+        let name = orch_name.clone();
+        async move {
+            // schedule_activity with an oversized name; limit check fires in validate_limits()
+            let result = ctx.schedule_activity(&name, "").await;
+            Ok(result.unwrap_or_else(|e| e))
+        }
+    };
+
+    let orchestration_registry = OrchestrationRegistry::builder()
+        .register("TestOrch", orchestration)
+        .build();
+
+    let rt = duroxide::runtime::Runtime::start_with_options(
+        store.clone(),
+        activity_registry,
+        orchestration_registry,
+        RuntimeOptions {
+            orchestration_concurrency: 1,
+            worker_concurrency: 1,
+            limits: Limits::recommended(),
+            ..RuntimeOptions::default()
+        },
+    )
+    .await;
+
+    let client = Client::new(store.clone());
+    client.start_orchestration("inst-act-name", "TestOrch", "").await.unwrap();
+
+    // Wait for orchestration to finish (it should fail due to the limit)
+    let result = client
+        .wait_for_orchestration("inst-act-name", Duration::from_secs(5))
+        .await;
+
+    rt.shutdown(None).await;
+
+    match result {
+        Ok(duroxide::OrchestrationStatus::Failed { details, .. }) => {
+            let msg = details.display_message();
+            assert!(
+                msg.contains("activity_name") || msg.contains("limit exceeded"),
+                "failure message should mention activity_name or limit exceeded: {msg}"
+            );
+        }
+        other => panic!("expected Failed, got: {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn tier2_oversized_sub_orchestration_name_fails_orchestration() {
+    let (store, _dir) = common::create_sqlite_store_disk().await;
+
+    let oversized_name = "s".repeat(Limits::recommended().max_name_bytes + 1);
+    let orch_name = oversized_name.clone();
+
+    let activity_registry = ActivityRegistry::builder().build();
+
+    let parent_fn = move |ctx: OrchestrationContext, _input: String| {
+        let name = orch_name.clone();
+        async move {
+            // This tries to schedule a sub-orch with an oversized name
+            let result = ctx.schedule_sub_orchestration(&name, "").await;
+            Ok(result.unwrap_or_else(|e| e))
+        }
+    };
+
+    // We can't register a handler with oversized name (registry check would panic),
+    // so we just attempt to schedule a non-existent oversized-name sub-orchestration
+    // and verify that validate_limits() fires before the "not found" path.
+    let orchestration_registry = OrchestrationRegistry::builder()
+        .register("ParentOrch", parent_fn)
+        .build();
+
+    let rt = duroxide::runtime::Runtime::start_with_options(
+        store.clone(),
+        activity_registry,
+        orchestration_registry,
+        RuntimeOptions {
+            orchestration_concurrency: 1,
+            worker_concurrency: 1,
+            limits: Limits::recommended(),
+            ..RuntimeOptions::default()
+        },
+    )
+    .await;
+
+    let client = Client::new(store.clone());
+    client
+        .start_orchestration("inst-sub-name", "ParentOrch", "")
+        .await
+        .unwrap();
+
+    let result = client
+        .wait_for_orchestration("inst-sub-name", Duration::from_secs(5))
+        .await;
+
+    rt.shutdown(None).await;
+
+    match result {
+        Ok(duroxide::OrchestrationStatus::Failed { details, .. }) => {
+            let msg = details.display_message();
+            assert!(
+                msg.contains("sub_orchestration_name") || msg.contains("limit exceeded"),
+                "failure message should mention sub_orchestration_name or limit exceeded: {msg}"
+            );
+        }
+        other => panic!("expected Failed, got: {other:?}"),
+    }
+}
+
+// ============================================================================
+// Registry-time name limit checks
+// ============================================================================
+
+#[test]
+#[should_panic(expected = "handler registration rejected")]
+fn registry_panics_on_oversized_orchestration_name() {
+    let oversized = "o".repeat(Limits::recommended().max_name_bytes + 1);
+    let _reg = OrchestrationRegistry::builder()
+        .register(&oversized, |_ctx: OrchestrationContext, _: String| async move { Ok("done".to_string()) })
+        .build();
+}
+
+#[test]
+#[should_panic(expected = "handler registration rejected")]
+fn registry_panics_on_oversized_activity_name() {
+    let oversized = "a".repeat(Limits::recommended().max_name_bytes + 1);
+    let _reg = ActivityRegistry::builder()
+        .register(&oversized, |_ctx: ActivityContext, input: String| async move { Ok(input) })
+        .build();
+}
+
+#[test]
+fn registry_accepts_max_size_name() {
+    let max = "m".repeat(Limits::recommended().max_name_bytes);
+
+    // Should not panic for exactly 256 bytes
+    let _reg = OrchestrationRegistry::builder()
+        .register(&max, |_ctx: OrchestrationContext, _: String| async move { Ok("done".to_string()) })
+        .build();
+
+    let _reg2 = ActivityRegistry::builder()
+        .register(&max, |_ctx: ActivityContext, input: String| async move { Ok(input) })
+        .build();
+}
+
+#[test]
+fn registry_accepts_multibyte_utf8_name_at_limit() {
+    // 64 emoji × 4 bytes each = 256 bytes = exactly recommended max_name_bytes
+    let emoji_name = "🦀".repeat(64);
+    assert_eq!(measured_len(&emoji_name), 256);
+
+    // Should not panic
+    let _reg = ActivityRegistry::builder()
+        .register(&emoji_name, |_ctx: ActivityContext, input: String| async move { Ok(input) })
+        .build();
+}
+
+#[test]
+#[should_panic(expected = "handler registration rejected")]
+fn registry_rejects_multibyte_utf8_name_over_limit() {
+    // 65 emoji × 4 bytes = 260 bytes > recommended max_name_bytes (256)
+    let emoji_name = "🦀".repeat(65);
+    assert_eq!(measured_len(&emoji_name), 260);
+
+    let _reg = ActivityRegistry::builder()
+        .register(&emoji_name, |_ctx: ActivityContext, input: String| async move { Ok(input) })
+        .build();
+}
diff --git a/tests/tag_serde_tests.rs b/tests/tag_serde_tests.rs
index 236bdcd1..00f2a682 100644
--- a/tests/tag_serde_tests.rs
+++ b/tests/tag_serde_tests.rs
@@ -320,7 +320,10 @@ async fn e2e_tagged_activity_starvation_times_out() {
     rt.shutdown(None).await;
 }
 
-/// An activity tag exceeding MAX_TAG_NAME_BYTES (256) fails the orchestration.
+/// An activity tag exceeding the configured `max_name_bytes` limit fails the orchestration.
+///
+/// This test explicitly sets `RuntimeOptions::limits` to `Limits::recommended()` because
+/// the default is now `Limits::permissive()` (no checks). See the size-limits proposal.
 #[tokio::test]
 async fn e2e_oversized_tag_fails_orchestration() {
     use duroxide::runtime::{self, RuntimeOptions, limits, registry::ActivityRegistry};
@@ -335,7 +338,7 @@ async fn e2e_oversized_tag_fails_orchestration() {
         })
         .build();
 
-    let oversized_tag = "x".repeat(limits::MAX_TAG_NAME_BYTES + 1);
+    let oversized_tag = "x".repeat(limits::Limits::recommended().max_name_bytes + 1);
     let orchestration = move |ctx: OrchestrationContext, _input: String| {
         let tag = oversized_tag.clone();
         async move {
@@ -350,6 +353,7 @@ async fn e2e_oversized_tag_fails_orchestration() {
 
     let opts = RuntimeOptions {
         worker_tag_filter: TagFilter::Any,
+        limits: limits::Limits::recommended(),
         ..Default::default()
     };
 
@@ -369,8 +373,8 @@ async fn e2e_oversized_tag_fails_orchestration() {
         runtime::OrchestrationStatus::Failed { details, .. } => {
             let msg = details.display_message();
             assert!(
-                msg.contains("tag size") && msg.contains("exceeds limit"),
-                "Expected tag size limit error, got: {msg}"
+                msg.contains("tag_name"),
+                "Expected tag_name limit error, got: {msg}"
             );
         }
         runtime::OrchestrationStatus::Completed { output, .. } => {
@@ -382,7 +386,7 @@ async fn e2e_oversized_tag_fails_orchestration() {
     rt.shutdown(None).await;
 }
 
-/// A tag at exactly MAX_TAG_NAME_BYTES (256 bytes) succeeds.
+/// A tag at exactly `Limits::recommended().max_name_bytes` (256 bytes) succeeds.
 #[tokio::test]
 async fn e2e_tag_at_boundary_succeeds() {
     use duroxide::runtime::{self, RuntimeOptions, limits, registry::ActivityRegistry};
@@ -397,7 +401,7 @@ async fn e2e_tag_at_boundary_succeeds() {
         })
         .build();
 
-    let boundary_tag = "x".repeat(limits::MAX_TAG_NAME_BYTES);
+    let boundary_tag = "x".repeat(limits::Limits::recommended().max_name_bytes);
     let orchestration = move |ctx: OrchestrationContext, _input: String| {
         let tag = boundary_tag.clone();
         async move {
@@ -412,6 +416,7 @@ async fn e2e_tag_at_boundary_succeeds() {
 
     let opts = RuntimeOptions {
         worker_tag_filter: TagFilter::Any,
+        limits: limits::Limits::recommended(),
         ..Default::default()
     };
 

From 0535bb12a7c58024036934d1c1d85e7a7a466cc2 Mon Sep 17 00:00:00 2001
From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com>
Date: Wed, 29 Apr 2026 21:47:34 +0000
Subject: [PATCH 3/3] Fix UTF-8 char boundary slicing in LimitViolation;
 clarify Tier-2 test intent

Agent-Logs-Url: https://github.com/microsoft/duroxide/sessions/e00db919-62c8-4245-a786-c4e35f8d5fcc

Co-authored-by: pinodeca <32303022+pinodeca@users.noreply.github.com>
---
 src/runtime/limits.rs     | 20 ++++++++++++++++++--
 tests/scenarios/limits.rs |  7 +++++++
 2 files changed, 25 insertions(+), 2 deletions(-)

diff --git a/src/runtime/limits.rs b/src/runtime/limits.rs
index 78560b9c..d5fdc278 100644
--- a/src/runtime/limits.rs
+++ b/src/runtime/limits.rs
@@ -177,7 +177,7 @@ impl LimitViolation {
     pub fn encode_into_message(&self) -> String {
         let json = match self {
             LimitViolation::NameTooLong { kind, name, size, limit } => {
-                let safe_name = if name.len() > 128 { &name[..128] } else { name.as_str() };
+                let safe_name = truncate_at_char_boundary(name, 128);
                 format!(
                     r#"{{"v":"NameTooLong","kind":"{}","name":{},"size":{},"limit":{}}}"#,
                     kind.label(),
@@ -214,7 +214,7 @@ impl LimitViolation {
         match self {
             LimitViolation::NameTooLong { kind, name, size, limit } => {
                 let safe_name = if name.len() > 64 {
-                    format!("{}…", &name[..64])
+                    format!("{}…", truncate_at_char_boundary(name, 64))
                 } else {
                     name.clone()
                 };
@@ -266,6 +266,22 @@ pub fn measured_len(s: &str) -> usize {
     s.len()
 }
 
+/// Truncate a string to at most `max_bytes` UTF-8 bytes, ensuring the cut
+/// falls on a character boundary.
+///
+/// Returns a `&str` slice of the original string — always valid UTF-8.
+fn truncate_at_char_boundary(s: &str, max_bytes: usize) -> &str {
+    if s.len() <= max_bytes {
+        return s;
+    }
+    // Walk backward from max_bytes until we find a valid character boundary.
+    let mut boundary = max_bytes;
+    while boundary > 0 && !s.is_char_boundary(boundary) {
+        boundary -= 1;
+    }
+    &s[..boundary]
+}
+
 // ============================================================================
 // Legacy `pub const` aliases (kept for backward compatibility)
 // ============================================================================
diff --git a/tests/scenarios/limits.rs b/tests/scenarios/limits.rs
index 5e925023..28da8ce8 100644
--- a/tests/scenarios/limits.rs
+++ b/tests/scenarios/limits.rs
@@ -274,6 +274,10 @@ async fn tier2_oversized_activity_name_fails_orchestration() {
     )
     .await;
 
+    // The client uses permissive limits (default) here intentionally — we want to test
+    // that Tier-2 (runtime validate_limits) catches the oversized activity name even when
+    // Tier-1 (client-side) would have allowed it through. The orchestration name itself
+    // ("TestOrch", 8 bytes) is within limits, so start_orchestration succeeds.
     let client = Client::new(store.clone());
     client.start_orchestration("inst-act-name", "TestOrch", "").await.unwrap();
 
@@ -334,6 +338,9 @@ async fn tier2_oversized_sub_orchestration_name_fails_orchestration() {
     )
     .await;
 
+    // The client uses permissive limits (default) intentionally — we test that Tier-2
+    // (runtime validate_limits) catches the oversized sub-orchestration name even when
+    // Tier-1 would allow it through. The orchestration name "ParentOrch" is within limits.
     let client = Client::new(store.clone());
     client
         .start_orchestration("inst-sub-name", "ParentOrch", "")