@@ -814,6 +814,16 @@ is general semantic triage, not a new steward-specific classifier or scheduler.
 Typed admission owns provider-independent trigger reasons; the App renders those
 facts without treating a configuration change as retroactive execution.
 
+Native steward private admission now supplies the existing inbox's exact
+provider provenance before Turn launch, including prepared-request replay.
+It follows the resolved coordination owner when Chat storage is separate, with
+the retained layout for legacy controllers; source evidence does not fork into
+the private Chat store.
+Operator read/delivery configuration recognizes its full App/source channel;
+portfolio read alone still grants no delivery. Source/policy/replay fixtures
+qualify this entry prerequisite, not worker launch, adoption, automatic wake or
+original-route completion. Those remain the same R3 acceptance gates above.
+
 ### R4: Shared Goal Alignment and Evolution
 
 - **Owner:** alignment RFC Stage 3–5 and TS Goal/work-graph owners.
 
@@ -53,6 +53,21 @@ external delivery, including future replay. Existing inbox records are retained.
 Provider ingress receipts bind message digest, channel and sender; a model cannot
 create or widen that provenance through its response.
 
+Native steward private conversations record that provenance before the canonical
+Turn can start, using this App's independently verified opaque `operator_ref`.
+Use the exact Session channel `manager.external.native.<binding-ref>.<source-ref>`
+and that owner reference in the private source policy. The same read-scope and
+delivery-target commands accept this exact channel alongside legacy channels;
+partial channels and wildcards are rejected. Native admission and crash replay
+retain the original Session, request and message. An App's portfolio read grant
+alone does not grant context delivery, and inbox delivery does not launch a worker.
+Provenance uses the controller's resolved coordination runtime root when present,
+so a separate Chat store cannot split it from the recipient policy and inbox.
+Controllers predating that root retain the existing Chat-parent layout.
+Messages outside the shared inbox's source-context bounds remain intact in
+ordinary Chat, but cannot be handed off as inline context; use a scoped artifact
+for larger material. Image content is not transferred by this text ingress record.
+
 The existing worker turn-start hook exposes only a bounded pending count and
 required read command, without copying private content into status projections.
 Read and record a decision through the installed CLI:
 
@@ -324,11 +324,17 @@ def evidence_goal_scope(runtime_root: Path, channel: str) -> list[str] | None:
         return []
 
 
+def _require_external_channel(channel: str) -> None:
+    # Legacy manager connections and native App/source bindings both identify
+    # one exact audience. Never accept a prefix or a partially specified binding.
+    if not re.fullmatch(r"manager\.external\.(?:[a-f0-9]{24}|native\.[a-f0-9]{24}\.[a-f0-9]{24})", channel):
+        raise ValueError("an exact external manager channel is required")
+
+
 def configure_evidence_scope(runtime_root: Path, registry_path: Path, *, channel: str,
                              goal_ids: list[str], execute: bool = False) -> dict:
     """Local operator grants only selected Goal summaries to an exact audience."""
-    if not re.fullmatch(r"manager\.external\.[a-f0-9]{24}", channel):
-        raise ValueError("an exact external manager channel is required")
+    _require_external_channel(channel)
     registry = load_project_registry(registry_path)
     available = {g.get("id") for g in registry.get("goals", []) if isinstance(g, dict)}
     if any(g not in available for g in goal_ids):
@@ -361,8 +367,7 @@ def configure_delivery_target(
     execute: bool = False,
 ) -> dict:
     """Observe registry/policy and persist a typed sender-bound recipient change."""
-    if not re.fullmatch(r"manager\.external\.[a-f0-9]{24}", channel):
-        raise ValueError("an exact external manager channel is required")
+    _require_external_channel(channel)
     path = _root(runtime_root) / "policy.json"
 
     def update(*, apply: bool) -> dict:
 
@@ -94,6 +94,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
             if controller.store.turn_for_client(current["session_id"], client_id) is not None:
                 # The canonical store validates exact replay before its closed
                 # Session check. Never move an accepted request to a new Session.
+                self._record_steward_ingress(row, selected, current, client_id)
                 turn, _ = controller.store.create_queued_turn(current["session_id"],
                     client_turn_id=client_id, message=row["message"], origin="lark",
                     external_agent_target={"target": target, "context": selected["context"]} if target else None)
@@ -176,6 +177,7 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
                     mode="resume_latest", conversation_binding_id=row["binding_id"], source_context=row["source"])
             row["session_id"] = current["session_id"]
             _atomic_write_json(path, row)
