diff --git a/ms_agent/agent_hub/_workspace.py b/ms_agent/agent_hub/_workspace.py
index d1ee571f4..cd2d07322 100644
--- a/ms_agent/agent_hub/_workspace.py
+++ b/ms_agent/agent_hub/_workspace.py
@@ -69,31 +69,283 @@ def is_secret_key(name: str) -> bool:
     return bool(_SECRET_KEY_RE.search(name.strip()))
 
 
-def scrub_yaml_secrets(
-    text: str, mcp_block_keys: tuple[str, ...] = ('mcp_servers', 'mcpServers')
-) -> str:
+# Mapping keys whose VALUES are secret bags regardless of the inner key names:
+# ``env`` (MCP server environment variables) and ``headers`` (HTTP headers are
+# bearer credentials in disguise -- their names are arbitrary, e.g. a gateway
+# may demand ``X-Auth-Code``, which no key-name vocabulary could enumerate).
+# Every scrubber blanks the whole mapping, keeping only the key names.
+SECRET_BAG_KEYS = frozenset(('env', 'headers'))
+
+# Keys whose LIST value is a stdio command line to scrub positionally
+# (``args`` and its common alias ``argv``) -- shared by every scrubber.
+ARGS_LIST_KEYS = ('args', 'argv')
+
+_URL_RE = re.compile(r'^[A-Za-z][A-Za-z0-9+.\-]*://')
+
+# Query-parameter names treated as secrets BEYOND :func:`is_secret_key`:
+# an OAuth ``code`` or a request ``signature`` is a credential in a URL even
+# though a bare config key named ``code`` is not (too broad for the global
+# vocabulary). Header names need no counterpart -- they have the bag rule.
+_URL_SECRET_PARAMS = frozenset(('sig', 'signature', 'pwd', 'code'))
+
+
+def scrub_url_secrets(url: str) -> str:
+    """Strip credentials embedded in a URL, leaving the rest byte-identical.
+
+    Three carriers are removed:
+
+    * userinfo passwords -- ``https://user:pass@host`` becomes
+      ``https://user@host`` (the username is kept);
+    * bare userinfo -- ``https://<token>@host`` (no colon) is dropped
+      entirely: colon-less userinfo is how PAT-style tokens travel
+      (``https://ghp_xxx@host``) and cannot be told from a username, so
+      fail-closed wins;
+    * query parameters whose NAME matches :func:`is_secret_key` or
+      :data:`_URL_SECRET_PARAMS` -- the name is kept and the VALUE blanked
+      (``?api_key=X&model=y`` -> ``?api_key=&model=y``), mirroring how
+      headers / env / args keep names and blank values.
+
+    Only a string that IS a bare absolute URL is processed: it must start
+    with ``scheme://`` and contain no whitespace, so free text that merely
+    CONTAINS a URL (a prompt sentence, a doc note) is never truncated or
+    rewritten, and a ``URL # comment`` scalar is handled by the caller's
+    comment split. Non-secret URLs round-trip byte-identically; malformed
+    input is returned as-is (this helper must never raise on the outbound
+    path).
+    """
+    if not isinstance(url, str) or not _URL_RE.match(url):
+        return url
+    if re.search(r'\s', url):
+        return url
+    scheme, sep, rest = url.partition('://')
+    # The authority ends at the first '/', '?' or '#'.
+    cut = len(rest)
+    for ch in ('/', '?', '#'):
+        idx = rest.find(ch)
+        if idx != -1:
+            cut = min(cut, idx)
+    authority, tail = rest[:cut], rest[cut:]
+    at = authority.rfind('@')
+    if at != -1:
+        userinfo = authority[:at]
+        user, colon, password = userinfo.partition(':')
+        if colon and password:
+            authority = (f'{user}@' if user else '') + authority[at + 1:]
+        elif userinfo:
+            # Bare userinfo (no colon): indistinguishable from a token.
+            authority = authority[at + 1:]
+    path = tail
+    query = fragment = ''
+    if '#' in path:
+        path, frag = path.split('#', 1)
+        fragment = '#' + frag
+    if '?' in path:
+        path, q = path.split('?', 1)
+        pairs = []
+        for p in q.split('&'):
+            name = p.split('=', 1)[0]
+            if p and (is_secret_key(name)
+                      or name.lower() in _URL_SECRET_PARAMS):
+                pairs.append(name + '=')
+            else:
+                pairs.append(p)
+        query = '?' + '&'.join(pairs)
+    return f'{scheme}{sep}{authority}{path}{query}{fragment}'
+
+
+_SECRET_FLAG_RE = re.compile(r'^-{1,2}[A-Za-z][\w.\-]*$')
+
+
+def scrub_args_secrets(args: list) -> list:
+    """Blank secrets passed on a stdio command line (an ``args`` list).
+
+    Three spellings are covered:
+
+    * ``--api-key VALUE`` -- when a flag's name matches
+      :func:`is_secret_key`, the FOLLOWING element (the value) is blanked;
+    * ``--api-key=VALUE`` / ``-token=VALUE`` -- the value after ``=`` is
+      blanked in place (dash count does not matter);
+    * ``NAME=VALUE`` -- env-style assignments like the ones docker ``-e``
+      forwards (``["-e", "GITHUB_TOKEN=ghp_x"]``): NAME is checked against
+      the vocabulary and the value blanked, keeping the ``NAME=`` prefix.
+
+    Elements containing ``://`` are URLs, never NAME=VALUE pairs -- they are
+    left to the callers' URL pass. Non-secret elements pass through as-is;
+    non-string items are untouched.
+    """
+    if not isinstance(args, list):
+        return args
+    out = list(args)
+    n = len(out)
+    i = 0
+    while i < n:
+        item = out[i]
+        if isinstance(item, str):
+            if '=' in item and '://' not in item:
+                name = item.partition('=')[0]
+                if is_secret_key(name.lstrip('-')):
+                    out[i] = name + '='
+            elif _SECRET_FLAG_RE.match(item):
+                name = item.lstrip('-')
+                if is_secret_key(name) and i + 1 < n \
+                        and isinstance(out[i + 1], str) \
+                        and not _SECRET_FLAG_RE.match(out[i + 1]):
+                    out[i + 1] = ''
+                    i += 1
+        i += 1
+    return out
+
+
+def _split_inline_comment(val: str) -> tuple[str, str]:
+    """Split a trailing ``# ...`` comment off a scalar value.
+
+    A ``#`` opens a comment only outside quotes and either at the start of
+    the value or right after whitespace (the YAML/TOML rule), so
+    ``https://h/p#frag`` and ``"a # b"`` keep their ``#``. Returns
+    ``(value_region, comment)`` where *comment* is ``''`` when there is no
+    comment; *value_region* keeps any whitespace that preceded the ``#``.
+    """
+    quote = ''
+    for idx, ch in enumerate(val):
+        if quote:
+            if ch == quote:
+                quote = ''
+        elif ch in '\'"':
+            quote = ch
+        elif ch == '#' and (idx == 0 or val[idx - 1] in ' \t'):
+            return val[:idx], val[idx:]
+    return val, ''
+
+
+def scrub_scalar_url_token(val: str) -> str:
+    """Scrub credentials in a URL scalar, keeping quotes/whitespace/comment.
+
+    Used by the line-based YAML/TOML scrubbers on raw ``key: value``
+    captures: the value may carry surrounding whitespace, quotes and a
+    trailing ``# comment``. The comment is split off FIRST (outside quotes)
+    so ``url: https://h?token=X # note`` scrubs the URL part only -- the
+    comment is re-appended verbatim instead of being glued into the value.
+    Returns *val* byte-identical unless a secret was actually removed.
+    """
+    region, comment = _split_inline_comment(val)
+    lead = region[:len(region) - len(region.lstrip())]
+    trail = region[len(region.rstrip()):]
+    core = region.strip()
+    quote = ''
+    if len(core) >= 2 and core[0] == core[-1] and core[0] in '\'"':
+        quote = core[0]
+        core = core[1:-1]
+    cleaned = scrub_url_secrets(core)
+    if cleaned == core:
+        return val
+    return f'{lead}{quote}{cleaned}{quote}{trail}{comment}'
+
+
+def _scrub_yaml_flow_args(val: str) -> str:
+    """Positional scrub of a flow-style ``args: [-y, srv, --api-key, X]``.
+
+    Returns the rewritten value text (brackets included) or *val* unchanged
+    when nothing secret was found. Items may be quoted; quotes are stripped
+    for flag detection and dropped in the (rewritten) output. Mirrors
+    :func:`scrub_args_secrets` plus URL scrubbing of each item.
+    """
+    lead = val[:len(val) - len(val.lstrip())]
+    body = val.strip()
+    if not body.startswith('['):
+        return val
+    end = body.find(']')
+    if end == -1:
+        return val
+    inner, trail = body[1:end], body[end + 1:]
+    items = [it.strip() for it in inner.split(',')] if inner.strip() else []
+    norm = []
+    for it in items:
+        if len(it) >= 2 and it[0] == it[-1] and it[0] in '\'"':
+            norm.append(it[1:-1])
+        else:
+            norm.append(it)
+    scrubbed = [scrub_url_secrets(it) for it in scrub_args_secrets(norm)]
+    if scrubbed == norm:
+        return val
+    parts = []
+    for orig, norm_it, scrub_it in zip(items, norm, scrubbed):
+        if scrub_it == norm_it:
+            parts.append(orig)
+        elif scrub_it == '':
+            parts.append("''")
+        else:
+            parts.append(scrub_it)
+    return f'{lead}[{", ".join(parts)}]{trail}'
+
+
+def scrub_toml_array_args(val: str) -> str:
+    """Positional scrub of a single-line TOML ``args = ["-y", "--token", X]``.
+
+    Like :func:`_scrub_yaml_flow_args` but TOML strings must stay quoted, so
+    each item keeps (or receives) its quotes in the rewritten output. Returns
+    *val* unchanged when nothing secret was found.
+    """
+    lead = val[:len(val) - len(val.lstrip())]
+    body = val.strip()
+    if not body.startswith('['):
+        return val
+    end = body.find(']')
+    if end == -1:
+        return val
+    inner, trail = body[1:end], body[end + 1:]
+    items = [it.strip() for it in inner.split(',') if it.strip()]
+    norm, quotes = [], []
+    for it in items:
+        if len(it) >= 2 and it[0] == it[-1] and it[0] in '\'"':
+            quotes.append(it[0])
+            norm.append(it[1:-1])
+        else:
+            quotes.append('"')
+            norm.append(it)
+    scrubbed = [scrub_url_secrets(it) for it in scrub_args_secrets(norm)]
+    if scrubbed == norm:
+        return val
+    parts = []
+    for orig, norm_it, scrub_it, q in zip(items, norm, scrubbed, quotes):
+        if scrub_it == norm_it:
+            parts.append(orig)
+        else:
+            parts.append(f'{q}{scrub_it}{q}')
+    return f'{lead}[{", ".join(parts)}]{trail}'
+
+
+def scrub_yaml_secrets(text: str) -> str:
     """Blank secret values in a YAML config *text*, line-by-line.
 
     Regex/line based (not a YAML round-trip) so user comments, key order and
-    formatting are preserved -- only the secret *values* are cleared. Two rules
-    are applied per ``key: value`` line:
+    formatting are preserved -- only the secret *values* are cleared. Rules
+    per ``key: value`` line:
 
     * a key whose name matches :func:`is_secret_key` -> value cleared, anywhere
       in the tree (e.g. top-level ``model.api_key`` or ``llm.modelscope_api_key``);
-    * every scalar nested inside an MCP-server ``env`` mapping (a block opened
-      by any key in *mcp_block_keys*, then a nested ``env:``) -> value cleared
-      regardless of key name (that block is a free-form secret bag where API
-      keys live under arbitrary names).
+    * every scalar nested inside an ``env`` or ``headers`` mapping
+      (:data:`SECRET_BAG_KEYS`) -> value cleared regardless of key name, at
+      ANY depth (those blocks are free-form secret bags where API keys live
+      under arbitrary names) -- same policy as the JSON/TOML scrubbers;
+    * any scalar that IS a bare absolute URL -> userinfo credentials and
+      secret query parameters stripped (:func:`scrub_url_secrets`), wherever
+      it lives (``base_url``, MCP ``url``, ...); a trailing ``# comment`` is
+      split off first so it is never glued into the value;
+    * ``args`` lists anywhere (block ``- item`` and flow ``[a, b]`` styles)
+      -> secret flag values, ``NAME=VALUE`` env assignments and URL items are
+      scrubbed, mirroring :func:`scrub_args_secrets`.
 
     Beyond simple ``key: <scalar>`` lines the scrubber also covers the other
     legal spellings a secret can hide in:
 
     * flow mappings (``llm: {api_key: X, model: y}``) -- secret pairs inside
-      the braces are cleared in place; an ``env: {...}`` flow mapping is
-      cleared wholesale;
+      the braces are cleared in place; an ``env: {...}`` / ``headers: {...}``
+      flow mapping is cleared wholesale, and URL values inside are scrubbed;
     * block/folded scalars (``api_key: |`` / ``>`` and their chomping
       variants) -- the opener is blanked and the indented content lines that
-      carry the secret are dropped.
+      carry the secret are dropped;
+    * a block opener carrying an inline comment (``mcp_servers:  # remote``)
+      is still recognized as an opener.
 
     Comments, blank lines and non-secret lines are emitted verbatim.
 
@@ -106,10 +358,15 @@ def scrub_yaml_secrets(
     flow_pair = re.compile(r'(?P<key>["\']?[\w.\-]+["\']?)(?P<sep>\s*:\s*)'
                            r'(?P<val>[^,{}\[\]]+)')
     block_opener = re.compile(r'^[|>][+-]?\d*$')
-    # Indent of the active mcp-servers / nested ``env:`` block openers;
-    # ``None`` means we are not currently inside that block.
-    mcp_indent: int | None = None
-    env_indent: int | None = None
+    list_item = re.compile(r'^(?P<indent>[ \t]*)-[ \t]+(?P<val>\S.*)$')
+    # Indent of the active secret-bag (``env`` / ``headers``) / ``args``
+    # block openers; ``None`` = not inside. Both are tracked at ANY depth,
+    # mirroring the JSON scrubber's global policy.
+    bag_indent: int | None = None
+    args_indent: int | None = None
+    # Inside a block ``args:`` list: the previous item was a secret flag whose
+    # value is still owed (``--api-key`` -> the NEXT item is that value).
+    args_pending = False
     # Inside a secret block scalar: drop lines indented deeper than this.
     skip_indent: int | None = None
     out: list[str] = []
@@ -126,25 +383,65 @@ def scrub_yaml_secrets(
         if not stripped or stripped.startswith('#'):
             out.append(line)
             continue
+        # Block ``args:`` list items: positional scrub of the stdio command
+        # line (secret flag values blanked, URL items scrubbed).
+        if args_indent is not None:
+            cur = len(line) - len(line.lstrip(' \t'))
+            am = list_item.match(line)
+            if am and cur >= args_indent:
+                core = am.group('val').strip()
+                quote = ''
+                if len(core) >= 2 and core[0] == core[-1] \
+                        and core[0] in '\'"':
+                    quote = core[0]
+                    core = core[1:-1]
+                if args_pending and not _SECRET_FLAG_RE.match(core):
+                    # The value owed to the previous secret flag: blank it.
+                    out.append(f"{am.group('indent')}- ''")
+                    args_pending = False
+                    continue
+                args_pending = False
+                if _SECRET_FLAG_RE.match(core) and '=' not in core \
+                        and is_secret_key(core.lstrip('-')):
+                    args_pending = True
+                    out.append(line)
+                    continue
+                if '=' in core and '://' not in core:
+                    name = core.partition('=')[0]
+                    if is_secret_key(name.lstrip('-')):
+                        out.append(f"{am.group('indent')}- {quote}{name}="
+                                   f"{quote}")
+                        continue
+                cleaned = scrub_url_secrets(core)
+                if cleaned != core:
+                    out.append(f"{am.group('indent')}- {quote}{cleaned}"
+                               f"{quote}")
+                    continue
+                out.append(line)
+                continue
+            if cur <= args_indent:
+                args_indent = None
+                args_pending = False
         m = kv.match(line)
         if not m:
             out.append(line)
             continue
         indent = len(m.group('indent'))
         # Dedenting to <= a block opener's indent leaves that block.
-        if env_indent is not None and indent <= env_indent:
-            env_indent = None
-        if mcp_indent is not None and indent <= mcp_indent:
-            mcp_indent = None
+        if bag_indent is not None and indent <= bag_indent:
+            bag_indent = None
+        if args_indent is not None and indent <= args_indent:
+            args_indent = None
+            args_pending = False
         key = m.group('key').strip()
         val = m.group('val')
-        in_env = env_indent is not None and indent > env_indent
+        in_bag = bag_indent is not None and indent > bag_indent
         secret_key = is_secret_key(key)
         # Block/folded scalar opener (| / > + chomping variants): the secret
         # lives on the FOLLOWING deeper-indented lines -- blank the key and
         # drop that content.
         if val is not None and block_opener.match(val.strip()):
-            if secret_key or in_env:
+            if secret_key or in_bag:
                 out.append(
                     f"{m.group('indent')}{m.group('key')}{m.group('sep')}''")
                 skip_indent = indent
@@ -152,34 +449,57 @@ def scrub_yaml_secrets(
                 out.append(line)
             continue
         # Flow mapping value: scrub secret pairs inside the braces. An
-        # ``env`` (or secret-named) flow mapping is a free-form secret bag ->
-        # clear every value in it.
+        # ``env`` / ``headers`` (or secret-named) flow mapping is a free-form
+        # secret bag -> clear every value in it.
         if val is not None and val.lstrip().startswith(
                 '{') and val.strip() != '{}':
-            clear_all = secret_key or in_env or key == 'env'
+            clear_all = secret_key or in_bag or key in SECRET_BAG_KEYS
 
             def _repl(pm, _all=clear_all):
                 if _all or is_secret_key(pm.group('key').strip('"\'')):
                     return f"{pm.group('key')}{pm.group('sep')}''"
+                cleaned = scrub_scalar_url_token(pm.group('val'))
+                if cleaned != pm.group('val'):
+                    return (f"{pm.group('key')}{pm.group('sep')}{cleaned}")
                 return pm.group(0)
 
             out.append(f"{m.group('indent')}{m.group('key')}{m.group('sep')}"
                        f'{flow_pair.sub(_repl, val)}')
             continue
-        has_scalar = val is not None and val.strip() not in ('{}', '[]')
+        # A trailing comment must not hide a block opener:
+        # ``mcp_servers:  # remote`` still opens a block.
+        bare_val = _split_inline_comment(val)[0].strip() if val else None
+        has_scalar = bare_val not in (None, '', '{}', '[]')
         # A key with no scalar value opens a mapping/list block: track the
-        # mcp-servers and nested ``env`` openers so their descendants can be
-        # scoped, then emit the opener line unchanged.
+        # secret-bag and ``args`` openers (at any depth, matching the JSON
+        # policy) so their descendants can be scoped, then emit the opener
+        # line unchanged.
         if not has_scalar:
-            if key in mcp_block_keys:
-                mcp_indent = indent
-            elif key == 'env' and mcp_indent is not None:
-                env_indent = indent
+            if key in SECRET_BAG_KEYS:
+                bag_indent = indent
+            elif key in ARGS_LIST_KEYS:
+                args_indent = indent
+                args_pending = False
             out.append(line)
             continue
-        if secret_key or in_env:
+        if secret_key or in_bag:
             out.append(
                 f"{m.group('indent')}{m.group('key')}{m.group('sep')}''")
+            continue
+        # Flow-list ``args: [...]``: positional scrub inside the brackets.
+        if key in ARGS_LIST_KEYS and val.lstrip().startswith('['):
+            cleaned = _scrub_yaml_flow_args(val)
+            if cleaned != val:
+                out.append(f"{m.group('indent')}{m.group('key')}"
+                           f"{m.group('sep')}{cleaned}")
+            else:
+                out.append(line)
+            continue
+        cleaned = scrub_scalar_url_token(val)
+        if cleaned != val:
+            out.append(
+                f"{m.group('indent')}{m.group('key')}{m.group('sep')}"
+                f'{cleaned}')
         else:
             out.append(line)
     return '\n'.join(out)
@@ -194,23 +514,40 @@ def scrub_json_secrets(obj) -> None:
     * a key matching :func:`is_secret_key` -> value blanked to ``''`` whatever
       its type (a nested mapping under e.g. ``credentials`` is wiped wholesale
       -- its inner field names may look harmless);
-    * every value inside an ``env`` mapping -> blanked regardless of key name
-      (an ``env`` block, e.g. under an MCP server, is a free-form secret bag).
+    * every value inside an ``env`` or ``headers`` mapping -> blanked
+      regardless of key name (:data:`SECRET_BAG_KEYS`; both are free-form
+      secret bags -- env var and header names are arbitrary);
+    * command-line lists under ``args`` / ``argv`` (:data:`ARGS_LIST_KEYS`)
+      -> secret flag values and ``NAME=VALUE`` env assignments blanked
+      (:func:`scrub_args_secrets`);
+    * every string that IS a bare absolute URL -> userinfo credentials and
+      secret query parameters stripped (:func:`scrub_url_secrets`), wherever
+      it lives (provider ``base_url``, MCP ``url``, ...). Free text that
+      merely CONTAINS a URL (a prompt sentence) is left untouched -- only
+      whitespace-free, scheme-led strings are treated as URLs.
 
     Non-secret structure and values are preserved; nested dicts / lists are
     walked recursively.
     """
     if isinstance(obj, dict):
         for key, val in obj.items():
