Runtime robustness: shell execution, safety policy, permission asks, MCP startup, and tool-call handling

.gitignore

@@ -132,6 +132,7 @@ venv.bak/
 .firecrawl
 .claude/
 .claude-trace/
+.qoder
 
 # custom
 *.pkl

ms_agent/agent/llm_agent.py

@@ -444,6 +444,10 @@ def _build_system_content(self) -> str:
                 content += '\n\n' + LIVE_FILES_HINT.format(
                     home=str(global_home()))
 
+        internals = self._build_workspace_internals_section()
+        if internals:
+            content += '\n\n' + internals
+
         if self._memory_guidance:
             content += '\n\n' + self._memory_guidance
 
@@ -461,6 +465,46 @@ def _build_system_content(self) -> str:
 
         return content
 
+    def _build_workspace_internals_section(self) -> str:
+        """Describe the framework's own directories, when they sit in the
+        working directory.
+
+        Only for the layout where they do. A project opened from an existing
+        folder keeps its records elsewhere, and telling that agent to watch out
+        for a ``sessions/`` directory it will never encounter would be a
+        fabricated warning.
+        """
+        from ms_agent.prompting.builtin import WORKSPACE_INTERNALS_HINT
+        from ms_agent.utils.workspace_context import resolve_workspace_root
+
+        try:
+            workspace_root = Path(resolve_workspace_root(self.config))
+        except Exception:  # noqa: BLE001 - never break prompt assembly
+            return ''
+        if not (workspace_root / 'sessions').is_dir():
+            return ''
+
+        # The directory the log is actually writing to, not the agent's tag:
+        # naming a path that does not exist is worse than naming none, since
+        # the model will go looking for it.
+        session_dir = None
+        log = getattr(self, 'session_log', None)
+        directory = getattr(log, 'directory', None)
+        if directory is not None:
+            session_dir = Path(directory).name
+        if not session_dir:
+            session_dir = getattr(self.runtime, 'session_id', None)
+
+        try:
+            home = str(global_home())
+        except Exception:  # noqa: BLE001
+            home = '~/.ms_agent'
+        hint = WORKSPACE_INTERNALS_HINT.format(
+            session_line=(f' This conversation is `sessions/{session_dir}/`.'
+                          if session_dir else ''),
+            home=home)
+        return hint
+
     def _check_skill_tool_dependencies(self):
         """Warn if skills are enabled but essential tools are missing."""
         if (not self._skill_catalog
@@ -858,10 +902,8 @@ def _build_permission_objects(self):
         perm_config = PermissionConfig.from_dict(
             raw, project_root=workspace_root)
 
-        allowed_dirs = [workspace_root]
-        for directory in perm_config.safety.allowed_directories:
-            if directory not in allowed_dirs:
-                allowed_dirs.append(directory)
+        allowed_dirs = list(
+            perm_config.safety.effective_allowed_directories(workspace_root))
         read_only_dirs = list(perm_config.safety.read_only_directories)
         safety_guard = SafetyGuard(
             config=perm_config.safety,

ms_agent/permission/ask_resolver.py

@@ -19,9 +19,31 @@
     'command_validator': 'deny',
     'shell_expansion': 'deny',
     'read_outside_dirs': 'deny',
+    # Running code is what full access is FOR. Refusing every `python3 -c` in
+    # the mode whose whole meaning is "stop asking me" would make the mode
+    # useless, and the confirmation this category exists for is the one an
+    # interactive user gets. What auto mode cannot honestly claim is that it
+    # inspected the code — hence the message on the ask, and the setting-page
+    # copy that says so.
+    'interpreter_exec': 'allow',
+    # A private key does not become less private because the user is not
+    # watching. There is no path here that reads it without someone deciding.
+    'sensitive_read': 'deny',
 }
 
 
+#: Safety confirmations a standing answer may satisfy.
+#:
+#: The rest are deliberately not here. "This reads a private key" has to be
+#: decided each time, because what makes it risky is the specific file, and a
+#: pattern broad enough to remember would cover files the user never saw.
+#: "This runs code I cannot analyse" is different: every inline `python3 -c` is
+#: the same decision, it comes up constantly, and an ask the user cannot settle
+#: is one they answer by turning confirmations off — which is the outcome the
+#: confirmation existed to prevent.
+REMEMBERABLE_ASK_CATEGORIES: frozenset = frozenset({'interpreter_exec'})
+
+
 def resolve_ask(
     decision: SafetyDecision,
     mode: str,

ms_agent/permission/config.py

@@ -7,6 +7,7 @@
 from __future__ import annotations
 
 import os
+from pathlib import Path
 from dataclasses import dataclass
 from typing import Any, Literal
 
@@ -33,6 +34,30 @@
     '**/.git/**',
 )
 
+#: Locations whose CONTENTS are a credential. Kept apart from
+#: ``sensitive_paths`` because the two lists answer different questions:
+#: that one protects things from being CHANGED (``.git/config``,
+#: ``~/.bashrc``), which is no reason to refuse reading them — an agent that
+#: cannot run ``cat .git/config`` to find a remote is just broken. These are
+#: things that must not be COPIED, most of all into a transcript that gets
+#: written to disk and replayed to a model.
+#:
+#: Patterns are fnmatch, where ``*`` crosses ``/``, so a leading ``*/`` covers
+#: any user's home rather than only the one the process happens to run as.
+_DEFAULT_SENSITIVE_READ_PATHS: tuple[str, ...] = (
+    '*/.ssh/*',
+    '*/.gnupg/*',
+    '*/.aws/*',
+    '*/.kube/config',
+    '*/.docker/config.json',
+    '*/.netrc',
+    '*/id_rsa',
+    '*/id_dsa',
+    '*/id_ecdsa',
+    '*/id_ed25519',
+    '*/*.pem',
+)
+
 _DEFAULT_DANGEROUS_REMOVAL: tuple[str, ...] = (
     '*',
     '/*',
@@ -41,23 +66,54 @@
 )
 
 
+def default_temp_directories() -> tuple[str, ...]:
+    """Scratch directories the agent may write to besides its workspace.
+
+    Refusing these buys nothing: the OS temp directory is world-writable by
+    design and holds nothing to protect. It costs plenty, though — every tool
+    that stages output through a temp file, and every ``python3 - <<EOF`` the
+    agent would rather write to a real file first, comes back refused. Both
+    spellings are listed because ``TMPDIR`` and a literal ``/tmp`` are
+    different paths on macOS, and either may be what a command actually uses.
+
+    Turn off with ``permission.safety_rules.allow_temp_dir: false``.
+    """
+    import tempfile
+
+    out: list[str] = []
+    for raw in (tempfile.gettempdir(), '/tmp'):
+        if not raw:
+            continue
+        try:
+            resolved = str(Path(raw).resolve())
+        except OSError:
+            continue
+        if resolved not in out:
+            out.append(resolved)
+    return tuple(out)
+
+
 @dataclass(frozen=True)
 class SafetyConfig:
     """Inner-layer safety configuration (non-bypassable)."""
     patterns: tuple[str, ...] = _DEFAULT_SAFETY_PATTERNS
     sensitive_paths: tuple[str, ...] = _DEFAULT_SENSITIVE_PATHS
+    sensitive_read_paths: tuple[str, ...] = _DEFAULT_SENSITIVE_READ_PATHS
     dangerous_removal_paths: tuple[str, ...] = _DEFAULT_DANGEROUS_REMOVAL
     read_policy: Literal['loose', 'strict'] = 'loose'
     max_command_chars: int = 8192
     allowed_directories: tuple[str, ...] = ()
     read_only_directories: tuple[str, ...] = ()
+    allow_temp_dir: bool = True
 
     @classmethod
     def from_dict(cls,
                   d: dict[str, Any],
                   project_root: str | None = None) -> SafetyConfig:
         patterns = tuple(d.get('patterns', _DEFAULT_SAFETY_PATTERNS))
         sensitive = tuple(d.get('sensitive_paths', _DEFAULT_SENSITIVE_PATHS))
+        sensitive_read = tuple(
+            d.get('sensitive_read_paths', _DEFAULT_SENSITIVE_READ_PATHS))
         dangerous = tuple(
             d.get('dangerous_removal_paths', _DEFAULT_DANGEROUS_REMOVAL))
 
@@ -80,13 +136,28 @@ def _expand_dirs(raw: list[str]) -> tuple[str, ...]:
         return cls(
             patterns=patterns,
             sensitive_paths=sensitive,
+            sensitive_read_paths=sensitive_read,
             dangerous_removal_paths=dangerous,
             read_policy=read_policy,
             max_command_chars=max_chars,
             allowed_directories=allowed,
             read_only_directories=read_only,
+            allow_temp_dir=bool(d.get('allow_temp_dir', True)),
         )
 
+    def effective_allowed_directories(
+            self, workspace_root: str) -> tuple[str, ...]:
+        """Every directory writes are permitted in, workspace root first."""
+        out = [workspace_root]
+        for directory in self.allowed_directories:
+            if directory not in out:
+                out.append(directory)
+        if self.allow_temp_dir:
+            for directory in default_temp_directories():
+                if directory not in out:
+                    out.append(directory)
+        return tuple(out)
+
 
 #: Nothing by default. A blacklist entry can never be overridden — not by the
 #: mode, not by a whitelist, not by the user answering a prompt — so it is the

ms_agent/permission/enforcer.py

@@ -17,13 +17,27 @@
 from .matcher import CONTENT_SEP, PermissionMatcher
 from .memory import PermissionMemory
 from .suggestions import generate_suggestions
+from ms_agent.utils import get_logger
+
+logger = get_logger()
 
 
 @dataclass(frozen=True)
 class PermissionDecision:
     action: Literal['allow', 'deny', 'ask']
     reason: str
     updated_args: dict[str, Any] | None = None
+    #: Whether a standing answer may satisfy this confirmation next time.
+    #:
+    #: Safety confirmations default to False — a remembered answer must not
+    #: stand in for looking at THIS call. But that is not true of all of them
+    #: equally. "This command runs code I cannot analyse" is a thing a user can
+    #: reasonably decide once for a project, the way they decide about `git`;
+    #: "this reads a private key" is not. Marking the first kind rememberable
+    #: is what keeps the confirmation useful — an ask that reappears no matter
+    #: how the user answers it does not make anyone safer, it just teaches them
+    #: to turn confirmations off entirely.
+    rememberable: bool = False
 
 
 class PermissionEnforcer:
@@ -77,6 +91,11 @@ async def _ask_user(self,
         if 'call_id' in kwargs and not self._handler_accepts('call_id'):
             kwargs.pop('call_id')
         if getattr(self._handler, 'supports_concurrent_asks', False):
+            # A handler that services asks concurrently needs to know which of
+            # them are safety confirmations, so a remembered answer is never
+            # applied to one. Older handlers with a fixed signature don't.
+            if self._handler_accepts('forced'):
+                return await self._handler.ask(forced=forced, **kwargs)
             return await self._handler.ask(**kwargs)
         async with self._ask_lock_for_loop():
             if not forced and self._memory.matches(kwargs['tool_name'],
@@ -125,9 +144,15 @@ async def check(
                 )
 
         if force_decision and force_decision.action == 'ask':
+            rememberable = getattr(force_decision, 'rememberable', False)
+            if rememberable and self._memory.matches(tool_name, tool_args):
+                return PermissionDecision(
+                    action='allow',
+                    reason='Allowed by remembered permission',
+                )
             suggestions = generate_suggestions(tool_name, tool_args)
             response = await self._ask_user(
-                forced=True,
+                forced=not rememberable,
                 tool_name=tool_name,
                 tool_args=tool_args,
                 context=force_decision.reason or '',
@@ -209,6 +234,30 @@ def _remember_pattern(self, response: PermissionResponse, tool_name: str,
                 return s
         return tool_name
 
+    def _release_asks_covered_by_memory(self, pattern: str) -> int:
+        """Apply a just-remembered answer to the other cards still on screen.
+
+        Only reaches handlers that show several cards at once. There, "always
+        allow" is a statement about a pattern, so re-asking about a sibling the
+        pattern covers asks a question the user has answered. It also stops
+        mattering only in one direction: since a wait has no deadline, an
+        unanswered sibling now holds the turn open instead of being denied
+        after a couple of minutes, so leaving them up turns a mis-set
+        expectation into a stuck conversation.
+        """
+        resolver = getattr(self._handler, 'resolve_matching', None)
+        if resolver is None:
+            return 0
+        released = resolver(
+            lambda name, args: self._memory.matches(name, args),
+            PermissionResponse(action=PermissionAction.ALLOW_ONCE),
+        )
+        if released:
+            logger.info(
+                'permission pattern %r also released %d waiting request(s)',
+                pattern, released)
+        return released
+
     def _process_response(
         self,
         response: PermissionResponse | None,
@@ -230,6 +279,7 @@ def _process_response(
         if response.action == PermissionAction.ALLOW_SESSION:
             pattern = self._remember_pattern(response, tool_name, tool_args)
             self._memory.add_session(pattern)
+            self._release_asks_covered_by_memory(pattern)
             return PermissionDecision(
                 action='allow',
                 reason=f'User allowed for session (pattern: {pattern})',
@@ -238,6 +288,7 @@ def _process_response(
         if response.action == PermissionAction.ALLOW_ALWAYS:
             pattern = self._remember_pattern(response, tool_name, tool_args)
             self._memory.add(pattern, scope='project', source='user')
+            self._release_asks_covered_by_memory(pattern)
             return PermissionDecision(
                 action='allow',
                 reason=f'User allowed always (pattern: {pattern})',