+            self._record_steward_ingress(row, selected, current, plan["client_turn_id"])
             try:
                 turn, _ = controller.enqueue_turn(session_id=current["session_id"],
                     client_turn_id=plan["client_turn_id"], message=row["message"],
@@ -189,6 +191,32 @@ def _admit_prepared(self, path: Path, row: dict[str, Any], selected: dict[str, A
         _atomic_write_json(path, row)
         return row
 
+    def _record_steward_ingress(self, row: dict[str, Any], selected: dict[str, Any],
+                               session: dict[str, Any], client_turn_id: str) -> None:
+        if selected["binding"]["context_kind"] != "steward":
+            return
+        if session.get("goal_id") != "loopx-manager" or session.get("channel_id") != selected["channel_id"]:
+            raise ValueError("the verified steward source belongs to another Session")
+        from ..manager_context import register_ingress
+        from ...control_plane.collaboration.inbox import normalize_source_context
+
+        try:
+            normalize_source_context(row["message"])
+        except ValueError:
+            # Inbox bounds are not ordinary Chat admission bounds. Preserve the
+            # full native message; no provenance means no context delivery.
+            return
+
+        # Persist provenance before enqueue can launch the model. This is the
+        # existing inbox owner, not a delivery grant: its separate sender/target
+        # policy still decides which registered recipient may receive context.
+        # The independently verified App-scoped owner reference is deliberately
+        # used instead of copying another profile's raw provider identity.
+        register_ingress(getattr(self.controller, "coordination_runtime_root", self.controller.store.root.parent),
+            session_id=session["session_id"], client_turn_id=client_turn_id,
+            channel=selected["channel_id"], sender_id=selected["context"]["operator_ref"],
+            message=row["message"], source_id=row["request_ref"])
+
     def pending(self) -> list[dict[str, Any]]:
         return [_read_json(path) for path in sorted(self.root.glob("*.json"))]
 
 
@@ -231,15 +231,15 @@
     },
     {
       "site": "loopx/capabilities/manager_context/__init__.py::<module>.configure_delivery_target.update::codec_read:load_project_registry#1",
-      "line": 377,
+      "line": 382,
       "column": 51,
       "kind": "codec_read",
       "api": "load_project_registry",
       "classification": "codec_api"
     },
     {
       "site": "loopx/capabilities/manager_context/__init__.py::<module>.configure_evidence_scope::codec_read:load_project_registry#1",
-      "line": 332,
+      "line": 338,
       "column": 16,
       "kind": "codec_read",
       "api": "load_project_registry",
 
@@ -0,0 +1,185 @@
+"""Verified native sources can reach the existing, separately granted inbox."""
+import json
+
+import pytest
+
+from test_native_steward_private import steward, finish  # noqa: F401
+from loopx.capabilities.manager_context import (
+    POLICY_SCHEMA, _hash, _root, _write, authority, configure_delivery_target,
+    configure_evidence_scope, deliver, pending,
+)
+
+
+def ingress_path(store, row):
+    return _root(store.root.parent) / "ingress" / (
+        _hash([row["session_id"], "external-" + row["request_ref"]]) + ".json")
+
+
+@pytest.mark.parametrize("separate_coordination_root", [False, True])
+def test_native_ingress_uses_coordination_owner_with_legacy_fallback(steward, monkeypatch, tmp_path, separate_coordination_root):  # noqa: F811
+    store, runtime, provider, transport, binding, _, workspace = steward
+    root = tmp_path / "coordination" if separate_coordination_root else store.root.parent
+    if separate_coordination_root:
+        runtime.coordination_runtime_root = root
+    target = {"goal_id": "registered-work", "agent_id": "worker"}
+    registry = json.loads(runtime.registry_path.read_text())
+    registry["runtime_root"] = str(root)
+    registry["goals"] = [{"id": target["goal_id"], "repo": str(workspace),
+                          "coordination": {"registered_agents": [target["agent_id"]]}}]
+    runtime.registry_path.write_text(json.dumps(registry))
+    # Observe provenance at the actual enqueue boundary, before model execution.
+    monkeypatch.setattr(runtime, "resume_session_queue", lambda **kwargs: None)
+    enqueue = runtime.enqueue_turn
+    saved = []
+
+    def observe(**kwargs):
+        path = _root(root) / "ingress" / (_hash([kwargs["session_id"], kwargs["client_turn_id"]]) + ".json")
+        saved.append(path.read_bytes())
+        return enqueue(**kwargs)
+
+    monkeypatch.setattr(runtime, "enqueue_turn", observe)
+    event = provider.event("steward-app", "canonical", "Original request for the registered worker")
+    assert transport.admit("steward-app", event)["status"] == "durably_accepted"
+    row = transport.core.pending()[0]
+    session = store.load_session(row["session_id"])
+    turn = store.load_turn(row["session_id"], row["turn_id"])
+    assert turn["status"] == "queued" and len(saved) == 1
+    if separate_coordination_root:
+        assert not ingress_path(store, row).exists()
+    policy_path = _root(root) / "policy.json"
+    _write(policy_path, {"schema_version": POLICY_SCHEMA, "sources": {session["channel_id"]: {
+        "sender_ids": [binding["operator_ref"]], "local_delivery_scope": "selected", "targets": [],
+    }}})
+    assert authority(root, runtime.registry_path, session, turn)["targets"] == []
+    configure_evidence_scope(root, runtime.registry_path, channel=session["channel_id"],
+                             goal_ids=[target["goal_id"]], execute=True)
+    configure_delivery_target(root, runtime.registry_path, channel=session["channel_id"],
+                              **target, grant=True, execute=True)
+    receipt = deliver(root, runtime.registry_path, session=session, turn=turn, request=target)
+    assert pending(root, **target)["items"][0]["message"] == event["content"]
+    assert deliver(root, runtime.registry_path, session=session, turn=turn,
+                   request=target) == {**receipt, "replayed": True}
+    assert transport.admit("steward-app", event)["status"] == "durably_accepted"
+    assert len(saved) == 1 and len(store.list_sessions()) == 1
+    # A sibling App's project source cannot acquire manager provenance or policy.
+    monkeypatch.setattr(runtime, "enqueue_turn", enqueue)
+    transport.admit("notes-app", provider.event("notes-app", "other", "Other audience"))
+    notes = next(item for item in transport.core.pending() if item["message"] == "Other audience")
+    assert not authority(root, runtime.registry_path, store.load_session(notes["session_id"]),
+                         store.load_turn(notes["session_id"], notes["turn_id"]))["targets"]
+    configure_delivery_target(root, runtime.registry_path, channel=session["channel_id"],
+                              **target, grant=False, execute=True)
+    with pytest.raises(ValueError, match="not authorized"):
+        deliver(root, runtime.registry_path, session=session, turn=turn, request=target)
+
+
+def test_verified_ingress_precedes_model_launch_and_survives_admission_replay(steward, monkeypatch):  # noqa: F811
+    store, runtime, provider, transport, binding, _, _ = steward
+    enqueue = runtime.enqueue_turn
+    observations = []
+
+    def interrupted(**kwargs):
+        row = transport.core.pending()[0]
+        saved = json.loads(ingress_path(store, row).read_text())
+        assert saved["client_turn_id"] == kwargs["client_turn_id"]
+        assert saved["sender_id"] == binding["operator_ref"]
+        assert saved["source_id"] == row["request_ref"]
+        assert saved["message_digest"] == _hash(kwargs["message"])
+        observations.append(saved)
+        enqueue(**kwargs)
+        raise OSError("owner interrupted after canonical admission")
+
+    event = provider.event("steward-app", "first", "Original owner context")
+    monkeypatch.setattr(runtime, "enqueue_turn", interrupted)
+    with pytest.raises(OSError):
+        transport.admit("steward-app", event)
+    prepared = transport.core.pending()[0]
+    before = ingress_path(store, prepared).read_bytes()
+    turn = store.turn_for_client(prepared["session_id"], "external-" + prepared["request_ref"])
+    assert finish(runtime, {**prepared, "turn_id": turn["turn_id"]})["status"] == "completed"
+    monkeypatch.setattr(runtime, "enqueue_turn", enqueue)
+    assert transport.admit("steward-app", event)["status"] == "durably_accepted"
+    accepted = transport.core.pending()[0]
+    assert accepted["session_id"] == prepared["session_id"] and accepted["turn_id"] == turn["turn_id"]
+    assert ingress_path(store, accepted).read_bytes() == before and len(observations) == 1
+    assert len(store.list_sessions()) == 1
+    # A regular project source cannot create manager provenance or a Goal.
+    transport.admit("notes-app", provider.event("notes-app", "notes", "Private project context"))
+    notes = next(row for row in transport.core.pending() if row["message"] == "Private project context")
+    assert finish(runtime, notes)["status"] == "completed"
+    assert not ingress_path(store, notes).exists()
+
+
+def test_native_steward_uses_separate_sender_target_grants_and_exact_return_route(steward):  # noqa: F811
+    store, runtime, provider, transport, binding, _, workspace = steward
+    target = {"goal_id": "fresh-commission", "agent_id": "operations"}
+    registry = json.loads(runtime.registry_path.read_text())
+    registry["goals"] = [{"id": target["goal_id"], "repo": str(workspace),
+                          "coordination": {"registered_agents": [target["agent_id"]]}}]
+    runtime.registry_path.write_text(json.dumps(registry))
+    registry_before = runtime.registry_path.read_bytes()
+    event = provider.event("steward-app", "source", "Read this source; return private findings, never publish.")
+    assert transport.admit("steward-app", event)["status"] == "durably_accepted"
+    row = transport.core.pending()[0]
+    assert finish(runtime, row)["status"] == "completed"
+    session = store.load_session(row["session_id"])
+    turn = store.load_turn(row["session_id"], row["turn_id"])
+    root = store.root.parent
+    assert authority(root, runtime.registry_path, session, turn)["targets"] == []
+    policy = {"schema_version": POLICY_SCHEMA, "sources": {session["channel_id"]: {
+        "sender_ids": [binding["operator_ref"]], "local_delivery_scope": "selected", "targets": [],
+    }}}
+    policy_path = _root(root) / "policy.json"
+    _write(policy_path, policy)
+    # Reading a portfolio does not confer delivery authority.
+    configure_evidence_scope(root, runtime.registry_path, channel=session["channel_id"],
+                             goal_ids=[target["goal_id"]], execute=True)
+    assert authority(root, runtime.registry_path, session, turn)["targets"] == []
+    preview = configure_delivery_target(root, runtime.registry_path, channel=session["channel_id"], **target, grant=True)
+    assert preview["would_change"] and not preview["executed"]
+    assert authority(root, runtime.registry_path, session, turn)["targets"] == []
+    applied = configure_delivery_target(root, runtime.registry_path, channel=session["channel_id"], **target, grant=True, execute=True)
+    assert applied["readback_verified"]
+    assert authority(root, runtime.registry_path, session, turn)["targets"] == [target]
+    assert not authority(root, runtime.registry_path, session, {**turn, "message": "forged"})["targets"]
+    assert not authority(root, runtime.registry_path, session, {**turn, "origin": "web"})["targets"]
+    receipt = deliver(root, runtime.registry_path, session=session, turn=turn, request=target)
+    assert not receipt["todo_created"] and not receipt["execution_interrupted"]
+    assert deliver(root, runtime.registry_path, session=session, turn=turn, request=target) == {**receipt, "replayed": True}
+    assert pending(root, **target)["items"][0]["message"] == event["content"]
+    assert runtime.registry_path.read_bytes() == registry_before
+    assert policy_path.stat().st_mode & 0o777 == 0o600
+    # The independently verified second App cannot use this channel's grant.
+    transport.admit("notes-app", provider.event("notes-app", "notes", "Other private audience"))
+    notes = next(row for row in transport.core.pending() if row["message"] == "Other private audience")
+    assert finish(runtime, notes)["status"] == "completed"
+    assert not authority(root, runtime.registry_path, store.load_session(notes["session_id"]),
+                         store.load_turn(notes["session_id"], notes["turn_id"]))["targets"]
+    policy["sources"][session["channel_id"]]["targets"] = [target]
+    policy["sources"][session["channel_id"]]["sender_ids"] = ["another-App-owner"]
+    _write(policy_path, policy)
+    with pytest.raises(ValueError, match="not authorized"):
+        deliver(root, runtime.registry_path, session=session, turn=turn, request=target)
+
+
+@pytest.mark.parametrize("channel", ["manager.external.native", "manager.external.native." + "a" * 24,
+                                    "manager.external.native." + "a" * 24 + ".*",
+                                    "manager.external." + "a" * 23])
+def test_configuration_rejects_partial_or_wildcard_audiences(tmp_path, channel):
+    with pytest.raises(ValueError, match="exact external"):
+        configure_evidence_scope(tmp_path, tmp_path / "missing.json", channel=channel, goal_ids=[])
+    with pytest.raises(ValueError, match="exact external"):
+        configure_delivery_target(tmp_path, tmp_path / "missing.json", channel=channel,
+                                  goal_id="fresh-commission", agent_id="operations", grant=True)
+
+
+def test_inbox_bound_does_not_truncate_or_reject_ordinary_native_chat(steward, monkeypatch):  # noqa: F811
+    store, runtime, provider, transport, _, _, _ = steward
+    monkeypatch.setattr(runtime, "resume_session_queue", lambda **kwargs: None)
+    message = "x" * 32001
+    assert transport.admit("steward-app", provider.event("steward-app", "large", message))["status"] == "durably_accepted"
+    row = transport.core.pending()[0]
+    turn = store.load_turn(row["session_id"], row["turn_id"])
+    assert turn["message"] == message and turn["status"] == "queued"
+    assert not ingress_path(store, row).exists()
+    assert not authority(store.root.parent, runtime.registry_path, store.load_session(row["session_id"]), turn)["targets"]