-            if key == 'env' and isinstance(val, dict):
+            if key in SECRET_BAG_KEYS and isinstance(val, dict):
                 obj[key] = {k: '' for k in val}
             elif is_secret_key(key):
                 obj[key] = ''
+            elif key in ARGS_LIST_KEYS and isinstance(val, list):
+                obj[key] = scrub_args_secrets(val)
+                scrub_json_secrets(obj[key])
+            elif isinstance(val, str):
+                obj[key] = scrub_url_secrets(val)
             else:
                 scrub_json_secrets(val)
     elif isinstance(obj, list):
-        for item in obj:
-            scrub_json_secrets(item)
+        for idx, item in enumerate(obj):
+            if isinstance(item, str):
+                obj[idx] = scrub_url_secrets(item)
+            else:
+                scrub_json_secrets(item)
 
 
 class WorkspaceSpec(ABC):
diff --git a/ms_agent/agent_hub/frameworks/hermes.py b/ms_agent/agent_hub/frameworks/hermes.py
index 5f7a93478..5fb84a359 100644
--- a/ms_agent/agent_hub/frameworks/hermes.py
+++ b/ms_agent/agent_hub/frameworks/hermes.py
@@ -128,9 +128,10 @@ def _scrub_yaml_secrets(self, text: str) -> str:
         """Blank secret values in ``config.yaml`` line-by-line.
 
         Thin wrapper over the shared :func:`scrub_yaml_secrets` (which every
-        YAML-config framework reuses) scoped to hermes' ``mcp_servers`` block.
+        YAML-config framework reuses): secret-named keys, ``env`` / ``headers``
+        bags, ``args`` command lines and URL credentials, at any depth.
         """
-        return scrub_yaml_secrets(text, mcp_block_keys=('mcp_servers', ))
+        return scrub_yaml_secrets(text)
 
     def _sanitize_config_file(self, rel_path: str, content: bytes) -> bytes:
         """Blank secrets in ``config.yaml`` (identical on up- and download).
diff --git a/ms_agent/agent_hub/frameworks/openhuman.py b/ms_agent/agent_hub/frameworks/openhuman.py
index 68b87097a..5750dee22 100644
--- a/ms_agent/agent_hub/frameworks/openhuman.py
+++ b/ms_agent/agent_hub/frameworks/openhuman.py
@@ -8,8 +8,9 @@
 from pathlib import Path
 
 from ms_agent.utils.logger import get_logger
-from .._workspace import (DEFAULT_AGENT_NAME, WorkspaceSpec, is_secret_key,
-                          register_framework)
+from .._workspace import (ARGS_LIST_KEYS, DEFAULT_AGENT_NAME, SECRET_BAG_KEYS,
+                          WorkspaceSpec, is_secret_key, register_framework,
+                          scrub_scalar_url_token, scrub_toml_array_args)
 
 logger = get_logger()
 
@@ -316,64 +317,140 @@ def _scrub_toml_secrets(self, text: str) -> str:
         # so ``a.b.api_key`` is caught, not just a bare top-level ``api_key``.
         # Beyond simple ``key = <scalar>`` lines this also covers:
         # * inline tables  ``provider = { api_key = "X" }`` (incl. nested) --
-        #   secret pairs inside the braces are cleared in place;
-        # * arrays         ``tokens = ["X"]`` -> ``tokens = []``;
+        #   secret pairs inside the braces are cleared in place; an
+        #   ``env`` / ``headers`` inline table (SECRET_BAG_KEYS) is cleared
+        #   wholesale -- those names are arbitrary bearer bags;
+        # * table sections ``[mcp.fs.headers]`` / ``[[...env]]`` -- when the
+        #   LAST path segment is a secret bag every assignment inside the
+        #   section is blanked, including quoted keys (``"X-Auth-Code"``)
+        #   that the bare-key pattern cannot match;
+        # * arrays         ``tokens = ["X"]`` -> ``tokens = []``; an ``args``
+        #   array (single- or multi-line) is positionally scrubbed (secret
+        #   flag values blanked); a clean multi-line array keeps its layout;
         # * multi-line strings (``secret = '''`` / ``\"\"\"``) -- the opener is
-        #   blanked and the content lines up to the closing delimiter dropped.
+        #   blanked and the content lines up to the closing delimiter dropped;
+        # * URL string values under any key -- userinfo passwords and
+        #   secret-named query parameters are stripped.
         pattern = re.compile(
             r'^(?P<pre>\s*(?P<key>[A-Za-z0-9_.-]+)\s*=\s*)(?P<val>.*)$')
         inline_pair = re.compile(r'(?P<key>[A-Za-z0-9_.-]+)(?P<sep>\s*=\s*)'
                                  r'(?P<val>"[^"]*"|\'[^\']*\'|[^,{}\s][^,}]*)')
+        section = re.compile(r'^\s*\[{1,2}\s*(?P<path>[^#\]\[]+?)\s*\]{1,2}'
+                             r'\s*(?:#.*)?$')
+        bagged_assign = re.compile(r'^(?P<pre>\s*.+?=\s*)(?P<val>.*)$')
         out: list[str] = []
         lines = text.split('\n')
+        # True while inside a ``[...headers]`` / ``[...env]`` table section.
+        in_bag_section = False
         i = 0
         while i < len(lines):
             line = lines[i]
+            sm = section.match(line)
+            if sm:
+                seg = sm.group('path').split('.')[-1].strip().strip('"\'')
+                in_bag_section = seg in SECRET_BAG_KEYS
+                out.append(line)
+                i += 1
+                continue
             m = pattern.match(line)
             if not m:
+                # Quoted keys (``"X-Auth-Code" = ...``) fail the bare-key
+                # pattern; inside a bag section they must still be blanked.
+                if in_bag_section and not line.strip().startswith('#'):
+                    bm = bagged_assign.match(line)
+                    if bm:
+                        i = self._blank_toml_value(
+                            out, lines, i, bm.group('pre'), bm.group('val'))
+                        continue
                 out.append(line)
                 i += 1
                 continue
-            key = m.group('key').split('.')[-1]
+            segs = [s.strip().strip('"\'')
+                    for s in m.group('key').split('.')]
+            key = segs[-1]
             val = m.group('val')
             vstrip = val.strip()
-            if is_secret_key(key):
-                # Multi-line string: blank the opener and drop content lines
-                # up to (and including) the closing delimiter.
-                delim = next((d for d in ('"""', "'''")
-                              if vstrip.startswith(d) and vstrip.count(d) < 2),
-                             None)
-                if delim:
-                    out.append(m.group('pre') + '""')
+            # Dotted path THROUGH a bag (``mcp.fs.headers.X = v``) is the
+            # same as living in a ``[...headers]`` section.
+            in_bag_path = any(s in SECRET_BAG_KEYS for s in segs[:-1])
+            if is_secret_key(key) or in_bag_section or in_bag_path:
+                i = self._blank_toml_value(out, lines, i, m.group('pre'), val)
+                continue
+            # Secret-bag inline table (``headers = {...}`` / ``env = {...}``):
+            # clear every pair -- the names inside are arbitrary.
+            if key in SECRET_BAG_KEYS and vstrip.startswith('{'):
+                out.append(
+                    m.group('pre') + inline_pair.sub(
+                        lambda pm: f"{pm.group('key')}{pm.group('sep')}\"\"",
+                        val))
+                i += 1
+                continue
+            # ``args = [...]``: positional flag scrub. A multi-line array is
+            # joined for the scrub and only re-emitted on one line when a
+            # secret was actually removed (clean arrays keep their layout).
+            if key in ARGS_LIST_KEYS and vstrip.startswith('['):
+                if ']' in vstrip:
+                    out.append(m.group('pre') + scrub_toml_array_args(val))
                     i += 1
-                    while i < len(lines) and delim not in lines[i]:
-                        i += 1
-                    i += 1  # skip the closing-delimiter line
                     continue
-                if vstrip.startswith('['):
-                    out.append(m.group('pre') + '[]')
-                    if ']' not in vstrip:  # multi-line array
-                        i += 1
-                        while i < len(lines) and ']' not in lines[i]:
-                            i += 1
-                    i += 1
+                j = i + 1
+                parts = [val.strip()]
+                while j < len(lines) and ']' not in lines[j]:
+                    parts.append(lines[j].strip())
+                    j += 1
+                if j < len(lines):
+                    parts.append(lines[j].strip())
+                    joined = ' '.join(parts)
+                    cleaned = scrub_toml_array_args(joined)
+                    if cleaned == joined:
+                        out.extend(lines[i:j + 1])
+                    else:
+                        out.append(m.group('pre') + cleaned.strip())
+                    i = j + 1
                     continue
-                out.append(m.group('pre') + '""')
-                i += 1
-                continue
             # Non-secret key: still scrub secret pairs inside inline tables
-            # (``provider = { api_key = "X" }``, nested tables included).
+            # (``provider = { api_key = "X" }``, nested tables included), and
+            # strip credentials from URL string values.
             if '{' in vstrip:
-                out.append(
-                    m.group('pre') + inline_pair.sub(
-                        lambda pm: f"{pm.group('key')}{pm.group('sep')}\"\""
-                        if is_secret_key(pm.group('key').split('.')[-1]) else
-                        pm.group(0), val))
+                def _repl(pm):
+                    if is_secret_key(pm.group('key').split('.')[-1]):
+                        return f"{pm.group('key')}{pm.group('sep')}\"\""
+                    cleaned = scrub_scalar_url_token(pm.group('val'))
+                    if cleaned != pm.group('val'):
+                        return f"{pm.group('key')}{pm.group('sep')}{cleaned}"
+                    return pm.group(0)
+
+                out.append(m.group('pre') + inline_pair.sub(_repl, val))
                 i += 1
                 continue
-            out.append(line)
+            out.append(m.group('pre') + scrub_scalar_url_token(val))
             i += 1
         return '\n'.join(out)
 
+    @staticmethod
+    def _blank_toml_value(out: list[str], lines: list[str], i: int,
+                            pre: str, val: str) -> int:
+        """Blank the value of the assignment at ``lines[i]``, appending to
+        *out*. Handles multi-line strings (``'''`` / ``\"\"\"``), (multi-line)
+        arrays and plain scalars; returns the next line index to process."""
+        vstrip = val.strip()
+        delim = next((d for d in ('"""', "'''")
+                      if vstrip.startswith(d) and vstrip.count(d) < 2), None)
+        if delim:
+            out.append(pre + '""')
+            i += 1
+            while i < len(lines) and delim not in lines[i]:
+                i += 1
+            return i + 1
+        if vstrip.startswith('['):
+            out.append(pre + '[]')
+            if ']' not in vstrip:
+                i += 1
+                while i < len(lines) and ']' not in lines[i]:
+                    i += 1
+            return i + 1
+        out.append(pre + '""')
+        return i + 1
+
 
 register_framework('openhuman', OpenhumanWorkspace)
diff --git a/ms_agent/agent_hub/frameworks/qwenpaw.py b/ms_agent/agent_hub/frameworks/qwenpaw.py
index 385d01276..4801fc93a 100644
--- a/ms_agent/agent_hub/frameworks/qwenpaw.py
+++ b/ms_agent/agent_hub/frameworks/qwenpaw.py
@@ -8,8 +8,8 @@
 
 from ms_agent.utils.logger import get_logger
 from .._workspace import (ALL_AGENT_NAME, DEFAULT_AGENT_NAME,
-                          GLOBAL_AGENT_NAME, WorkspaceSpec, is_secret_key,
-                          register_framework)
+                          GLOBAL_AGENT_NAME, WorkspaceSpec,
+                          register_framework, scrub_json_secrets)
 from ._bundled_skills import BundledSkillFilterMixin
 
 logger = get_logger()
@@ -128,38 +128,17 @@ def _strip_agent_json_secrets(self, data: dict) -> None:
         key never lands on disk from a remote agent, and a local key is never
         pushed to the remote repo / its git history.
 
-        The whole JSON tree is walked recursively and any key matching
-        :func:`is_secret_key` is blanked wherever it lives (top-level
-        ``model.api_key``, ``channels.*.client_secret``, future additions...).
-        Two structural rules are applied on top of the vocabulary:
-
-        * channel configs additionally blank ``_CHANNEL_LOCAL_KEYS``
-          (machine-local paths such as ``db_path``);
-        * every ``env`` mapping is cleared wholesale WHEREVER it lives -- env
-          var names are arbitrary, so values there are treated as secrets
-          even when the name does not match the vocabulary.  Anchoring on the
-          ``env`` key (not on a parent block name) keeps every MCP schema
-          spelling covered (``mcp.clients`` / ``mcp_clients`` /
-          ``mcpClients``), mirroring :func:`scrub_json_secrets`.
+        The walk itself is delegated to the shared :func:`scrub_json_secrets`
+        so every framework scrubs with one policy: secret-named keys are
+        blanked at any depth (top-level ``model.api_key``,
+        ``channels.*.client_secret``, ...), ``env`` / ``headers`` mappings are
+        cleared wholesale, ``args`` command lines have their flag values
+        blanked, and URLs are stripped of userinfo passwords / secret query
+        parameters. On top of that vocabulary, channel configs additionally
+        blank ``_CHANNEL_LOCAL_KEYS`` (machine-local paths such as
+        ``db_path``) whose names carry no secret suffix.
         """
-
-        def scrub(node: Any) -> None:
-            if isinstance(node, dict):
-                for key, value in node.items():
-                    # Secret-named key: wipe the WHOLE value (even a nested
-                    # mapping) -- credentials blobs must not survive because
-                    # their inner field names happen to look harmless.
-                    if is_secret_key(key):
-                        node[key] = ''
-                    elif key == 'env' and isinstance(value, dict):
-                        node[key] = {k: '' for k in value}
-                    elif isinstance(value, (dict, list)):
-                        scrub(value)
-            elif isinstance(node, list):
-                for item in node:
-                    scrub(item)
-
-        scrub(data)
+        scrub_json_secrets(data)
         # Channel configs: also blank machine-local (non-secret-named) keys.
         channels = data.get('channels')
         if isinstance(channels, dict):
diff --git a/tests/agent_hub/test_workspace.py b/tests/agent_hub/test_workspace.py
index 9d817199a..6f1d8b27f 100644
--- a/tests/agent_hub/test_workspace.py
+++ b/tests/agent_hub/test_workspace.py
@@ -685,7 +685,7 @@ class TestScrubYamlTomlSpellings(unittest.TestCase):
 
     def _yaml(self, text):
         from ms_agent.agent_hub._workspace import scrub_yaml_secrets
-        return scrub_yaml_secrets(text, mcp_block_keys=("mcp_servers", ))
+        return scrub_yaml_secrets(text)
 
     def _toml(self, text):
         from ms_agent.agent_hub.frameworks.openhuman import OpenhumanWorkspace
@@ -916,5 +916,672 @@ def test_frameworks_without_subdirs_are_untouched(self):
                 self.assertEqual(spec.root, given, f"{fw} root changed")
 
 
+class TestUrlAndArgSecretHelpers(unittest.TestCase):
+    """Shared carrier helpers: URL credentials and stdio flag values.
+
+    Secrets travel in places no key-name vocabulary can reach: inside URLs
+    (query parameters, userinfo passwords) and as positional values after a
+    secret-named CLI flag (``--api-key VALUE``). These helpers strip them
+    while leaving clean input byte-identical.
+    """
+
+    def test_url_secret_param_values_blanked_name_kept(self):
+        # Names stay, values go -- consistent with headers / env / args, and
+        # the remote still sees the parameter structure.
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        out = scrub_url_secrets(
+            "https://api.example.com/v1?api_key=S1&token=S2&model=y")
+        self.assertEqual(out, "https://api.example.com/v1?api_key=&token="
+                            "&model=y")
+
+    def test_url_only_secret_params_keep_query_mark(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        out = scrub_url_secrets("https://api.example.com/sse?api_key=S1")
+        self.assertEqual(out, "https://api.example.com/sse?api_key=")
+
+    def test_url_extra_query_vocabulary(self):
+        # OAuth codes / signatures / passwords are credentials in a URL even
+        # though a bare config key named ``code`` is not.
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        out = scrub_url_secrets(
+            "https://cb.example.com/x?code=oa1&sig=S&signature=S&pwd=S&v=1")
+        self.assertEqual(out, "https://cb.example.com/x?code=&sig="
+                              "&signature=&pwd=&v=1")
+
+    def test_url_userinfo_password_stripped(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        out = scrub_url_secrets("https://user:pa55@host.example.com/p?x=1")
+        self.assertEqual(out, "https://user@host.example.com/p?x=1")
+
+    def test_url_bare_token_userinfo_dropped(self):
+        # No colon -> cannot be told from a PAT (https://ghp_xxx@host);
+        # fail-closed: the whole userinfo goes.
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        self.assertEqual(
+            scrub_url_secrets("https://ghp_LEAKTOKEN@host.example.com/x"),
+            "https://host.example.com/x")
+
+    def test_url_whitespace_guard_no_truncation(self):
+        # A string containing whitespace is prose that merely CONTAINS a url,
+        # not a bare URL: it must survive byte-identical, never truncated.
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        text = "https://docs.example.com/g?tokens=abc rest of sentence"
+        self.assertEqual(scrub_url_secrets(text), text)
+
+    def test_url_fragment_and_port_preserved(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        url = "https://host.example.com:8443/p?a=1#frag"
+        self.assertEqual(scrub_url_secrets(url), url)
+
+    def test_url_no_scheme_is_not_a_url(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        for val in ("not a url ?token=x", "example.com?token=x", "", 42):
+            self.assertEqual(scrub_url_secrets(val), val)
+
+    def test_clean_url_round_trips_byte_identical(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        for url in ("https://dashscope.aliyuncs.com/compatible-mode/v1",
+                    "https://host/x?model=qwen&temp=0.7",
+                    "https://host",
+                    "http://[::1]:8080/path"):
+            self.assertEqual(scrub_url_secrets(url), url)
+
+    def test_args_flag_value_blanked(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        out = scrub_args_secrets(
+            ["-y", "srv", "--api-key", "SECRET", "--model", "qwen"])
+        self.assertEqual(out, ["-y", "srv", "--api-key", "",
+                               "--model", "qwen"])
+
+    def test_args_flag_equals_form_blanked(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        out = scrub_args_secrets(["run", "--token=SECRET", "--port", "8080"])
+        self.assertEqual(out, ["run", "--token=", "--port", "8080"])
+
+    def test_args_single_dash_equals_form_blanked(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        out = scrub_args_secrets(["srv", "-token=SECRET"])
+        self.assertEqual(out, ["srv", "-token="])
+
+    def test_args_docker_env_assignment_blanked(self):
+        # Official docker-MCP spelling: the flag is ``-e`` (not in the
+        # vocabulary); the NEXT element is a NAME=VALUE pair whose NAME is.
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        out = scrub_args_secrets(
+            ["run", "-i", "--rm", "-e", "GITHUB_TOKEN=ghp_LEAK", "img"])
+        self.assertEqual(out, ["run", "-i", "--rm", "-e", "GITHUB_TOKEN=",
+                               "img"])
+
+    def test_args_benign_env_assignment_preserved(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        args = ["run", "-e", "RUST_LOG=debug", "-e", "IMAGE=nginx", "img"]
+        self.assertEqual(scrub_args_secrets(args), args)
+
+    def test_args_url_element_not_a_pair(self):
+        # ``://`` marks a URL, never a NAME=VALUE assignment.
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        args = ["--url", "https://h.example.com/p?a=1"]
+        self.assertEqual(scrub_args_secrets(args), args)
+
+    def test_args_flag_value_followed_by_flag_not_eaten(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        out = scrub_args_secrets(["--api-key", "--verbose", "x"])
+        self.assertEqual(out, ["--api-key", "--verbose", "x"])
+
+    def test_args_clean_list_byte_identical(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        args = ["-y", "mcp-server", "--port", "8080", "--model", "qwen"]
+        self.assertEqual(scrub_args_secrets(args), args)
+
+    def test_args_trailing_secret_flag_without_value(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        self.assertEqual(scrub_args_secrets(["srv", "--token"]),
+                         ["srv", "--token"])
+
+
+class TestMsAgentOutboundLeakCarriers(unittest.TestCase):
+    """ms-agent upload must not leak url / headers / args carriers.
+
+    ``settings.json`` and ``mcp.json`` carry provider and MCP server
+    definitions; credentials hide in URL query strings / userinfo, in
+    arbitrary-named HTTP headers and in stdio command lines -- none of which
+    a key-name vocabulary can catch.
+    """
+
+    def setUp(self):
+        from ms_agent.agent_hub.frameworks.ms_agent import MsAgentWorkspace
+        self.spec = MsAgentWorkspace(agent_name="default")
+
+    SETTINGS = {
+        "providers": {
+            "dashscope": {
+                "name": "dashscope", "protocol": "openai",
+                "api_key": "sk-LEAK-provider",
+                "base_url": "https://dashscope.aliyuncs.com/compatible/v1",
+                "models": ["qwen3-max"],
+            },
+            "my-gateway": {
+                "name": "gateway", "protocol": "openai",
+                "api_key": "sk-LEAK-gateway",
+                "base_url": "https://gw.example.com/v1?token=LEAK-url-token",
+                "models": [],
+            },
+        },
+        "default_model": "dashscope/qwen3-max",
+    }
+
+    MCP = {
+        "mcpServers": {
+            "remote-http": {
+                "url": "https://api.example.com/msse?api_key=LEAK-url-key",
+                "transport": "sse",
+                "headers": {
+                    "Authorization": "Bearer LEAK-bearer",
+                    "x-api-key": "LEAK-header-key",
+                    "X-Auth-Code": "LEAK-custom-header",
+                },
+                "enabled": True,
+            },
+            "userinfo": {
+                "url": "https://user:LEAK-pass@host.example.com/sse",
+                "transport": "sse",
+            },
+            "local-stdio": {
+                "command": "npx",
+                "args": ["-y", "mcp-server", "--api-key", "LEAK-arg-key",
+                         "--model", "qwen"],
+                "env": {"OPENAI_API_KEY": "LEAK-env-key"},
+                "enabled": True,
+            },
+        },
+    }
+
+    def _out(self, rel, data):
+        raw = json.dumps(data, ensure_ascii=False, indent=2).encode("utf-8")
+        return json.loads(self.spec.sanitize_outbound_file(rel, raw))
+
+    def test_no_leak_tokens_survive_upload(self):
+        for rel, data in (("settings.json", self.SETTINGS),
+                          ("mcp.json", self.MCP)):
+            out = json.dumps(self._out(rel, data))
+            self.assertNotIn("LEAK", out, f"{rel} leaked secrets")
+
+    def test_structure_and_non_secrets_preserved(self):
+        s = self._out("settings.json", self.SETTINGS)
+        self.assertEqual(s["default_model"], "dashscope/qwen3-max")
+        self.assertEqual(s["providers"]["dashscope"]["models"], ["qwen3-max"])
+        # clean URL untouched; secret query param blanked (name kept):
+        self.assertEqual(
+            s["providers"]["dashscope"]["base_url"],
+            "https://dashscope.aliyuncs.com/compatible/v1")
+        self.assertEqual(
+            s["providers"]["my-gateway"]["base_url"],
+            "https://gw.example.com/v1?token=")
+        m = self._out("mcp.json", self.MCP)
+        remote = m["mcpServers"]["remote-http"]
+        self.assertEqual(remote["url"],
+                         "https://api.example.com/msse?api_key=")
+        self.assertEqual(remote["transport"], "sse")
+        self.assertEqual(remote["enabled"], True)
+        # header names kept (structure visible), values wiped:
+        self.assertEqual(set(remote["headers"]),
+                         {"Authorization", "x-api-key", "X-Auth-Code"})
+        self.assertEqual(set(remote["headers"].values()), {""})
+        self.assertEqual(m["mcpServers"]["userinfo"]["url"],
+                         "https://user@host.example.com/sse")
+        stdio = m["mcpServers"]["local-stdio"]
+        self.assertEqual(stdio["command"], "npx")
+        self.assertEqual(stdio["args"],
+                         ["-y", "mcp-server", "--api-key", "",
+                          "--model", "qwen"])
+        self.assertEqual(stdio["env"], {"OPENAI_API_KEY": ""})
+
+    def test_free_text_with_url_not_truncated(self):
+        # Review must-fix: a prompt/description string that merely CONTAINS
+        # a URL (whitespace present) must not be treated as a URL -- the old
+        # behavior cut everything after the query string away.
+        data = {"description": "see https://docs.example.com/g?tokens=abc "
+                               "for the rest of this sentence"}
+        out = self._out("settings.json", data)
+        self.assertEqual(out["description"], data["description"])
+
+    def test_docker_env_args_and_bare_userinfo(self):
+        # Review must-fix: docker-style ``-e NAME=VALUE`` and colon-less
+        # ``https://<token>@host`` userinfo both leaked before.
+        data = {"mcpServers": {
+            "docker": {
+                "command": "docker",
+                "args": ["run", "-i", "--rm", "-e",
+                         "GITHUB_TOKEN=ghp_LEAK", "img"],
+            },
+            "pat": {"url": "https://ghp_LEAKTOKEN@host.example.com/sse"},
+        }}
+        out = self._out("mcp.json", data)
+        dumped = json.dumps(out)
+        self.assertNotIn("LEAK", dumped)
+        self.assertEqual(out["mcpServers"]["docker"]["args"],
+                         ["run", "-i", "--rm", "-e", "GITHUB_TOKEN=", "img"])
+        self.assertEqual(out["mcpServers"]["pat"]["url"],
+                         "https://host.example.com/sse")
+
+    def test_argv_alias_scrubbed_like_args(self):
+        data = {"mcpServers": {"s": {
+            "command": "srv", "argv": ["--token", "LEAK-v"]}}}
+        out = self._out("mcp.json", data)
+        self.assertEqual(out["mcpServers"]["s"]["argv"], ["--token", ""])
+
+    def test_inbound_direction_scrubs_too(self):
+        raw = json.dumps(self.MCP, ensure_ascii=False).encode("utf-8")
+        out = self.spec.sanitize_inbound_file("mcp.json", raw).decode()
+        self.assertNotIn("LEAK", out)
+
+
+class TestQwenpawOutboundLeakCarriers(unittest.TestCase):
+    """qwenpaw agent.json upload must close the same three carriers."""
+
+    def setUp(self):
+        self.spec = QwenpawWorkspace(agent_name="paw_qa_01")
+
+    SRC = json.dumps({
+        "id": "bot",
+        "model": {"api_key": "sk-SECRET-1", "model": "qwen-max"},
+        "mcp": {
+            "clients": {
+                "remote": {
+                    "url": "https://mcp.example.com/sse?api_key=SECRET-URL",
+                    "headers": {"X-Auth-Code": "SECRET-HEADER"},
+                },
+                "stdio": {
+                    "command": "npx",
+                    "args": ["-y", "srv", "--token=SECRET-ARG"],
+                },
+            },
+        },
+    })
+
+    def test_outbound_scrubs_url_headers_args(self):
+        out = self.spec._strip_outbound_agent_json(self.SRC)
+        self.assertNotIn("SECRET", out)
+        data = json.loads(out)
+        remote = data["mcp"]["clients"]["remote"]
+        self.assertEqual(remote["url"], "https://mcp.example.com/sse?api_key=")
+        self.assertEqual(remote["headers"], {"X-Auth-Code": ""})
+        self.assertEqual(data["mcp"]["clients"]["stdio"]["args"],
+                         ["-y", "srv", "--token="])
+        self.assertEqual(data["model"]["model"], "qwen-max")
+
+
+class TestYamlOutboundLeakCarriers(unittest.TestCase):
+    """hermes config.yaml must close the url / headers / args carriers in
+    every legal YAML spelling, without touching clean lines."""
+
+    def _scrub(self, text):
+        from ms_agent.agent_hub._workspace import scrub_yaml_secrets
+        return scrub_yaml_secrets(text)
+
+    CONFIG = (
+        "# hermes config\n"
+        "model: qwen3-max\n"
+        "llm:\n"
+        "  base_url: https://gw.example.com/v1?token=LEAK-url-token\n"
+        "  api_key: LEAK-key\n"
+        "mcp_servers:\n"
+        "  remote:\n"
+        "    url: https://api.example.com/msse?api_key=LEAK-url-key&v=2\n"
+        "    transport: sse\n"
+        "    headers:\n"
+        "      Authorization: Bearer LEAK-bearer\n"
+        "      X-Auth-Code: LEAK-custom-header\n"
+        "    enabled: true\n"
+        "  gateway:\n"
+        "    url: https://user:LEAK-pass@host.example.com/path\n"
+        "  stdio:\n"
+        "    command: npx\n"
+        "    args:\n"
+        "      - -y\n"
+        "      - mcp-server\n"
+        "      - --api-key\n"
+        "      - LEAK-arg-key\n"
+        "      - --model\n"
+        "      - qwen\n"
+        "    env:\n"
+        "      OPENAI_API_KEY: LEAK-env-key\n"
+        "  flow:\n"
+        "    command: uvx\n"
+        "    args: [run, --token, LEAK-flow-arg, --port, \"8080\"]\n"
+        "    headers: {X-Custom: LEAK-flow-header}\n"
+        "plain: value\n"
+    )
+
+    def test_no_leak_tokens_survive(self):
+        out = self._scrub(self.CONFIG)
+        self.assertNotIn("LEAK", out)
+
+    def test_structure_preserved(self):
+        out = self._scrub(self.CONFIG)
+        self.assertIn("model: qwen3-max", out)
+        self.assertIn("plain: value", out)
+        self.assertIn("url: https://api.example.com/msse?api_key=&v=2", out)
+        self.assertIn("url: https://user@host.example.com/path", out)
+        self.assertIn("base_url: https://gw.example.com/v1?token=", out)
+        self.assertIn("Authorization: ''", out)
+        self.assertIn("X-Auth-Code: ''", out)
+        self.assertIn("- -y", out)
+        self.assertIn("- mcp-server", out)
+        self.assertIn("- --api-key", out)
+        self.assertIn("- --model", out)
+        self.assertIn("- qwen", out)
+        self.assertIn("- ''", out)
+        self.assertIn("args: [run, --token, '', --port, \"8080\"]", out)
+        self.assertIn("headers: {X-Custom: ''}", out)
+
+    def test_clean_config_byte_identical(self):
+        clean = (
+            "# comment\n"
+            "model: qwen3-max\n"
+            "mcp_servers:\n"
+            "  remote:\n"
+            "    url: https://api.example.com/msse?v=2\n"
+            "    transport: sse\n"
+            "  stdio:\n"
+            "    args:\n"
+            "      - -y\n"
+            "      - --port\n"
+            "      - \"8080\"\n"
+        )
+        self.assertEqual(self._scrub(clean), clean)
+
+    def test_innocuous_headers_cleared_too(self):
+        """headers is a wholesale secret bag: header names cannot be
+        enumerated (a gateway may demand ``X-Auth-Code``), so even innocuous
+        values like ``Accept`` are deliberately cleared -- fail-closed."""
+        text = ("mcp_servers:\n"
+                "  remote:\n"
+                "    headers:\n"
+                "      Accept: application/json\n")
+        self.assertIn("Accept: ''", self._scrub(text))
+
+    def test_top_level_bags_and_args_scrubbed_anywhere(self):
+        # Review fix: bags/args used to be scoped to the mcp_servers block,
+        # so top-level ones leaked. Policy is now global (JSON/TOML parity).
+        text = ("headers:\n"
+                "  X-Auth-Code: LEAK-header\n"
+                "env:\n"
+                "  RANDOM_NAME: LEAK-env\n"
+                "args:\n"
+                "  - --token\n"
+                "  - LEAK-arg\n")
+        out = self._scrub(text)
+        self.assertNotIn("LEAK", out)
+        self.assertIn("X-Auth-Code: ''", out)
+        self.assertIn("RANDOM_NAME: ''", out)
+        self.assertIn("- ''", out)
+
+    def test_block_opener_with_inline_comment_recognized(self):
+        # ``mcp_servers:  # remote`` must still open the block.
+        text = ("mcp_servers:  # remote tools\n"
+                "  fs:\n"
+                "    headers:\n"
+                "      X-Auth-Code: LEAK\n")
+        out = self._scrub(text)
+        self.assertNotIn("LEAK", out)
+        self.assertIn("mcp_servers:  # remote tools", out)
+
+    def test_url_scalar_trailing_comment_kept_separate(self):
+        # Review fix: the comment used to be glued into the scrubbed value.
+        out = self._scrub("url: https://h.example.com/p?token=X # note\n")
+        self.assertEqual(out, "url: https://h.example.com/p?token= # note\n")
+
+    def test_block_args_env_assignment_blanked(self):
+        text = ("mcp_servers:\n"
+                "  d:\n"
+                "    args:\n"
+                "      - run\n"
+                "      - -e\n"
+                "      - GITHUB_TOKEN=ghp_LEAK\n")
+        out = self._scrub(text)
+        self.assertNotIn("LEAK", out)
+        self.assertIn("- GITHUB_TOKEN=", out)
+
+
+class TestTomlOutboundLeakCarriers(unittest.TestCase):
+    """openhuman config.toml must close the url / headers / args carriers."""
+
+    def _scrub(self, text):
+        from ms_agent.agent_hub.frameworks.openhuman import \
+            OpenhumanWorkspace
+        return OpenhumanWorkspace(
+            agent_name="default")._scrub_toml_secrets(text)
+
+    CONFIG = (
+        "# openhuman config\n"
+        "[model]\n"
+        'provider = "openai"\n'
+        'api_key = "LEAK-key"\n'
+        "[providers.gateway]\n"
+        'base_url = "https://gw.example.com/v1?token=LEAK-url&mode=fast"\n'
+        'endpoint = "https://user:LEAK-pass@ep.example.com/x"\n'
+        "[mcp.remote]\n"
+        "headers = { Authorization = \"Bearer LEAK-b\", X-Custom = "
+        "\"LEAK-c\" }\n"
+        'args = ["-y", "srv", "--api-key", "LEAK-arg", "--port", "8080"]\n'
+        'url = "https://mcp.example.com/sse?api_key=LEAK-u&v=2"\n'
+        # Block table-section spellings (NOT inline tables): a headers / env
+        # section whose inner key names are arbitrary (incl. a quoted key that
+        # the bare-key pattern cannot match).
+        "[mcp.block.headers]\n"
+        'X-Auth-Code = "LEAK-block-header"\n'
+        '"X-Quoted-Key" = "LEAK-quoted-header"\n'
+        "[mcp.block.env]\n"
+        'SOME_RANDOM_NAME = "LEAK-block-env"\n'
+        # Multi-line args array with the secret on a continuation line.
+        "[mcp.spawner]\n"
+        "args = [\n"
+        '  "-y",\n'
+        '  "--token",\n'
+        '  "LEAK-ml-arg",\n'
+        '  "--model",\n'
+        '  "qwen",\n'
+        "]\n"
+    )
+
+    def test_no_leak_tokens_survive(self):
+        self.assertNotIn("LEAK", self._scrub(self.CONFIG))
+
+    def test_structure_preserved(self):
+        out = self._scrub(self.CONFIG)
+        self.assertIn('provider = "openai"', out)
+        self.assertIn('base_url = "https://gw.example.com/v1?token=&mode=fast"',
+                      out)
+        self.assertIn('endpoint = "https://user@ep.example.com/x"', out)
+        self.assertIn("headers = { Authorization = \"\", X-Custom = \"\" }",
+                      out)
+        self.assertIn(
+            'args = ["-y", "srv", "--api-key", "", "--port", "8080"]', out)
+        self.assertIn('url = "https://mcp.example.com/sse?api_key=&v=2"', out)
+        # Block sections: the section headers survive, the inner values don't.
+        self.assertIn("[mcp.block.headers]", out)
+        self.assertIn('X-Auth-Code = ""', out)
+        self.assertIn('"X-Quoted-Key" = ""', out)
+        self.assertIn("[mcp.block.env]", out)
+        self.assertIn('SOME_RANDOM_NAME = ""', out)
+        # Multi-line args collapsed to one line with the secret value blanked.
+        self.assertIn('args = ["-y", "--token", "", "--model", "qwen"]', out)
+
+    def test_clean_config_byte_identical(self):
+        # No bag section here: [...headers] / [...env] are fail-closed and
+        # blank every value even when innocuous (same as the YAML scrubber).
+        clean = (
+            "# comment\n"
+            'name = "bot"\n'
+            'base_url = "https://gw.example.com/v1?mode=fast"\n'
+            'args = ["-y", "srv", "--port", "8080"]\n'
+            "[mcp.spawner]\n"
+            'command = "srv"\n'
+            "args = [\n"
+            '  "-y",\n'
+            '  "--port",\n'
+            '  "8080",\n'
+            "]\n"
+        )
+        self.assertEqual(self._scrub(clean), clean)
+
+    def test_toml_dotted_key_through_bag(self):
+        # ``mcp.fs.headers.X = v`` is the same as living under
+        # ``[mcp.fs.headers]`` -- the dotted path must trigger the bag rule.
+        out = self._scrub('mcp.fs.headers.X-Auth-Code = "LEAK"\n')
+        self.assertEqual(out, 'mcp.fs.headers.X-Auth-Code = ""\n')
+
+    def test_toml_quoted_url_trailing_comment(self):
+        # Review fix: 'URL' # note used to pass through untouched.
+        out = self._scrub('url = "https://h.example.com/p?token=LEAK" # x\n')
+        self.assertEqual(out,
+                         'url = "https://h.example.com/p?token=" # x\n')
+
+    def test_toml_argv_alias_scrubbed(self):
+        out = self._scrub('argv = ["--token", "LEAK"]\n')
+        self.assertEqual(out, 'argv = ["--token", ""]\n')
+
+
+class TestLeakCarrierEdgeCases(unittest.TestCase):
+    """Tricky-but-legal shapes beyond the canonical test configs.
+
+    Generated from an edge-case probe: every assertion locks behavior that was
+    manually verified correct, so a future refactor cannot silently regress a
+    corner nobody looked at (fragment handling, quoting, scoping, word
+    boundaries...).
+    """
+
+    # ---- shared URL / args helpers -----------------------------------
+
+    def test_url_userinfo_and_query_combined(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        self.assertEqual(
+            scrub_url_secrets(
+                "https://user:pa55@h.example.com/p?api_key=X&v=1"),
+            "https://user@h.example.com/p?api_key=&v=1")
+
+    def test_url_no_path_query_only(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        self.assertEqual(
+            scrub_url_secrets("https://host.example.com?token=X"),
+            "https://host.example.com?token=")
+
+    def test_url_ws_scheme(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        self.assertEqual(
+            scrub_url_secrets("ws://mcp.local/sse?api_key=X"),
+            "ws://mcp.local/sse?api_key=")
+
+    def test_url_fragment_preserved_when_query_stripped(self):
+        from ms_agent.agent_hub._workspace import scrub_url_secrets
+        self.assertEqual(
+            scrub_url_secrets("https://h.example.com/p?token=X#frag=1"),
+            "https://h.example.com/p?token=#frag=1")
+
+    def test_args_chained_secret_flags(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        self.assertEqual(
+            scrub_args_secrets(["--api-key", "A", "--token", "B"]),
+            ["--api-key", "", "--token", ""])
+
+    def test_args_word_boundary_no_false_positive(self):
+        # ``tokenizer`` / ``keymap`` end in secret-ish substrings but are NOT
+        # secret flags -- the vocabulary anchors on [_-] boundaries + end.
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        args = ["--tokenizer", "cl100k", "--keymap", "vim"]
+        self.assertEqual(scrub_args_secrets(args), args)
+
+    def test_args_non_string_values_pass_through(self):
+        from ms_agent.agent_hub._workspace import scrub_args_secrets
+        self.assertEqual(
+            scrub_args_secrets(["--api-key", 123, "-y"]),
+            ["--api-key", 123, "-y"])
+
+    # ---- JSON (ms-agent) ---------------------------------------------
+
+    def test_json_headers_inside_list_item(self):
+        from ms_agent.agent_hub._workspace import scrub_json_secrets
+        data = {"servers": [{"headers": {"X-Auth-Code": "LEAK"}}]}
+        scrub_json_secrets(data)
+        self.assertEqual(data, {"servers": [{"headers": {"X-Auth-Code": ""}}]})
+
+    def test_json_dict_nested_in_args_list(self):
+        from ms_agent.agent_hub._workspace import scrub_json_secrets
+        data = {"args": [{"env": {"K": "LEAK"}}]}
+        scrub_json_secrets(data)
+        self.assertEqual(data, {"args": [{"env": {"K": ""}}]})
+
+    def test_json_uppercase_scheme_and_params(self):
+        from ms_agent.agent_hub._workspace import scrub_json_secrets
+        data = {"u": "HTTPS://HOST.example.com/p?TOKEN=LEAK&v=1"}
+        scrub_json_secrets(data)
+        self.assertEqual(data, {"u": "HTTPS://HOST.example.com/p?TOKEN=&v=1"})
+
+    def test_json_empty_value_param_before_fragment(self):
+        from ms_agent.agent_hub._workspace import scrub_json_secrets
+        data = {"u": "https://h.example.com/p?api_key=#frag"}
+        scrub_json_secrets(data)
+        self.assertEqual(data, {"u": "https://h.example.com/p?api_key=#frag"})
+
+    # ---- YAML (hermes) -----------------------------------------------
+
+    def _yscrub(self, text):
+        from ms_agent.agent_hub._workspace import scrub_yaml_secrets
+        return scrub_yaml_secrets(text)
+
+    def test_yaml_quoted_url_scalar(self):
+        out = self._yscrub(
+            'url: "https://host.example.com/sse?api_key=LEAK"\n')
+        self.assertEqual(out, 'url: "https://host.example.com/sse?api_key="\n')
+
+    def test_yaml_url_fragment_preserved(self):
+        out = self._yscrub(
+            "mcp_servers:\n"
+            "  fs:\n"
+            "    url: https://h.example.com/p?token=LEAK#frag=1\n")
+        self.assertIn("url: https://h.example.com/p?token=#frag=1", out)
+
+    def test_yaml_args_outside_mcp_block_scrubbed_globally(self):
+        # Bags and args lists are scrubbed at ANY depth now (JSON/TOML
+        # parity) -- a top-level args list gets the same positional scrub.
+        text = "args:\n  - --api-key\n  - VALUE\n"
+        self.assertEqual(self._yscrub(text), "args:\n  - --api-key\n  - ''\n")
+
+    # ---- TOML (openhuman) --------------------------------------------
+
+    def _tscrub(self, text):
+        from ms_agent.agent_hub.frameworks.openhuman import \
+            OpenhumanWorkspace
+        return OpenhumanWorkspace(
+            agent_name="default")._scrub_toml_secrets(text)
+
+    def test_toml_array_of_tables_bag(self):
+        out = self._tscrub("[[mcp.fs.headers]]\n"
+                           'X-Auth-Code = "LEAK"\n')
+        self.assertIn("[[mcp.fs.headers]]", out)
+        self.assertIn('X-Auth-Code = ""', out)
+        self.assertNotIn("LEAK", out)
+
+    def test_toml_section_trailing_comment(self):
+        out = self._tscrub("[mcp.fs.headers]  # auth bag\n"
+                           'X-Auth-Code = "LEAK"\n')
+        self.assertNotIn("LEAK", out)
+        self.assertIn('X-Auth-Code = ""', out)
+
+    def test_toml_multiline_array_in_bag_section(self):
+        out = self._tscrub("[mcp.fs.env]\n"
+                           "vals = [\n"
+                           '  "LEAK",\n'
+                           "]\n")
+        self.assertNotIn("LEAK", out)
+        self.assertIn("vals = []", out)
+
+    def test_toml_single_quoted_url(self):
+        out = self._tscrub(
+            "base_url = 'https://h.example.com/v1?token=LEAK'\n")
+        self.assertEqual(out, "base_url = 'https://h.example.com/v1?token='\n")
+
+
 if __name__ == "__main__":
     unittest.main()