build(container): pin gitleaks instead of resolving the latest release

Co-authored-by: Ahmed Allam <ahmed39652003@gmail.com>

.github/workflows/sandbox-image.yml

@@ -0,0 +1,112 @@
+name: Sandbox Image
+
+on:
+  workflow_dispatch:
+    inputs:
+      tag:
+        description: 'Image tag to publish (e.g. 1.2.0)'
+        required: true
+      latest:
+        description: 'Also tag as latest'
+        type: boolean
+        default: true
+
+permissions:
+  contents: read
+
+env:
+  IMAGE: ghcr.io/${{ github.repository_owner }}/strix-sandbox
+
+jobs:
+  build:
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          - os: ubuntu-22.04
+            platform: linux/amd64
+          - os: ubuntu-22.04-arm
+            platform: linux/arm64
+
+    runs-on: ${{ matrix.os }}
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1  # v7.0.1
+        with:
+          persist-credentials: false
+
+      - uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c  # v4.2.0
+
+      - uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f  # v4.6.0
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Build and push by digest
+        id: build
+        uses: docker/build-push-action@53b7df96c91f9c12dcc8a07bcb9ccacbed38856a  # v7.3.0
+        with:
+          context: .
+          file: containers/Dockerfile
+          platforms: ${{ matrix.platform }}
+          provenance: mode=max
+          sbom: true
+          outputs: type=image,name=${{ env.IMAGE }},push-by-digest=true,name-canonical=true,push=true
+
+      - name: Export digest
+        env:
+          DIGEST: ${{ steps.build.outputs.digest }}
+        run: |
+          set -euo pipefail
+          mkdir -p /tmp/digests
+          touch "/tmp/digests/${DIGEST#sha256:}"
+
+      - uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a  # v7.0.1
+        with:
+          name: digest-${{ runner.arch }}
+          path: /tmp/digests/*
+          if-no-files-found: error
+          retention-days: 1
+
+  publish:
+    needs: build
+    runs-on: ubuntu-latest
+    permissions:
+      contents: read
+      packages: write
+
+    steps:
+      - uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c  # v8.0.1
+        with:
+          path: /tmp/digests
+          pattern: digest-*
+          merge-multiple: true
+
+      - uses: docker/setup-buildx-action@bb05f3f5519dd87d3ba754cc423b652a5edd6d2c  # v4.2.0
+
+      - uses: docker/login-action@dbcb813823bdd20940b903addbd779551569679f  # v4.6.0
+        with:
+          registry: ghcr.io
+          username: ${{ github.actor }}
+          password: ${{ secrets.GITHUB_TOKEN }}
+
+      - name: Create manifest list
+        env:
+          TAG: ${{ inputs.tag }}
+          ALSO_LATEST: ${{ inputs.latest }}
+        run: |
+          set -euo pipefail
+          tags=(-t "${IMAGE}:${TAG}")
+          if [ "${ALSO_LATEST}" = "true" ]; then
+            tags+=(-t "${IMAGE}:latest")
+          fi
+          digests=()
+          for file in /tmp/digests/*; do
+            digests+=("${IMAGE}@sha256:$(basename "$file")")
+          done
+          docker buildx imagetools create "${tags[@]}" "${digests[@]}"
+          docker buildx imagetools inspect "${IMAGE}:${TAG}"

containers/Dockerfile

@@ -162,6 +162,7 @@ USER root
 ARG TRUFFLEHOG_VERSION=3.95.9
 RUN curl -sSfL https://raw.githubusercontent.com/trufflesecurity/trufflehog/main/scripts/install.sh | sh -s -- -b /home/pentester/.local/bin "v${TRUFFLEHOG_VERSION}" && \
     chown -R pentester:pentester /home/pentester/.local
+ARG GITLEAKS_VERSION=8.30.1
 RUN set -eux; \
     ARCH="$(uname -m)"; \
     case "$ARCH" in \
@@ -169,8 +170,7 @@ RUN set -eux; \
         aarch64|arm64) GITLEAKS_ARCH="arm64" ;; \
         *) echo "Unsupported architecture: $ARCH" >&2; exit 1 ;; \
     esac; \
-    TAG="$(curl -fsSL https://api.github.com/repos/gitleaks/gitleaks/releases/latest | jq -r .tag_name)"; \
-    curl -fsSL "https://github.com/gitleaks/gitleaks/releases/download/${TAG}/gitleaks_${TAG#v}_linux_${GITLEAKS_ARCH}.tar.gz" -o /tmp/gitleaks.tgz; \
+    curl -fsSL "https://github.com/gitleaks/gitleaks/releases/download/v${GITLEAKS_VERSION}/gitleaks_${GITLEAKS_VERSION}_linux_${GITLEAKS_ARCH}.tar.gz" -o /tmp/gitleaks.tgz; \
     tar -xzf /tmp/gitleaks.tgz -C /tmp; \
     install -m 0755 /tmp/gitleaks /usr/local/bin/gitleaks; \
     rm -f /tmp/gitleaks /tmp/gitleaks.tgz

containers/docker-entrypoint.sh

@@ -1,6 +1,22 @@
 #!/bin/bash
 set -e
 
+if [ -n "${STRIX_HOST_UID:-}" ] && [ "${STRIX_HOST_UID}" != "0" ] && [ "${STRIX_HOST_UID}" != "$(id -u)" ]; then
+  exec sudo -E -- bash -c '
+    set -e
+    gid="${STRIX_HOST_GID:-$STRIX_HOST_UID}"
+    old_uid="$1"
+    old_gid="$2"
+    export PATH="$3"
+    shift 3
+    sed -i "s|^pentester:x:${old_uid}:${old_gid}:|pentester:x:${STRIX_HOST_UID}:${gid}:|" /etc/passwd
+    sed -i "s|^pentester:x:${old_gid}:|pentester:x:${gid}:|" /etc/group
+    chown -R "${STRIX_HOST_UID}:${gid}" /home/pentester /app/certs
+    chown "${STRIX_HOST_UID}:${gid}" /workspace
+    exec setpriv --reuid "${STRIX_HOST_UID}" --regid "${gid}" --init-groups "$0" "$@"
+  ' "$0" "$(id -u)" "$(id -g)" "$PATH" "$@"
+fi
+
 CAIDO_PORT=48080
 CAIDO_LOG="/tmp/caido_startup.log"
 

docs/advanced/configuration.mdx

@@ -106,7 +106,7 @@ When remote vars are set, Strix dual-writes telemetry to both local JSONL and th
 
 ## Docker Configuration
 
-<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.0.0" type="string">
+<ParamField path="STRIX_IMAGE" default="ghcr.io/usestrix/strix-sandbox:1.2.0" type="string">
   Docker image to use for the sandbox container.
 </ParamField>
 
@@ -118,10 +118,6 @@ When remote vars are set, Strix dual-writes telemetry to both local JSONL and th
   Runtime backend for the sandbox environment.
 </ParamField>
 
-<ParamField path="STRIX_MAX_LOCAL_COPY_MB" default="1024" type="integer">
-  Maximum size (in MB) of a local directory target that Strix will copy into the sandbox file-by-file. Larger targets exit early with a suggestion to use `--mount` instead. Set to `0` to disable the check.
-</ParamField>
-
 ## Sandbox Configuration
 
 <ParamField path="STRIX_SANDBOX_EXECUTION_TIMEOUT" default="120" type="integer">

docs/usage/cli.mdx

@@ -6,33 +6,23 @@ description: "Command-line options for Strix"
 ## Basic Usage
 
 ```bash
-strix (--target <target> | --target-list <path> | --mount <path>) [options]
+strix (--target <target> | --target-list <path>) [options]
 ```
 
 ## Options
 
 <ParamField path="--target, -t" type="string">
-  Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times. Fresh runs require at least one target source: `--target`, `--target-list`, or `--mount`.
+  Target to test. Accepts URLs, repositories, local directories, domains, or IP addresses. Can be specified multiple times. Fresh runs require at least one target source: `--target` or `--target-list`.
+
+  <Note>
+    A local directory is mounted into the sandbox live and **writable**, so the agent edits your real files (`.git` excepted). Commit or stash first.
+  </Note>
 </ParamField>
 
 <ParamField path="--target-list" type="string">
   Path to a file containing targets, one per non-empty, non-comment line. Lines starting with `#` are ignored. Can be specified multiple times and combined with `--target`.
 </ParamField>
 
-<ParamField path="--mount" type="string">
-  Bind-mount a local directory into the sandbox (read-only) instead of copying it in file-by-file. Use this for large repositories that are too big to stream into the container. Can be specified multiple times.
-
-  Strix copies local `--target` directories into the sandbox one file at a time, which stalls on very large trees. When a local target exceeds the copy limit (see `STRIX_MAX_LOCAL_COPY_MB`, default 1024 MB) Strix exits early and asks you to re-run with `--mount`.
-
-  <Note>
-    The mount is read-only to protect your source from accidental modification. This is not a hard security boundary: a root process inside the container can remount it writable, so treat `--mount` as "scan my own code", not as isolation from untrusted code.
-  </Note>
-
-  <Note>
-    The size pre-flight only covers local directory targets. Remote repositories (cloned at scan time) are not size-checked.
-  </Note>
-</ParamField>
-
 <ParamField path="--instruction" type="string">
   Custom instructions for the scan. Use for credentials, focus areas, or specific testing approaches.
 </ParamField>
@@ -140,9 +130,6 @@ strix -t https://github.com/org/app -t https://staging.example.com
 
 # Targets from a file
 strix --target-list ./targets.txt
-
-# Large local repository — bind-mount instead of copying it in
-strix --mount ./huge-monorepo
 ```
 
 ## Exit Codes

scripts/install.sh

@@ -4,7 +4,7 @@ set -euo pipefail
 
 APP=strix
 REPO="usestrix/strix"
-STRIX_IMAGE="ghcr.io/usestrix/strix-sandbox:1.1.0"
+STRIX_IMAGE="ghcr.io/usestrix/strix-sandbox:1.2.0"
 
 MUTED='\033[0;2m'
 RED='\033[0;31m'

strix/config/settings.py

@@ -97,15 +97,10 @@ class RuntimeSettings(BaseSettings):
     model_config = _BASE_CONFIG
 
     image: str = Field(
-        default="ghcr.io/usestrix/strix-sandbox:1.1.0",
+        default="ghcr.io/usestrix/strix-sandbox:1.2.0",
         alias="STRIX_IMAGE",
     )
     backend: str = Field(default="docker", alias="STRIX_RUNTIME_BACKEND")
-    # Hard cap on a local target's size before we refuse to stream it into the
-    # sandbox file-by-file (the SDK copies every file individually, which stalls
-    # on large repos). Above this, the user must bind-mount via ``--mount``.
-    # Set to 0 (or less) to disable the pre-flight check entirely.
-    max_local_copy_mb: int = Field(default=1024, alias="STRIX_MAX_LOCAL_COPY_MB")
     # Max screenshot/image tool outputs kept live per agent context (0 = none).
     max_context_images: int = Field(default=3, ge=0, alias="STRIX_MAX_CONTEXT_IMAGES")
 

strix/core/agents.py

@@ -55,6 +55,7 @@ class AgentCoordinator:
         self.idle_resume_counts: dict[str, int] = {}
         self.wait_kinds: dict[str, WaitKind] = {}
         self.runtimes: dict[str, AgentRuntime] = {}
+        self._parent_notified: set[str] = set()
         self._lock = asyncio.Lock()
         self._snapshot_path: Path | None = None
         self.is_shutting_down = False
@@ -191,6 +192,7 @@ class AgentCoordinator:
                 self.errors.pop(agent_id, None)
                 self.wait_kinds.pop(agent_id, None)
                 self.runtimes.setdefault(agent_id, AgentRuntime()).user_wake_required = False
+                self._parent_notified.discard(agent_id)
         await self._maybe_snapshot()
 
     async def park_waiting(self, agent_id: str, *, wait_kind: WaitKind) -> None:
@@ -252,12 +254,27 @@ class AgentCoordinator:
                 self.errors[agent_id] = error
             elif status == "running":
                 self.errors.pop(agent_id, None)
+            if status == "running":
+                # Running again means a fresh stint that owes its parent its own notice.
+                self._parent_notified.discard(agent_id)
             runtime = self.runtimes.setdefault(agent_id, AgentRuntime())
             runtime.user_wake_required = status in {"failed", "crashed"}
             runtime.wake.set()
         logger.info("agent.status %s=%s", agent_id, status)
         await self._maybe_snapshot()
 
+    async def claim_parent_notice(self, agent_id: str) -> bool:
+        """Reserve the one notice a child owes its parent when it stops running.
+
+        A completion report and a terminal notice carry the same information, so
+        whichever comes first claims the slot and the other is skipped.
+        """
+        async with self._lock:
+            if agent_id in self._parent_notified:
+                return False
+            self._parent_notified.add(agent_id)
+            return True
+
     async def send(
         self, target_agent_id: str, message: dict[str, Any], *, interrupt: bool = True
     ) -> bool:
@@ -366,12 +383,15 @@ class AgentCoordinator:
         if tasks:
             await asyncio.gather(*tasks, return_exceptions=True)
 
-    async def cancel_descendants_graceful(self, agent_id: str) -> None:
+    async def cancel_descendants_graceful(self, agent_id: str) -> list[str]:
+        """Stop a subtree leaves-first and report which agents were stopped."""
         async with self._lock:
             order = self._subtree_order_locked(agent_id)
-        for aid in reversed(order):
+        stopped = list(reversed(order))
+        for aid in stopped:
             await self.request_stop(aid)
         await self._maybe_snapshot()
+        return stopped
 
     async def attach_stream(
         self,

strix/core/execution.py

@@ -539,7 +539,7 @@ async def _exhausted_recovery(
     """
     if not interactive:
         await coordinator.set_status(agent_id, "crashed")
-        await _notify_parent_on_terminal(coordinator, agent_id, "crashed")
+        await notify_parent_on_terminal(coordinator, agent_id, "crashed")
         raise MaxTurnsExceeded(
             "Agent exhausted recovery attempts without calling finish_scan or agent_finish."
         )
@@ -622,7 +622,7 @@ async def _run_cycle_parked(
     except Exception as exc:
         logger.exception("error escaped the run cycle for %s; parking as failed", agent_id)
         await coordinator.set_status(agent_id, "failed", error=str(exc) or type(exc).__name__)
-        await _notify_parent_on_terminal(coordinator, agent_id, "failed")
+        await notify_parent_on_terminal(coordinator, agent_id, "failed")
         return None
 
 
@@ -778,7 +778,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
             if isinstance(exc, ProviderRefusalError):
                 logger.warning("agent %s refused by the model provider: %s", agent_id, exc)
                 await coordinator.set_status(agent_id, "failed", error=str(exc))
-                await _notify_parent_on_terminal(coordinator, agent_id, "failed")
+                await notify_parent_on_terminal(coordinator, agent_id, "failed")
                 return None
             if not interactive:
                 raise
@@ -790,7 +790,7 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
                 status = "crashed"
             logger.exception("agent run failed for %s; parking as %s", agent_id, status)
             await coordinator.set_status(agent_id, status, error=str(exc) or type(exc).__name__)
-            await _notify_parent_on_terminal(coordinator, agent_id, status)
+            await notify_parent_on_terminal(coordinator, agent_id, status)
             return None
         else:
             return cast("RunResultBase | None", stream)
@@ -851,6 +851,11 @@ async def _append_tool_required_message(
 
 
 _TERMINAL_NOTICE = {
+    "completed": (
+        "[Agent completed] {name} ({agent_id}) finished and is no longer running, but it "
+        "sent no completion report. Stop waiting on this child; ask it directly if you "
+        "need its results."
+    ),
     "crashed": (
         "[Agent crash] {name} ({agent_id}) terminated unexpectedly. "
         "Stop waiting on this child unless you want to message it again."
@@ -861,9 +866,9 @@ _TERMINAL_NOTICE = {
         "message it again."
     ),
     "stopped": (
-        "[Agent capped] {name} ({agent_id}) hit its turn limit and was stopped "
-        "before finishing. It will not send a completion report, so stop waiting "
-        "on this child; account for its capped subtask and continue."
+        "[Agent stopped] {name} ({agent_id}) was stopped before finishing (turn limit "
+        "or an explicit stop). It will not send a completion report, so stop waiting "
+        "on this child; account for its unfinished subtask and continue."
     ),
 }
 
@@ -898,7 +903,7 @@ async def _notify_parent_on_stall(
     )
 
 
-async def _notify_parent_on_terminal(
+async def notify_parent_on_terminal(
     coordinator: AgentCoordinator,
     agent_id: str,
     status: str,
@@ -911,6 +916,8 @@ async def _notify_parent_on_terminal(
         name = coordinator.names.get(agent_id, agent_id)
     if parent is None:
         return
+    if not await coordinator.claim_parent_notice(agent_id):
+        return
     await coordinator.send(
         parent,
         {
@@ -945,6 +952,21 @@ async def _notify_root_on_budget_reserve(coordinator: AgentCoordinator) -> None:
     await coordinator.send(root, _reserve_notice())
 
 
+async def _notify_parent_on_exit(
+    coordinator: AgentCoordinator,
+    agent_id: str,
+) -> None:
+    """Backstop for a child whose loop ended without telling its parent.
+
+    Every terminal state counts, including ``completed``: a child that skips its
+    completion report leaves the parent waiting on a message nobody will send.
+    """
+    status = await _agent_status(coordinator, agent_id)
+    if status is None:
+        return
+    await notify_parent_on_terminal(coordinator, agent_id, status)
+
+
 async def _start_child_runner(
     *,
     parent_ctx: dict[str, Any],
@@ -998,6 +1020,9 @@ async def _start_child_runner(
             logger.info("child %s stopped after reaching the scan budget limit", child_id)
         except SubagentBudgetReservedError:
             logger.info("child %s stopped at the sub-agent budget reserve", child_id)
+        finally:
+            if not coordinator.is_shutting_down:
+                await _notify_parent_on_exit(coordinator, child_id)
 
     task_handle = asyncio.create_task(_child_loop(), name=f"agent-{name}-{child_id}")
     await coordinator.attach_runtime(child_id, task=task_handle)

strix/core/inputs.py

@@ -59,8 +59,11 @@ def build_root_task(scan_config: dict[str, Any]) -> str:
             )
         elif ttype == "local_code":
             path = details.get("target_path", "unknown")
-            suffix = ", read-only mount" if details.get("mount") else ""
-            sections["Local Codebases"].append(f"- {path} (available at: {workspace_path}{suffix})")
+            sections["Local Codebases"].append(
+                f"- {path} (available at: {workspace_path}; "
+                "this is the user's real directory, mounted live and writable — "
+                ".git/.agents/.codex are read-only)"
+            )
         elif ttype == "web_application":
             sections["URLs"].append(f"- {details.get('target_url', '')}")
         elif ttype == "ip_address":

strix/core/runner.py

@@ -53,6 +53,8 @@ if TYPE_CHECKING:
     from agents.memory import SQLiteSession
     from agents.result import RunResultBase
 
+    from strix.runtime.status import StatusSink
+
 
 logger = logging.getLogger(__name__)
 
@@ -120,6 +122,7 @@ async def run_strix_scan(
     event_sink: StreamEventSink | None = None,
     root_instructions_override: str | None = None,
     extra_system_prompt_context: dict[str, Any] | None = None,
+    status_sink: StatusSink | None = None,
 ) -> RunResultBase | None:
     """Run or resume one Strix scan against a sandbox.
 
@@ -129,6 +132,11 @@ async def run_strix_scan(
     context before prompt rendering. Child agents keep the standard scan prompt
     and context.
     """
+
+    def report(phase: str) -> None:
+        if status_sink is not None:
+            status_sink(phase)
+
     if scan_id is None:
         scan_id = f"scan-{uuid.uuid4().hex[:8]}"
 
@@ -219,7 +227,9 @@ async def run_strix_scan(
         scan_id,
         image=image,
         local_sources=local_sources or [],
+        status_sink=status_sink,
     )
+    report("Waiting for the first model response")
     logger.info("Sandbox ready for scan %s", scan_id)
 
     sandbox_session = bundle["session"]

strix/interface/cli.py

@@ -21,6 +21,7 @@ from strix.runtime import session_manager
 from .utils import (
     build_live_stats_text,
     format_vulnerability_report,
+    has_model_response,
 )
 
 
@@ -135,11 +136,17 @@ async def run_cli(args: Any) -> None:  # noqa: PLR0915
 
     set_global_report_state(report_state)
 
+    startup_phase: list[str] = ["Starting up"]
+
     def create_live_status() -> Panel:
         status_text = Text()
         status_text.append("Penetration test in progress", style="bold #22c55e")
         status_text.append("\n\n")
 
+        if not has_model_response(report_state):
+            status_text.append(f"{startup_phase[0]}...", style="dim")
+            status_text.append("\n\n")
+
         stats_text = build_live_stats_text(report_state)
         if stats_text:
             status_text.append(stats_text)
@@ -152,6 +159,9 @@ async def run_cli(args: Any) -> None:  # noqa: PLR0915
             padding=(1, 2),
         )
 
+    def _note_startup_phase(phase: str) -> None:
+        startup_phase[:] = [phase]
+
     try:
         console.print()
 
@@ -186,6 +196,7 @@ async def run_cli(args: Any) -> None:  # noqa: PLR0915
                     interactive=bool(getattr(args, "interactive", False)),
                     max_budget_usd=getattr(args, "max_budget_usd", None),
                     max_turns=getattr(args, "max_turns", DEFAULT_MAX_TURNS),
+                    status_sink=_note_startup_phase,
                 )
             finally:
                 stop_updates.set()

strix/interface/main.py

@@ -33,12 +33,11 @@ from strix.interface.update_check import (
 from strix.interface.utils import (
     assign_workspace_subdirs,
     build_final_stats_text,
-    build_mount_targets_info,
     check_docker_connection,
+    check_mountable_dir,
     clone_repository,
     collect_local_sources,
     dedupe_local_targets,
-    find_oversized_local_targets,
     generate_run_name,
     image_exists,
     infer_target_type,
@@ -524,9 +523,6 @@ Examples:
   # Local code analysis
   strix --target ./my-project
 
-  # Large local repository (bind-mounted read-only instead of copied)
-  strix --mount ./huge-monorepo
-
   # Domain penetration test
   strix --target example.com
 
@@ -570,8 +566,9 @@ Examples:
         type=str,
         action="append",
         help="Target to test (URL, repository, local directory path, domain name, or IP address). "
+        "Local directories are mounted into the sandbox writable. "
         "Can be specified multiple times for multi-target scans. "
-        "Fresh runs require at least one of --target, --target-list, or --mount.",
+        "Fresh runs require --target or --target-list.",
     )
     parser.add_argument(
         "--target-list",
@@ -581,15 +578,6 @@ Examples:
         help="Path to a file containing targets, one per non-empty, non-comment line. "
         "Can be specified multiple times and combined with --target.",
     )
-    parser.add_argument(
-        "--mount",
-        type=str,
-        action="append",
-        metavar="PATH",
-        help="Bind-mount a local directory into the sandbox (read-only) instead of "
-        "copying it file-by-file. Use this for large repositories that are too big to "
-        "stream into the container. Can be specified multiple times.",
-    )
     parser.add_argument(
         "--instruction",
         type=str,
@@ -720,9 +708,9 @@ Examples:
     args.user_explicit_instruction = args.instruction if args.resume else None
 
     if args.resume:
-        if args.target or args.target_list or args.mount:
+        if args.target or args.target_list:
             parser.error(
-                "Cannot combine --resume with --target/--target-list/--mount. "
+                "Cannot combine --resume with --target/--target-list. "
                 "--resume picks up where the prior run left off, including the "
                 "original target list."
             )
@@ -736,9 +724,9 @@ Examples:
                 f"or remove --resume to start over with the same targets."
             )
     else:
-        if not args.target and not args.target_list and not args.mount:
+        if not args.target and not args.target_list:
             parser.error(
-                "the following arguments are required: -t/--target, --target-list, or --mount "
+                "the following arguments are required: -t/--target or --target-list "
                 "(or use --resume <run_name> to continue a prior scan)"
             )
         args.targets_info = []
@@ -761,33 +749,14 @@ Examples:
                 args.targets_info.append(
                     {"type": target_type, "details": target_dict, "original": display_target}
                 )
-            except ValueError:
-                parser.error(f"Invalid target '{target}'")
-
-        try:
-            args.targets_info.extend(build_mount_targets_info(args.mount or []))
-        except ValueError as e:
-            parser.error(str(e))
+            except ValueError as e:
+                parser.error(f"Invalid target '{target}': {e}")
 
         args.targets_info = dedupe_local_targets(args.targets_info)
 
         assign_workspace_subdirs(args.targets_info)
         rewrite_localhost_targets(args.targets_info, HOST_GATEWAY_HOSTNAME)
 
-        max_local_copy_mb = load_settings().runtime.max_local_copy_mb
-        max_copy_bytes = max_local_copy_mb * 1024 * 1024
-        oversized = find_oversized_local_targets(args.targets_info, max_copy_bytes)
-        if oversized:
-            details = "; ".join(
-                f"{path} ({size / (1024 * 1024):.0f} MB)" for path, size in oversized
-            )
-            parser.error(
-                f"Local target too large to stream into the sandbox: {details}. "
-                f"The limit is {max_local_copy_mb} MB "
-                "(set STRIX_MAX_LOCAL_COPY_MB to change it). Re-run with "
-                "--mount <path> to bind-mount the directory instead of copying it."
-            )
-
     return args
 
 
@@ -839,6 +808,12 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
         if not isinstance(target, dict):
             continue
         details = target.get("details") or {}
+        if target.get("type") == "local_code" and details.get("target_path"):
+            try:
+                check_mountable_dir(Path(details["target_path"]).expanduser())
+            except ValueError as exc:
+                parser.error(f"--resume {args.resume}: {exc}")
+            continue
         if target.get("type") != "repository":
             continue
         cloned = details.get("cloned_repo_path")
@@ -853,8 +828,7 @@ def _load_resume_state(args: argparse.Namespace, parser: argparse.ArgumentParser
 
     if args.instruction is None:
         args.instruction = state.get("instruction")
-    if state.get("local_sources"):
-        args.local_sources = state.get("local_sources")
+    args.local_sources = collect_local_sources(args.targets_info)
     if state.get("diff_scope"):
         args.diff_scope = state.get("diff_scope")
     persisted_scan_mode = state.get("scan_mode")

strix/interface/tui/app.py

@@ -815,6 +815,8 @@ class StrixTUIApp(App):  # type: ignore[misc]
         self._scan_stop_event = threading.Event()
         self._scan_completed = threading.Event()
         self._scan_error: BaseException | None = None
+        self._startup_status = "Starting up"
+        self._startup_status_step = 0
         self._error_noted_agents: set[str] = set()
         self._budget_pause_notified = False
 
@@ -1111,7 +1113,9 @@ class StrixTUIApp(App):  # type: ignore[misc]
         self,
     ) -> tuple[Any, str | None]:
         if not self.selected_agent_id:
-            return self._get_chat_placeholder_content("Loading...", "placeholder-no-agent")
+            return self._get_chat_placeholder_content(
+                f"{self._startup_status}...", f"placeholder-no-agent-{self._startup_status_step}"
+            )
 
         events = self._gather_agent_events(self.selected_agent_id)
 
@@ -1525,6 +1529,7 @@ class StrixTUIApp(App):  # type: ignore[misc]
                                 max_budget_usd=getattr(self.args, "max_budget_usd", None),
                                 max_turns=getattr(self.args, "max_turns", DEFAULT_MAX_TURNS),
                                 event_sink=self._capture_sdk_event,
+                                status_sink=self._capture_startup_status,
                             ),
                         )
 
@@ -1556,6 +1561,18 @@ class StrixTUIApp(App):  # type: ignore[misc]
         self._scan_thread = threading.Thread(target=scan_target, daemon=True)
         self._scan_thread.start()
 
+    def _capture_startup_status(self, phase: str) -> None:
+        try:
+            self.call_from_thread(self._record_startup_status, phase)
+        except RuntimeError:
+            self._record_startup_status(phase)
+
+    def _record_startup_status(self, phase: str) -> None:
+        self._startup_status = phase
+        self._startup_status_step += 1
+        if not self.show_splash and not self.selected_agent_id:
+            self.call_later(self._update_chat_view)
+
     def _capture_sdk_event(self, agent_id: str, event: Any) -> None:
         try:
             self.call_from_thread(self._record_sdk_event, agent_id, event)

strix/interface/utils.py

@@ -290,6 +290,11 @@ def _detail_value(usage: dict[str, Any], detail_key: str, value_key: str) -> int
     return _int_stat(details, value_key)
 
 
+def has_model_response(report_state: Any) -> bool:
+    usage = _llm_usage(report_state)
+    return bool(usage) and _int_stat(usage, "requests") > 0
+
+
 def _build_llm_usage_stats(
     stats_text: Text,
     report_state: Any,
@@ -1131,6 +1136,7 @@ def infer_target_type(target: str) -> tuple[str, dict[str, str]]:  # noqa: PLR09
     try:
         if path.exists():
             if path.is_dir():
+                check_mountable_dir(path)
                 return "local_code", {"target_path": str(path.resolve())}
             raise ValueError(f"Path exists but is not a directory: {target}")
     except (OSError, RuntimeError) as e:
@@ -1259,7 +1265,7 @@ def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str,
                 {
                     "source_path": details["target_path"],
                     "workspace_subdir": workspace_subdir,
-                    "mount": bool(details.get("mount", False)),
+                    "protect_metadata": True,
                 }
             )
 
@@ -1268,123 +1274,126 @@ def collect_local_sources(targets_info: list[dict[str, Any]]) -> list[dict[str,
                 {
                     "source_path": details["cloned_repo_path"],
                     "workspace_subdir": workspace_subdir,
-                    "mount": False,
+                    "protect_metadata": False,
                 }
             )
 
     return local_sources
 
 
-def directory_size_bytes(path: Path) -> int:
-    """Total size in bytes of regular files under ``path`` (symlinks not followed).
+# Refused along with everything under them.
+_FORBIDDEN_MOUNT_TREES = frozenset(
+    {
+        "/bin",
+        "/sbin",
+        "/usr",
+        "/etc",
+        "/lib",
+        "/lib64",
+        "/nix/store",
+        "/run/current-system/sw",
+        "/Applications",
+        "/Library",
+        "/System",
+        "/dev",
+        "/boot",
+        "/proc",
+        "/sys",
+    }
+)
 
-    Best-effort: files that disappear or can't be stat'd mid-walk are skipped.
-    Used as a cheap (stat-only) pre-flight to estimate the cost of streaming a
-    local target into the sandbox before we actually try to copy it.
+# Refused themselves, but they hold projects too, so their contents are fine.
+_FORBIDDEN_MOUNT_ROOTS = frozenset(
+    {
+        "/",
+        "/private",
+        "/var",
+        "/opt",
+        "/home",
+        "/root",
+        "/srv",
+        "/Users",
+        "/Volumes",
+    }
+)
 
-    Directories that can't be listed (e.g. permission denied) are logged and
-    skipped rather than silently dropped — so an under-count is at least
-    visible — but the returned total then excludes their contents.
-    """
+_FORBIDDEN_WINDOWS_TREE_NAMES = frozenset(
+    {"windows", "program files", "program files (x86)", "programdata"}
+)
 
-    def _on_walk_error(error: OSError) -> None:
-        logger.warning("Could not read %s while measuring size: %s", error.filename, error)
-
-    total = 0
-    for root, _dirs, files in os.walk(path, followlinks=False, onerror=_on_walk_error):
-        for name in files:
-            file_path = os.path.join(root, name)  # noqa: PTH118
-            try:
-                if os.path.islink(file_path):  # noqa: PTH114
-                    continue
-                total += os.path.getsize(file_path)  # noqa: PTH202
-            except OSError:
-                continue
-    return total
+_FORBIDDEN_MOUNT_DIR_NAMES = frozenset(
+    {
+        ".ssh",
+        ".tsh",
+        ".brev",
+        ".gnupg",
+        ".aws",
+        ".azure",
+        ".kube",
+        ".docker",
+        ".config",
+        ".npm",
+        ".pki",
+        ".terraform.d",
+    }
+)
 
 
-def find_oversized_local_targets(
-    targets_info: list[dict[str, Any]], max_bytes: int
-) -> list[tuple[str, int]]:
-    """Return ``(path, size_bytes)`` for non-mounted local targets over ``max_bytes``.
-
-    Mounted targets are bind-mounted rather than copied, so their size is
-    irrelevant and they are excluded. A ``max_bytes`` of zero or less disables
-    the check entirely (returns no targets).
-    """
-    if max_bytes <= 0:
-        return []
-    oversized: list[tuple[str, int]] = []
-    for target in targets_info:
-        if target.get("type") != "local_code":
-            continue
-        details = target.get("details") or {}
-        if details.get("mount"):
-            continue
-        target_path = details.get("target_path")
-        if not target_path:
-            continue
-        size = directory_size_bytes(Path(target_path))
-        if size > max_bytes:
-            oversized.append((target_path, size))
-    return oversized
+def _is_within(path: Path, ancestor: Path) -> bool:
+    ancestor_parts = [part.casefold() for part in ancestor.parts]
+    path_parts = [part.casefold() for part in path.parts]
+    return path_parts[: len(ancestor_parts)] == ancestor_parts
 
 
-def build_mount_targets_info(mount_paths: list[str]) -> list[dict[str, Any]]:
-    """Build ``targets_info`` entries for ``--mount`` directories.
+def check_mountable_dir(path: Path) -> None:
+    resolved = path.resolve()
+    if not resolved.is_dir():
+        raise ValueError(f"'{path}' is not an existing directory.")
 
-    Each path must be an existing local directory; it is bind-mounted into the
-    sandbox (read-only) instead of being copied file-by-file. Raises
-    ``ValueError`` for an empty path, or one that does not exist or is not a
-    directory.
-    """
-    targets_info: list[dict[str, Any]] = []
-    for raw in mount_paths:
-        if not raw or not raw.strip():
-            raise ValueError("--mount path must not be empty.")
-        path = Path(raw).expanduser()
-        try:
-            resolved = path.resolve()
-            is_dir = resolved.is_dir()
-        except (OSError, RuntimeError) as e:
-            raise ValueError(f"Invalid mount path '{raw}': {e!s}") from e
-        if not is_dir:
-            raise ValueError(
-                f"Mount path '{raw}' is not an existing directory. "
-                "--mount requires a path to a local directory."
-            )
-        targets_info.append(
-            {
-                "type": "local_code",
-                "details": {"target_path": str(resolved), "mount": True},
-                "original": str(resolved),
-            }
+    # Both the literal and the resolved form: macOS reaches /etc through the
+    # /private/etc symlink, and only the resolved path is compared below.
+    exact = {str(Path(root)).casefold() for root in _FORBIDDEN_MOUNT_ROOTS}
+    exact |= {str(Path(root).resolve()).casefold() for root in _FORBIDDEN_MOUNT_ROOTS}
+    exact.add(str(Path.home().resolve()).casefold())
+    tree_roots = set(_FORBIDDEN_MOUNT_TREES)
+    if os.name == "nt":
+        drive = Path(resolved.anchor)
+        tree_roots |= {str(drive / name) for name in _FORBIDDEN_WINDOWS_TREE_NAMES}
+        exact.add(str(drive / "Users").casefold())
+    trees = [Path(root) for root in tree_roots] + [Path(root).resolve() for root in tree_roots]
+    if (
+        str(resolved).casefold() in exact
+        or resolved.parent == resolved
+        or any(_is_within(resolved, tree) for tree in trees)
+    ):
+        raise ValueError(
+            f"Refusing to mount '{resolved}' into the sandbox: it is a system "
+            "or home directory, not a codebase. Point the target at the "
+            "project directory you want tested."
+        )
+
+    credential = next(
+        (part for part in resolved.parts if part.casefold() in _FORBIDDEN_MOUNT_DIR_NAMES), None
+    )
+    if credential is not None:
+        raise ValueError(
+            f"Refusing to mount '{resolved}' into the sandbox: '{credential}' "
+            "holds credentials, not code."
         )
-    return targets_info
 
 
 def dedupe_local_targets(targets_info: list[dict[str, Any]]) -> list[dict[str, Any]]:
-    """Collapse local_code targets that resolve to the same path.
-
-    When a directory is supplied both as a copied ``--target`` and via
-    ``--mount`` (or as duplicate values of either), keep one entry and prefer
-    the bind-mounted one — so the same tree is never both streamed in and
-    mounted. Order is preserved; non-local targets pass through untouched.
-    """
     result: list[dict[str, Any]] = []
-    index_by_path: dict[str, int] = {}
+    seen_paths: set[str] = set()
     for target in targets_info:
         details = target.get("details") or {}
         path = details.get("target_path")
         if target.get("type") != "local_code" or not path:
             result.append(target)
             continue
-        existing = index_by_path.get(path)
-        if existing is None:
-            index_by_path[path] = len(result)
+        if path not in seen_paths:
+            seen_paths.add(path)
             result.append(target)
-        elif details.get("mount") and not (result[existing].get("details") or {}).get("mount"):
-            result[existing] = target  # bind mount supersedes the copied entry
     return result
 
 

strix/runtime/backends.py

@@ -31,16 +31,11 @@ async def _docker_backend(
     ``docker`` lazily so deployments that target a non-Docker
     backend don't need the docker-py library installed.
 
-    ``session.start()`` is what materializes the manifest entries
-    (LocalDir copies and manifest-declared volume/FUSE mounts) into the
-    running container — the SDK's ``client.create()`` only builds the inner
-    session object without applying the manifest. ``async with session:``
-    would call it too, but Strix manages session lifetime explicitly via
-    ``client.delete()`` so we trigger ``start()`` ourselves.
-
-    ``bind_mounts`` are host directories (e.g. large repos passed via
-    ``--mount``) bind-mounted read-only; unlike manifest entries they are
-    applied by Docker at container-create time, not by ``start()``.
+    ``session.start()`` is what materializes the manifest into the running
+    container — the SDK's ``client.create()`` only builds the inner session
+    object without applying it. ``async with session:`` would call it too, but
+    Strix manages session lifetime explicitly via ``client.delete()`` so we
+    trigger ``start()`` ourselves.
     """
     import docker
     from agents.sandbox.sandboxes.docker import DockerSandboxClientOptions
@@ -59,6 +54,8 @@ _BACKENDS: dict[str, SandboxBackend] = {
     "docker": _docker_backend,
 }
 
+_BIND_MOUNT_BACKENDS: set[str] = {"docker"}
+
 
 def get_backend(name: str) -> SandboxBackend:
     """Return the backend factory for ``name`` or raise.
@@ -78,15 +75,30 @@ def get_backend(name: str) -> SandboxBackend:
     return backend
 
 
-def register_backend(name: str, backend: SandboxBackend) -> None:
+def register_backend(
+    name: str,
+    backend: SandboxBackend,
+    *,
+    supports_bind_mounts: bool = False,
+) -> None:
     """Register a custom backend under ``name``.
 
     Intended for downstream users who ship their own runtime — register
     before any ``session_manager.create_or_reuse`` call. Re-registering
-    an existing name overwrites the prior entry.
+    an existing name overwrites the prior entry. ``supports_bind_mounts``
+    defaults to False: a remote runtime cannot see the caller's filesystem, so
+    it is handed local sources as manifest entries to upload instead.
     """
     _BACKENDS[name] = backend
-    logger.info("Registered sandbox backend: %s", name)
+    if supports_bind_mounts:
+        _BIND_MOUNT_BACKENDS.add(name)
+    else:
+        _BIND_MOUNT_BACKENDS.discard(name)
+    logger.info("Registered sandbox backend: %s (bind mounts: %s)", name, supports_bind_mounts)
+
+
+def backend_supports_bind_mounts(name: str) -> bool:
+    return name in _BIND_MOUNT_BACKENDS
 
 
 def supported_backends() -> list[str]:

strix/runtime/docker_client.py

@@ -237,18 +237,18 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         _apply_log_limits(create_kwargs)
         _apply_run_labels(create_kwargs)
 
-        # Strix injection: host bind mounts (e.g. large repos passed via --mount)
-        # that bypass the SDK's file-by-file LocalDir copy.
-        bind_mounts = getattr(self, "strix_bind_mounts", ())
+        # Strix injection: local source trees, sorted shallowest-first so a
+        # nested spec lands on top of the tree it covers.
+        bind_mounts = self.strix_bind_mounts or ()
         if bind_mounts:
             mounts = create_kwargs.setdefault("mounts", [])
-            for spec in bind_mounts:
+            for spec in sorted(bind_mounts, key=lambda s: str(s["target"]).count("/")):
                 mounts.append(
                     DockerSDKMount(
                         target=spec["target"],
                         source=spec["source"],
                         type="bind",
-                        read_only=spec.get("read_only", True),
+                        read_only=spec.get("read_only", False),
                     )
                 )
 

strix/runtime/local_dir_staging.py

@@ -1,120 +0,0 @@
-"""Symlink-safe staging for ``LocalDir`` manifest uploads.
-
-The sandbox SDK's ``LocalDir`` walker refuses to copy symlinks at all — it
-raises ``LocalDirReadError(reason="symlink_not_supported")`` on the first one
-as a path-escape / TOCTOU safeguard. Real source trees (especially JS/TS
-monorepos with workspace or shared-config links) routinely commit symlinks, so
-handing such a tree straight to ``LocalDir`` aborts the upload before the agent
-even starts.
-
-:func:`stage_symlink_safe_dir` returns a path that is always safe to hand to
-``LocalDir``:
-
-* a tree with no symlinks is used as-is (no copy);
-* otherwise the tree is copied into a temp directory with symlinks resolved:
-
-  - a link whose target stays inside the tree is *dereferenced* (its target
-    content is materialized in place), so the agent still sees the file;
-  - a link that escapes the tree, dangles, or forms a cycle is *dropped* and
-    never followed. Refusing to follow out-of-tree links preserves the walker's
-    path-escape safety and keeps host/out-of-tree content from leaking into the
-    (hostile) sandbox.
-
-Regular files are hard-linked when possible (falling back to a copy across
-devices), so the staged tree adds negligible disk for the non-symlink bulk.
-"""
-
-from __future__ import annotations
-
-import logging
-import os
-import shutil
-import tempfile
-from pathlib import Path
-
-
-logger = logging.getLogger(__name__)
-
-_STAGING_PREFIX = "strix-localdir-"
-
-
-def _is_within(target: Path, root: Path) -> bool:
-    """Return whether ``target`` is ``root`` itself or nested under it."""
-    if target == root:
-        return True
-    try:
-        target.relative_to(root)
-    except ValueError:
-        return False
-    return True
-
-
-def tree_has_symlink(root: Path) -> bool:
-    """Return whether ``root`` contains any symlink (file or directory)."""
-    for dirpath, dirnames, filenames in os.walk(root, followlinks=False):
-        base = Path(dirpath)
-        for name in (*dirnames, *filenames):
-            if (base / name).is_symlink():
-                return True
-    return False
-
-
-def _link_or_copy(src: Path, dst: Path) -> None:
-    """Hard-link ``src`` to ``dst``, falling back to a content copy."""
-    try:
-        os.link(src, dst)
-    except OSError:
-        shutil.copy2(src, dst, follow_symlinks=True)
-
-
-def _stage_dir(src: Path, dst: Path, root: Path, seen: frozenset[Path]) -> None:
-    dst.mkdir(parents=True, exist_ok=True)
-    for entry in os.scandir(src):
-        entry_path = Path(entry.path)
-        dest_path = dst / entry.name
-
-        if entry.is_symlink():
-            target = Path(os.path.realpath(entry_path))
-            if not _is_within(target, root):
-                logger.warning("staging: dropping out-of-tree symlink %s -> %s", entry_path, target)
-                continue
-            if not target.exists():
-                logger.warning("staging: dropping dangling symlink %s", entry_path)
-                continue
-            if target in seen:
-                logger.warning("staging: dropping cyclic symlink %s -> %s", entry_path, target)
-                continue
-            if target.is_dir():
-                _stage_dir(target, dest_path, root, seen | {target})
-            else:
-                _link_or_copy(target, dest_path)
-        elif entry.is_dir(follow_symlinks=False):
-            _stage_dir(entry_path, dest_path, root, seen)
-        elif entry.is_file(follow_symlinks=False):
-            _link_or_copy(entry_path, dest_path)
-        else:
-            # Sockets, FIFOs, devices — not part of a source tree; skip.
-            logger.debug("staging: skipping non-regular entry %s", entry_path)
-
-
-def stage_symlink_safe_dir(src_root: Path) -> tuple[Path, Path | None]:
-    """Return ``(upload_path, staged_temp)`` for uploading ``src_root``.
-
-    ``upload_path`` is safe to hand to ``LocalDir``. When the tree contains no
-    symlinks it is ``src_root`` itself and ``staged_temp`` is ``None``.
-    Otherwise a symlink-safe copy is materialized in a temp directory and both
-    returned values point at it; the caller owns removing ``staged_temp`` once
-    the upload completes.
-    """
-    root = src_root.resolve()
-    if not tree_has_symlink(root):
-        return root, None
-
-    staged = Path(tempfile.mkdtemp(prefix=_STAGING_PREFIX)).resolve()
-    try:
-        _stage_dir(root, staged, root, frozenset({root}))
-    except OSError:
-        shutil.rmtree(staged, ignore_errors=True)
-        raise
-    logger.info("staging: materialized symlink-safe copy of %s at %s", root, staged)
-    return staged, staged

strix/runtime/session_manager.py

@@ -3,17 +3,21 @@
 from __future__ import annotations
 
 import logging
-import shutil
+import os
+import sys
 from pathlib import Path
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from agents.sandbox.entries import BaseEntry, LocalDir
 from agents.sandbox.manifest import Environment, Manifest
 
 from strix.config import load_settings
-from strix.runtime.backends import get_backend
+from strix.runtime.backends import backend_supports_bind_mounts, get_backend
 from strix.runtime.caido_bootstrap import bootstrap_caido
-from strix.runtime.local_dir_staging import stage_symlink_safe_dir
+
+
+if TYPE_CHECKING:
+    from strix.runtime.status import StatusSink
 
 
 logger = logging.getLogger(__name__)
@@ -28,43 +32,72 @@ _SESSION_CACHE: dict[str, dict[str, Any]] = {}
 # Manifest root inside the container; entry keys hang off this path.
 _WORKSPACE_ROOT = "/workspace"
 
+_PROTECTED_METADATA_NAMES = (".git", ".agents", ".codex")
 
-def build_session_entries(
-    local_sources: list[dict[str, Any]],
-) -> tuple[dict[str | Path, BaseEntry], list[dict[str, Any]], list[Path]]:
-    """Split local sources into copied manifest entries and host bind mounts.
 
-    Sources flagged ``mount`` are bind-mounted read-only at
-    ``/workspace/<workspace_subdir>`` (not added to the manifest, so the SDK
-    does not stream them in file-by-file). Every other source becomes a
-    ``LocalDir`` entry copied into the container as before. Trees containing
-    symlinks (which the SDK's ``LocalDir`` walker refuses outright) are first
-    staged into a symlink-safe temp copy; those temp dirs are returned so the
-    caller can remove them once the upload completes.
-    """
-    entries: dict[str | Path, BaseEntry] = {}
+def _host_identity_env() -> dict[str, str]:
+    if sys.platform != "linux":
+        return {}
+    return {"STRIX_HOST_UID": str(os.getuid()), "STRIX_HOST_GID": str(os.getgid())}
+
+
+def build_bind_mounts(local_sources: list[dict[str, Any]]) -> list[dict[str, Any]]:
     bind_mounts: list[dict[str, Any]] = []
-    staged_dirs: list[Path] = []
     for src in local_sources:
         ws_subdir = src.get("workspace_subdir") or ""
         host_path = src.get("source_path") or ""
         if not ws_subdir or not host_path:
             continue
         resolved = Path(host_path).expanduser().resolve()
-        if src.get("mount"):
-            bind_mounts.append(
-                {
-                    "source": str(resolved),
-                    "target": f"{_WORKSPACE_ROOT}/{ws_subdir}",
-                    "read_only": True,
-                }
+        target = f"{_WORKSPACE_ROOT}/{ws_subdir}"
+        bind_mounts.append({"source": str(resolved), "target": target, "read_only": False})
+        if src.get("protect_metadata"):
+            bind_mounts.extend(_metadata_mounts(resolved, target))
+    return bind_mounts
+
+
+def build_manifest_entries(local_sources: list[dict[str, Any]]) -> dict[str | Path, BaseEntry]:
+    entries: dict[str | Path, BaseEntry] = {}
+    for src in local_sources:
+        ws_subdir = src.get("workspace_subdir") or ""
+        host_path = src.get("source_path") or ""
+        if not ws_subdir or not host_path:
+            continue
+        entries[ws_subdir] = LocalDir(src=Path(host_path).expanduser().resolve())
+    return entries
+
+
+def _metadata_mounts(tree: Path, target: str) -> list[dict[str, Any]]:
+    mounts: list[dict[str, Any]] = []
+    for name in _PROTECTED_METADATA_NAMES:
+        metadata = tree / name
+        if not metadata.is_dir() and not metadata.is_file():
+            continue
+        if not metadata.resolve().is_relative_to(tree):
+            continue
+        mounts.append({"source": str(metadata), "target": f"{target}/{name}", "read_only": True})
+        gitdir = _gitdir_from_pointer(metadata) if metadata.is_file() else None
+        if gitdir is not None and gitdir.exists() and gitdir.is_relative_to(tree):
+            relative = gitdir.relative_to(tree).as_posix()
+            mounts.append(
+                {"source": str(gitdir), "target": f"{target}/{relative}", "read_only": True}
             )
-        else:
-            upload_path, staged = stage_symlink_safe_dir(resolved)
-            if staged is not None:
-                staged_dirs.append(staged)
-            entries[ws_subdir] = LocalDir(src=upload_path)
-    return entries, bind_mounts, staged_dirs
+    return mounts
+
+
+def _gitdir_from_pointer(git_file: Path) -> Path | None:
+    try:
+        content = git_file.read_text(encoding="utf-8", errors="replace")
+    except OSError:
+        return None
+    for line in content.splitlines():
+        prefix, _, value = line.partition(":")
+        if prefix.strip() == "gitdir" and value.strip():
+            candidate = Path(value.strip()).expanduser()
+            if not candidate.is_absolute():
+                candidate = git_file.parent / candidate
+            return candidate.resolve()
+    return None
 
 
 async def create_or_reuse(
@@ -72,19 +105,32 @@ async def create_or_reuse(
     *,
     image: str,
     local_sources: list[dict[str, Any]],
+    status_sink: StatusSink | None = None,
 ) -> dict[str, Any]:
     """Return the existing session bundle for ``scan_id`` or create a new one.
 
     Each ``local_sources`` entry exposes its host ``source_path`` at
-    ``/workspace/<workspace_subdir>`` inside the container — copied in, or
-    bind-mounted read-only when the entry is flagged ``mount``.
+    ``/workspace/<workspace_subdir>`` inside the container.
     """
+
+    def report(phase: str) -> None:
+        if status_sink is not None:
+            status_sink(phase)
+
     cached = _SESSION_CACHE.get(scan_id)
     if cached is not None:
         logger.info("Reusing existing sandbox session for scan %s", scan_id)
         return cached
 
-    entries, bind_mounts, staged_dirs = build_session_entries(local_sources)
+    backend_name = load_settings().runtime.backend
+    backend = get_backend(backend_name)
+
+    if backend_supports_bind_mounts(backend_name):
+        bind_mounts = build_bind_mounts(local_sources)
+        entries: dict[str | Path, BaseEntry] = {}
+    else:
+        bind_mounts = []
+        entries = build_manifest_entries(local_sources)
 
     # Caido runs as an in-container sidecar; HTTP(S) traffic from any
     # process started via ``session.exec`` (the SDK's Shell tool, etc.)
@@ -98,6 +144,7 @@ async def create_or_reuse(
             value={
                 "PYTHONUNBUFFERED": "1",
                 "HOST_GATEWAY": "host.docker.internal",
+                **_host_identity_env(),
                 "http_proxy": container_caido_url,
                 "https_proxy": container_caido_url,
                 "ALL_PROXY": container_caido_url,
@@ -106,26 +153,21 @@ async def create_or_reuse(
         ),
     )
 
-    backend_name = load_settings().runtime.backend
-    backend = get_backend(backend_name)
-
     logger.info(
         "Creating sandbox session for scan %s (backend=%s, image=%s)",
         scan_id,
         backend_name,
         image,
     )
-    try:
-        client, session = await backend(
-            image=image,
-            manifest=manifest,
-            exposed_ports=(_CONTAINER_CAIDO_PORT,),
-            bind_mounts=bind_mounts,
-        )
-    finally:
-        for staged in staged_dirs:
-            shutil.rmtree(staged, ignore_errors=True)
+    report("Starting sandbox container")
+    client, session = await backend(
+        image=image,
+        manifest=manifest,
+        exposed_ports=(_CONTAINER_CAIDO_PORT,),
+        bind_mounts=bind_mounts,
+    )
 
+    report("Setting up the proxy")
     caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
     scheme = "https" if caido_endpoint.tls else "http"
     host_caido_url = f"{scheme}://{caido_endpoint.host}:{caido_endpoint.port}"

strix/runtime/status.py

@@ -0,0 +1,8 @@
+"""Startup phase reporting."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+
+StatusSink = Callable[[str], None]

strix/tools/agents_graph/tools.py

@@ -13,6 +13,7 @@ from typing import Any, Literal, get_args
 from agents import RunContextWrapper, function_tool
 
 from strix.core.agents import Status, coordinator_from_context
+from strix.core.execution import notify_parent_on_terminal
 from strix.skills import validate_requested_skills
 
 
@@ -558,7 +559,7 @@ async def agent_finish(
         )
 
     parent_notified = False
-    if report_to_parent:
+    if report_to_parent and await coordinator.claim_parent_notice(me):
         async with coordinator._lock:
             agent_name = coordinator.names.get(me, me)
         report = _render_completion_report(
@@ -582,6 +583,11 @@ async def agent_finish(
         )
         parent_notified = True
 
+    await coordinator.set_status(me, "completed")
+    if not parent_notified:
+        # Silence here would leave a parent waiting on a report that is never coming.
+        await notify_parent_on_terminal(coordinator, me, "completed")
+
     logger.info(
         "agent_finish: %s success=%s findings=%d parent_notified=%s",
         me,
@@ -589,7 +595,6 @@ async def agent_finish(
         len(findings or []),
         parent_notified,
     )
-    await coordinator.set_status(me, "completed")
 
     return json.dumps(
         {
@@ -680,9 +685,16 @@ async def stop_agent(
         )
 
     if cascade:
-        await coordinator.cancel_descendants_graceful(target_agent_id)
+        stopped = await coordinator.cancel_descendants_graceful(target_agent_id)
     else:
         await coordinator.request_stop(target_agent_id)
+        stopped = [target_agent_id]
+
+    # The stopper knows what it just did; anyone else waiting on those agents does not.
+    async with coordinator._lock:
+        orphaned = [aid for aid in stopped if coordinator.parent_of.get(aid) not in (None, me)]
+    for aid in orphaned:
+        await notify_parent_on_terminal(coordinator, aid, "stopped")
 
     logger.info(
         "stop_agent: target=%s cascade=%s reason=%r",

tests/test_cli_target_list.py

@@ -30,9 +30,7 @@ def test_parse_arguments_accepts_target_list_file(
 ) -> None:
     target_list = tmp_path / "targets.txt"
     target_list.write_text(
-        "https://test1.com/\n"
-        "\n"
-        "http://test2.com:5789/\n",
+        "https://test1.com/\n\nhttp://test2.com:5789/\n",
         encoding="utf-8",
     )
     _stub_settings(monkeypatch)
@@ -84,7 +82,4 @@ def test_parse_arguments_rejects_resume_with_target_list(
     with pytest.raises(SystemExit):
         cli_main.parse_arguments()
 
-    assert (
-        "Cannot combine --resume with --target/--target-list/--mount"
-        in capsys.readouterr().err
-    )
+    assert "Cannot combine --resume with --target/--target-list" in capsys.readouterr().err

tests/test_config_loader.py

@@ -33,7 +33,6 @@ _LLM_ENV_KEYS = [
     # RuntimeSettings
     "STRIX_IMAGE",
     "STRIX_RUNTIME_BACKEND",
-    "STRIX_MAX_LOCAL_COPY_MB",
     # TelemetrySettings
     "STRIX_TELEMETRY",
 ]

tests/test_execution.py

@@ -18,10 +18,11 @@ from openai.types.responses import ResponseOutputMessage, ResponseOutputRefusal
 from strix.core import execution
 from strix.core.agents import AgentCoordinator
 from strix.core.execution import (
-    _notify_parent_on_terminal,
     _notify_root_on_budget_reserve,
+    notify_parent_on_terminal,
 )
 from strix.core.sessions import seed_initial_input
+from strix.tools.agents_graph.tools import agent_finish, stop_agent
 from strix.tools.finish.tool import finish_scan
 
 
@@ -67,6 +68,43 @@ async def _call_finish_scan(
     return parsed
 
 
+async def _call_agent_finish(
+    coordinator: AgentCoordinator,
+    agent_id: str,
+    parent_id: str | None,
+    *,
+    report_to_parent: bool,
+) -> dict[str, Any]:
+    ctx = ToolContext(
+        context={"coordinator": coordinator, "agent_id": agent_id, "parent_id": parent_id},
+        tool_name="agent_finish",
+        tool_call_id="call-1",
+        tool_arguments="{}",
+    )
+    result: str = await agent_finish.on_invoke_tool(
+        ctx,
+        json.dumps({"result_summary": "done", "report_to_parent": report_to_parent}),
+    )
+    parsed: dict[str, Any] = json.loads(result)
+    return parsed
+
+
+async def _call_stop_agent(
+    coordinator: AgentCoordinator, agent_id: str, target_agent_id: str
+) -> dict[str, Any]:
+    ctx = ToolContext(
+        context={"coordinator": coordinator, "agent_id": agent_id},
+        tool_name="stop_agent",
+        tool_call_id="call-1",
+        tool_arguments="{}",
+    )
+    result: str = await stop_agent.on_invoke_tool(
+        ctx, json.dumps({"target_agent_id": target_agent_id})
+    )
+    parsed: dict[str, Any] = json.loads(result)
+    return parsed
+
+
 @pytest.mark.asyncio
 async def test_reserve_stop_notifies_root_once(monkeypatch: pytest.MonkeyPatch) -> None:
     coordinator = AgentCoordinator()
@@ -466,11 +504,11 @@ async def test_snapshot_round_trip_preserves_budget_pause() -> None:
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("status", ["stopped", "failed", "crashed"])
+@pytest.mark.parametrize("status", ["completed", "stopped", "failed", "crashed"])
 async def test_terminal_child_wakes_parked_parent(tmp_path: Any, status: str) -> None:
-    # Regression for #870: a child reaching a terminal state (e.g. MaxTurnsExceeded
-    # -> "stopped") must wake the parent parked in wait_for_message, so the root can
-    # finalize the scan instead of hanging for a completion report that never arrives.
+    # Regression for #870 and #947: a child reaching any terminal state - including a
+    # plain "completed" - must wake the parent parked in wait_for_agents, so the root
+    # can finalize the scan instead of hanging for a report that never arrives.
     coordinator = AgentCoordinator()
     await coordinator.register("root", "strix", parent_id=None)
     await coordinator.register("child", "SQL Injection", parent_id="root")
@@ -482,13 +520,82 @@ async def test_terminal_child_wakes_parked_parent(tmp_path: Any, status: str) ->
     assert not root_waiter.done()
 
     await coordinator.set_status("child", status, error="Max turns (500) exceeded")
-    await _notify_parent_on_terminal(coordinator, "child", status)
+    await notify_parent_on_terminal(coordinator, "child", status)
 
     await asyncio.wait_for(root_waiter, timeout=1.0)
     assert coordinator.pending_counts.get("root", 0) > 0
     session.close()
 
 
+@pytest.mark.asyncio
+async def test_agent_finish_without_report_still_wakes_parent(tmp_path: Any) -> None:
+    # Regression for #947: a child that completes with report_to_parent=False owes its
+    # parent a terminal notice, otherwise the parent waits out its full timeout.
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child", "recon", parent_id="root")
+    session = SQLiteSession("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)
+
+    root_waiter = asyncio.create_task(coordinator.wait_for_message("root"))
+    await asyncio.sleep(0)
+
+    await _call_agent_finish(coordinator, "child", "root", report_to_parent=False)
+
+    await asyncio.wait_for(root_waiter, timeout=1.0)
+    assert coordinator.statuses["child"] == "completed"
+    assert coordinator.pending_counts.get("root", 0) == 1
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_agent_finish_report_suppresses_the_terminal_notice(tmp_path: Any) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child", "recon", parent_id="root")
+    session = SQLiteSession("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)
+
+    await _call_agent_finish(coordinator, "child", "root", report_to_parent=True)
+    # The exit backstop must not duplicate the report the child already delivered.
+    await execution._notify_parent_on_exit(coordinator, "child")
+
+    assert coordinator.pending_counts.get("root", 0) == 1
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_stop_agent_notifies_a_parent_that_is_not_the_stopper(tmp_path: Any) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child", "recon", parent_id="root")
+    await coordinator.register("grandchild", "sqli", parent_id="child")
+    session = SQLiteSession("child", tmp_path / "agents.db")
+    await coordinator.attach_runtime("child", session=session)
+
+    await _call_stop_agent(coordinator, "root", "grandchild")
+
+    assert coordinator.statuses["grandchild"] == "stopped"
+    assert coordinator.pending_counts.get("child", 0) == 1
+    # The stopper already knows; only the waiting parent needs telling.
+    assert coordinator.pending_counts.get("root", 0) == 0
+    session.close()
+
+
+@pytest.mark.asyncio
+async def test_stop_agent_does_not_notify_the_stopping_parent(tmp_path: Any) -> None:
+    coordinator = AgentCoordinator()
+    await coordinator.register("root", "strix", parent_id=None)
+    await coordinator.register("child", "recon", parent_id="root")
+    session = SQLiteSession("root", tmp_path / "agents.db")
+    await coordinator.attach_runtime("root", session=session)
+
+    await _call_stop_agent(coordinator, "root", "child")
+
+    assert coordinator.pending_counts.get("root", 0) == 0
+    session.close()
+
+
 @pytest.mark.asyncio
 async def test_notify_parent_on_terminal_ignores_non_terminal_status(tmp_path: Any) -> None:
     coordinator = AgentCoordinator()
@@ -497,7 +604,7 @@ async def test_notify_parent_on_terminal_ignores_non_terminal_status(tmp_path: A
     session = SQLiteSession("root", tmp_path / "agents.db")
     await coordinator.attach_runtime("root", session=session)
 
-    await _notify_parent_on_terminal(coordinator, "child", "waiting")
+    await notify_parent_on_terminal(coordinator, "child", "waiting")
 
     assert coordinator.pending_counts.get("root", 0) == 0
     session.close()
@@ -523,7 +630,7 @@ async def test_terminal_notice_does_not_cancel_parent_stream(tmp_path: Any) -> N
     await coordinator.attach_runtime("root", session=session, interrupt_on_message=True)
     await coordinator.attach_stream("root", stream)
 
-    await _notify_parent_on_terminal(coordinator, "child", "crashed")
+    await notify_parent_on_terminal(coordinator, "child", "crashed")
 
     assert stream.cancelled is False
     assert coordinator.pending_counts.get("root", 0) > 0

tests/test_local_dir_staging.py

@@ -1,143 +0,0 @@
-"""Tests for symlink-safe LocalDir staging."""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING
-
-from strix.runtime.local_dir_staging import stage_symlink_safe_dir, tree_has_symlink
-
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
-
-def _make_repo(tmp_path: Path) -> Path:
-    repo = tmp_path / "repo"
-    (repo / "pkg").mkdir(parents=True)
-    (repo / "pkg" / "mod.py").write_text("x = 1\n")
-    (repo / "README.md").write_text("readme\n")
-    return repo
-
-
-def test_tree_without_symlinks_used_as_is(tmp_path: Path) -> None:
-    repo = _make_repo(tmp_path)
-
-    upload_path, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is None
-    assert upload_path == repo.resolve()
-    assert not tree_has_symlink(repo)
-
-
-def test_in_tree_file_symlink_is_dereferenced(tmp_path: Path) -> None:
-    repo = _make_repo(tmp_path)
-    (repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
-
-    upload_path, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is not None
-    assert upload_path == staged
-    assert not (staged / "link.py").is_symlink()
-    assert (staged / "link.py").read_text() == "x = 1\n"
-    assert (staged / "pkg" / "mod.py").read_text() == "x = 1\n"
-    assert not tree_has_symlink(staged)
-
-
-def test_in_tree_relative_dir_symlink_is_dereferenced(tmp_path: Path) -> None:
-    repo = _make_repo(tmp_path)
-    (repo / "pkg_alias").symlink_to("pkg")
-
-    _upload, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is not None
-    assert (staged / "pkg_alias" / "mod.py").read_text() == "x = 1\n"
-    assert not tree_has_symlink(staged)
-
-
-def test_out_of_tree_symlink_is_dropped(tmp_path: Path) -> None:
-    repo = _make_repo(tmp_path)
-    outside = tmp_path / "outside.txt"
-    outside.write_text("secret\n")
-    (repo / "escape.txt").symlink_to(outside)
-    (repo / "abs_escape").symlink_to("/etc")
-
-    _upload, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is not None
-    assert not (staged / "escape.txt").exists()
-    assert not (staged / "abs_escape").exists()
-    assert (staged / "README.md").exists()
-
-
-def test_dangling_symlink_is_dropped(tmp_path: Path) -> None:
-    repo = _make_repo(tmp_path)
-    (repo / "dangling").symlink_to(repo / "does-not-exist")
-
-    _upload, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is not None
-    assert not (staged / "dangling").exists()
-    assert not (staged / "dangling").is_symlink()
-
-
-def test_cyclic_symlink_terminates(tmp_path: Path) -> None:
-    repo = _make_repo(tmp_path)
-    (repo / "self").symlink_to(repo)
-    (repo / "pkg" / "up").symlink_to("..")
-
-    _upload, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is not None
-    assert (staged / "README.md").exists()
-    assert not tree_has_symlink(staged)
-
-
-def test_nested_symlinks_inside_linked_dir(tmp_path: Path) -> None:
-    repo = _make_repo(tmp_path)
-    shared = repo / "shared"
-    shared.mkdir()
-    (shared / "conf.json").write_text("{}\n")
-    (shared / "escape").symlink_to("/etc/passwd")
-    (repo / "pkg" / "shared_link").symlink_to(shared)
-
-    _upload, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is not None
-    assert (staged / "pkg" / "shared_link" / "conf.json").read_text() == "{}\n"
-    assert not (staged / "pkg" / "shared_link" / "escape").exists()
-    assert not (staged / "shared" / "escape").exists()
-
-
-def test_staged_path_has_no_symlink_ancestor(tmp_path: Path, monkeypatch) -> None:  # noqa: ANN001
-    """The staging directory itself must never sit behind a symlink.
-
-    ``tempfile.mkdtemp()`` honors ``$TMPDIR``, and on macOS the default
-    ``$TMPDIR`` resolves through ``/var``, which is itself a symlink to
-    ``/private/var``. ``LocalDir`` rejects any symlink component in its
-    source path, so returning the raw ``mkdtemp()`` result breaks every
-    local-dir upload on macOS whenever the source tree contains a symlink.
-    This reproduces that shape without depending on the host OS layout.
-    """
-    repo = _make_repo(tmp_path)
-    (repo / "link.py").symlink_to(repo / "pkg" / "mod.py")
-
-    real_tmp_root = tmp_path / "real_tmp"
-    real_tmp_root.mkdir()
-    symlinked_tmp_root = tmp_path / "tmp_symlink"
-    symlinked_tmp_root.symlink_to(real_tmp_root)
-
-    def fake_mkdtemp(prefix: str = "") -> str:
-        real_dir = real_tmp_root / f"{prefix}fake"
-        real_dir.mkdir()
-        return str(symlinked_tmp_root / real_dir.name)
-
-    monkeypatch.setattr(
-        "strix.runtime.local_dir_staging.tempfile.mkdtemp", fake_mkdtemp
-    )
-
-    upload_path, staged = stage_symlink_safe_dir(repo)
-
-    assert staged is not None
-    assert upload_path == staged
-    for path in (staged, *staged.parents):
-        assert not path.is_symlink(), f"staged path has a symlink ancestor: {path}"

tests/test_local_sources.py

@@ -1,171 +1,138 @@
-"""Tests for local-source sizing and ``--mount`` target helpers in interface.utils."""
+"""Tests for local-source collection and mount policy in interface.utils."""
 
 from __future__ import annotations
 
-import logging
-import os
-import sys
-from typing import TYPE_CHECKING, Any
+from pathlib import Path
+from typing import Any
 
 import pytest
 
-
-if TYPE_CHECKING:
-    from pathlib import Path
-
 from strix.interface.utils import (
-    build_mount_targets_info,
+    check_mountable_dir,
     collect_local_sources,
     dedupe_local_targets,
-    directory_size_bytes,
-    find_oversized_local_targets,
+    infer_target_type,
     read_target_list_file,
 )
 
 
-def _write_file(path: Path, size: int) -> None:
-    path.write_bytes(b"x" * size)
+def _local_target(target_path: str) -> dict[str, Any]:
+    return {
+        "type": "local_code",
+        "details": {"target_path": target_path, "workspace_subdir": "repo"},
+        "original": target_path,
+    }
 
 
-def _local_target(target_path: str, *, mount: bool = False) -> dict[str, Any]:
-    details: dict[str, Any] = {"target_path": target_path, "workspace_subdir": "repo"}
-    if mount:
-        details["mount"] = True
-    return {"type": "local_code", "details": details, "original": target_path}
+def test_collect_local_sources_protects_the_users_own_git() -> None:
+    sources = collect_local_sources([_local_target("/code")])
+    assert sources == [
+        {"source_path": "/code", "workspace_subdir": "repo", "protect_metadata": True}
+    ]
 
 
-def test_directory_size_empty_dir_is_zero(tmp_path: Path) -> None:
-    assert directory_size_bytes(tmp_path) == 0
-
-
-def test_directory_size_sums_flat_and_nested_files(tmp_path: Path) -> None:
-    _write_file(tmp_path / "a.txt", 100)
-    nested = tmp_path / "sub" / "deep"
-    nested.mkdir(parents=True)
-    _write_file(nested / "b.txt", 250)
-    assert directory_size_bytes(tmp_path) == 350
-
-
-def test_directory_size_skips_symlinks(tmp_path: Path) -> None:
-    _write_file(tmp_path / "real.txt", 100)
-    (tmp_path / "link.txt").symlink_to(tmp_path / "real.txt")
-    # The symlink target is counted once via the real file, not doubled.
-    assert directory_size_bytes(tmp_path) == 100
-
-
-@pytest.mark.skipif(sys.platform == "win32", reason="relies on POSIX permissions")
-def test_directory_size_logs_and_skips_unreadable_subdir(
-    tmp_path: Path, caplog: pytest.LogCaptureFixture
-) -> None:
-    if hasattr(os, "geteuid") and os.geteuid() == 0:
-        pytest.skip("root bypasses directory permissions")
-    _write_file(tmp_path / "top.txt", 100)
-    locked = tmp_path / "locked"
-    locked.mkdir()
-    _write_file(locked / "secret.bin", 9999)
-    locked.chmod(0o000)
-    try:
-        with caplog.at_level(logging.WARNING):
-            size = directory_size_bytes(tmp_path)
-    finally:
-        locked.chmod(0o755)
-    # The unreadable subtree is excluded (not silently treated as readable) and
-    # the omission is logged rather than vanishing without a trace.
-    assert size == 100
-    assert any("Could not read" in record.message for record in caplog.records)
-
-
-def test_find_oversized_returns_nothing_under_limit(tmp_path: Path) -> None:
-    _write_file(tmp_path / "a.txt", 100)
-    targets = [_local_target(str(tmp_path))]
-    assert find_oversized_local_targets(targets, max_bytes=1000) == []
-
-
-def test_find_oversized_returns_target_over_limit(tmp_path: Path) -> None:
-    _write_file(tmp_path / "big.bin", 500)
-    targets = [_local_target(str(tmp_path))]
-    result = find_oversized_local_targets(targets, max_bytes=100)
-    assert result == [(str(tmp_path), 500)]
-
-
-def test_find_oversized_ignores_mounted_targets(tmp_path: Path) -> None:
-    _write_file(tmp_path / "big.bin", 500)
-    targets = [_local_target(str(tmp_path), mount=True)]
-    assert find_oversized_local_targets(targets, max_bytes=100) == []
-
-
-def test_find_oversized_ignores_non_local_targets() -> None:
-    targets = [{"type": "web_application", "details": {"target_url": "https://x"}}]
-    assert find_oversized_local_targets(targets, max_bytes=1) == []
-
-
-@pytest.mark.parametrize("disabled", [0, -1])
-def test_find_oversized_disabled_for_non_positive_limit(tmp_path: Path, disabled: int) -> None:
-    _write_file(tmp_path / "big.bin", 500)
-    targets = [_local_target(str(tmp_path))]
-    assert find_oversized_local_targets(targets, max_bytes=disabled) == []
-
-
-def test_collect_local_sources_propagates_mount_flag() -> None:
-    copied = _local_target("/copied")
-    copied["details"]["workspace_subdir"] = "copied"
-    mounted = _local_target("/mounted", mount=True)
-    mounted["details"]["workspace_subdir"] = "mounted"
-
-    sources = collect_local_sources([copied, mounted])
-
-    by_path = {s["source_path"]: s for s in sources}
-    assert by_path["/copied"]["mount"] is False
-    assert by_path["/mounted"]["mount"] is True
-
-
-def test_collect_local_sources_repository_is_never_mounted() -> None:
+def test_collect_local_sources_leaves_a_clone_writable() -> None:
     repo = {
         "type": "repository",
         "details": {"cloned_repo_path": "/clone", "workspace_subdir": "clone"},
     }
     sources = collect_local_sources([repo])
-    assert sources == [{"source_path": "/clone", "workspace_subdir": "clone", "mount": False}]
+    assert sources == [
+        {"source_path": "/clone", "workspace_subdir": "clone", "protect_metadata": False}
+    ]
 
 
-def test_build_mount_targets_info_for_valid_dir(tmp_path: Path) -> None:
-    result = build_mount_targets_info([str(tmp_path)])
-    assert len(result) == 1
-    entry = result[0]
-    assert entry["type"] == "local_code"
-    assert entry["details"]["mount"] is True
-    assert entry["details"]["target_path"] == str(tmp_path.resolve())
+def test_check_mountable_dir_accepts_a_project_dir(tmp_path: Path) -> None:
+    check_mountable_dir(tmp_path)
 
 
-def test_build_mount_targets_info_rejects_missing_path(tmp_path: Path) -> None:
-    missing = tmp_path / "does-not-exist"
+def test_check_mountable_dir_rejects_missing_path(tmp_path: Path) -> None:
     with pytest.raises(ValueError, match="not an existing directory"):
-        build_mount_targets_info([str(missing)])
+        check_mountable_dir(tmp_path / "nope")
 
 
-def test_build_mount_targets_info_rejects_file(tmp_path: Path) -> None:
-    file_path = tmp_path / "a-file.txt"
-    _write_file(file_path, 10)
-    with pytest.raises(ValueError, match="not an existing directory"):
-        build_mount_targets_info([str(file_path)])
+def test_check_mountable_dir_rejects_filesystem_root() -> None:
+    with pytest.raises(ValueError, match="Refusing to mount"):
+        check_mountable_dir(Path("/"))
 
 
-@pytest.mark.parametrize("empty", ["", "   "])
-def test_build_mount_targets_info_rejects_empty_path(empty: str) -> None:
-    # An empty path would otherwise resolve to the current working directory
-    # and silently bind-mount it into the sandbox.
-    with pytest.raises(ValueError, match="must not be empty"):
-        build_mount_targets_info([empty])
+def test_check_mountable_dir_rejects_home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    home = tmp_path / "home"
+    home.mkdir()
+    monkeypatch.setenv("HOME", str(home))
+    monkeypatch.setattr(Path, "home", classmethod(lambda _cls: home))
+
+    with pytest.raises(ValueError, match="Refusing to mount"):
+        check_mountable_dir(home)
+
+
+def test_check_mountable_dir_rejects_system_root() -> None:
+    etc = Path("/etc")
+    if not etc.is_dir():
+        pytest.skip("no /etc on this platform")
+    with pytest.raises(ValueError, match="Refusing to mount"):
+        check_mountable_dir(etc)
+
+
+def test_check_mountable_dir_rejects_the_shared_home_root() -> None:
+    home_root = Path("/home")
+    if not home_root.is_dir():
+        pytest.skip("no /home on this platform")
+    with pytest.raises(ValueError, match="Refusing to mount"):
+        check_mountable_dir(home_root)
+
+
+def test_check_mountable_dir_matches_forbidden_names_case_insensitively(tmp_path: Path) -> None:
+    ssh_dir = tmp_path / ".SSH"
+    ssh_dir.mkdir()
+
+    with pytest.raises(ValueError, match="holds credentials"):
+        check_mountable_dir(ssh_dir)
+
+
+def test_check_mountable_dir_rejects_credential_dirs(tmp_path: Path) -> None:
+    ssh_dir = tmp_path / ".ssh"
+    ssh_dir.mkdir()
+
+    with pytest.raises(ValueError, match="holds credentials"):
+        check_mountable_dir(ssh_dir)
+
+
+def test_check_mountable_dir_rejects_credential_subdirs(tmp_path: Path) -> None:
+    keys = tmp_path / ".ssh" / "keys"
+    keys.mkdir(parents=True)
+
+    with pytest.raises(ValueError, match="holds credentials"):
+        check_mountable_dir(keys)
+
+
+def test_check_mountable_dir_rejects_system_subdirs() -> None:
+    system_subdir = next((p for p in (Path("/etc/ssl"), Path("/usr/bin")) if p.is_dir()), None)
+    if system_subdir is None:
+        pytest.skip("no system subdirectory on this platform")
+    with pytest.raises(ValueError, match="Refusing to mount"):
+        check_mountable_dir(system_subdir)
+
+
+def test_check_mountable_dir_accepts_a_project_under_the_home_root(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    project = tmp_path / "home" / "dev" / "project"
+    project.mkdir(parents=True)
+    monkeypatch.setattr(Path, "home", classmethod(lambda _cls: tmp_path / "home" / "dev"))
+
+    check_mountable_dir(project)
+
+
+def test_infer_target_type_applies_the_mount_policy() -> None:
+    with pytest.raises(ValueError, match="Refusing to mount"):
+        infer_target_type("/etc")
 
 
 def test_read_target_list_file_strips_blank_lines(tmp_path: Path) -> None:
     target_list = tmp_path / "targets.txt"
     target_list.write_text(
-        "\n"
-        " https://test1.com/ \n"
-        "\n"
-        "http://test2.com:5789/\n"
-        "  \n",
+        "\n https://test1.com/ \n\nhttp://test2.com:5789/\n  \n",
         encoding="utf-8",
     )
 
@@ -178,10 +145,7 @@ def test_read_target_list_file_strips_blank_lines(tmp_path: Path) -> None:
 def test_read_target_list_file_ignores_comment_lines(tmp_path: Path) -> None:
     target_list = tmp_path / "targets.txt"
     target_list.write_text(
-        "# production targets\n"
-        "https://test1.com/\n"
-        "  # staging targets\n"
-        "http://test2.com:5789/\n",
+        "# production targets\nhttps://test1.com/\n  # staging targets\nhttp://test2.com:5789/\n",
         encoding="utf-8",
     )
 
@@ -222,28 +186,12 @@ def test_dedupe_keeps_distinct_targets_in_order() -> None:
     targets = [
         _local_target("/a"),
         {"type": "web_application", "details": {"target_url": "https://x"}},
-        _local_target("/b", mount=True),
+        _local_target("/b"),
     ]
     assert dedupe_local_targets(targets) == targets
 
 
-def test_dedupe_mount_supersedes_copied_same_path() -> None:
-    copied = _local_target("/repo")
-    mounted = _local_target("/repo", mount=True)
-
-    # Copied first, then mounted: the single surviving entry is the mount.
-    result = dedupe_local_targets([copied, mounted])
-    assert len(result) == 1
-    assert result[0]["details"]["mount"] is True
-
-    # Order-independent: mounted first, copied second also yields the mount.
-    result_rev = dedupe_local_targets([mounted, copied])
-    assert len(result_rev) == 1
-    assert result_rev[0]["details"]["mount"] is True
-
-
-def test_dedupe_collapses_duplicate_mounts() -> None:
-    result = dedupe_local_targets(
-        [_local_target("/repo", mount=True), _local_target("/repo", mount=True)]
-    )
-    assert len(result) == 1
+def test_dedupe_collapses_the_same_path() -> None:
+    assert dedupe_local_targets([_local_target("/repo"), _local_target("/repo")]) == [
+        _local_target("/repo")
+    ]

tests/test_session_entries.py

@@ -1,4 +1,4 @@
-"""Tests for build_session_entries: splitting copied vs bind-mounted sources."""
+"""Tests for how local sources reach the sandbox: bind mounts or manifest upload."""
 
 from __future__ import annotations
 
@@ -6,82 +6,175 @@ from typing import TYPE_CHECKING, Any
 
 from agents.sandbox.entries import LocalDir
 
-from strix.runtime.session_manager import build_session_entries
+from strix.runtime.backends import (
+    _BACKENDS,
+    _BIND_MOUNT_BACKENDS,
+    backend_supports_bind_mounts,
+    register_backend,
+)
+from strix.runtime.session_manager import build_bind_mounts, build_manifest_entries
 
 
 if TYPE_CHECKING:
     from pathlib import Path
 
 
-def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
-    return {"source_path": path, "workspace_subdir": subdir, "mount": mount}
+def _source(subdir: str, path: str, *, protect_metadata: bool = False) -> dict[str, Any]:
+    return {"source_path": path, "workspace_subdir": subdir, "protect_metadata": protect_metadata}
 
 
-def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
-    entries, bind_mounts, staged_dirs = build_session_entries([_source("repo", str(tmp_path))])
-
-    assert bind_mounts == []
-    assert staged_dirs == []
-    assert isinstance(entries["repo"], LocalDir)
-    assert entries["repo"].src == tmp_path.resolve()
-
-
-def test_mounted_source_becomes_bind_mount(tmp_path: Path) -> None:
-    entries, bind_mounts, _staged = build_session_entries(
-        [_source("repo", str(tmp_path), mount=True)]
-    )
-
-    assert entries == {}
-    assert bind_mounts == [
+def test_source_becomes_writable_bind_mount(tmp_path: Path) -> None:
+    assert build_bind_mounts([_source("repo", str(tmp_path))]) == [
         {
             "source": str(tmp_path.resolve()),
             "target": "/workspace/repo",
-            "read_only": True,
+            "read_only": False,
         }
     ]
 
 
-def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
-    copied = tmp_path / "copied"
-    mounted = tmp_path / "mounted"
-    copied.mkdir()
-    mounted.mkdir()
+def test_git_dir_is_remounted_read_only_when_protected(tmp_path: Path) -> None:
+    (tmp_path / ".git").mkdir()
 
-    entries, bind_mounts, _staged = build_session_entries(
-        [
-            _source("copied", str(copied)),
-            _source("mounted", str(mounted), mount=True),
-        ]
-    )
+    mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
 
-    assert list(entries) == ["copied"]
-    assert isinstance(entries["copied"], LocalDir)
-    assert [m["target"] for m in bind_mounts] == ["/workspace/mounted"]
+    assert mounts == [
+        {"source": str(tmp_path.resolve()), "target": "/workspace/repo", "read_only": False},
+        {
+            "source": str((tmp_path / ".git").resolve()),
+            "target": "/workspace/repo/.git",
+            "read_only": True,
+        },
+    ]
+
+
+def test_agent_instruction_dirs_are_protected_too(tmp_path: Path) -> None:
+    (tmp_path / ".agents").mkdir()
+    (tmp_path / ".codex").mkdir()
+
+    mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
+
+    assert [(m["target"], m["read_only"]) for m in mounts] == [
+        ("/workspace/repo", False),
+        ("/workspace/repo/.agents", True),
+        ("/workspace/repo/.codex", True),
+    ]
+
+
+def test_worktree_git_pointer_file_is_protected(tmp_path: Path) -> None:
+    gitdir = tmp_path / "nested" / "gitdir"
+    gitdir.mkdir(parents=True)
+    (tmp_path / ".git").write_text(f"gitdir: {gitdir}\n", encoding="utf-8")
+
+    mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
+
+    assert [(m["target"], m["read_only"]) for m in mounts] == [
+        ("/workspace/repo", False),
+        ("/workspace/repo/.git", True),
+        ("/workspace/repo/nested/gitdir", True),
+    ]
+
+
+def test_git_pointer_to_a_missing_gitdir_is_not_mounted(tmp_path: Path) -> None:
+    (tmp_path / ".git").write_text(f"gitdir: {tmp_path / 'gone'}\n", encoding="utf-8")
+
+    mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
+
+    assert [m["target"] for m in mounts] == ["/workspace/repo", "/workspace/repo/.git"]
+
+
+def test_git_pointer_outside_the_tree_needs_no_nested_mount(tmp_path: Path) -> None:
+    tree = tmp_path / "worktree"
+    tree.mkdir()
+    (tree / ".git").write_text(f"gitdir: {tmp_path / 'main' / '.git'}\n", encoding="utf-8")
+
+    mounts = build_bind_mounts([_source("repo", str(tree), protect_metadata=True)])
+
+    assert [m["target"] for m in mounts] == ["/workspace/repo", "/workspace/repo/.git"]
+
+
+def test_metadata_symlinked_outside_the_tree_is_not_mounted(tmp_path: Path) -> None:
+    outside = tmp_path / "elsewhere"
+    outside.mkdir()
+    tree = tmp_path / "repo"
+    tree.mkdir()
+    (tree / ".git").symlink_to(outside, target_is_directory=True)
+
+    mounts = build_bind_mounts([_source("repo", str(tree), protect_metadata=True)])
+
+    assert [m["target"] for m in mounts] == ["/workspace/repo"]
+
+
+def test_no_git_guard_without_a_git_dir(tmp_path: Path) -> None:
+    mounts = build_bind_mounts([_source("repo", str(tmp_path), protect_metadata=True)])
+    assert [m["target"] for m in mounts] == ["/workspace/repo"]
+
+
+def test_clone_keeps_its_git_writable(tmp_path: Path) -> None:
+    (tmp_path / ".git").mkdir()
+    mounts = build_bind_mounts([_source("clone", str(tmp_path), protect_metadata=False)])
+    assert [m["target"] for m in mounts] == ["/workspace/clone"]
+
+
+def test_multiple_sources_each_get_a_mount(tmp_path: Path) -> None:
+    first = tmp_path / "first"
+    second = tmp_path / "second"
+    first.mkdir()
+    second.mkdir()
+
+    mounts = build_bind_mounts([_source("first", str(first)), _source("second", str(second))])
+
+    assert [m["target"] for m in mounts] == ["/workspace/first", "/workspace/second"]
+    assert all(m["read_only"] is False for m in mounts)
 
 
 def test_incomplete_sources_are_skipped() -> None:
-    entries, bind_mounts, staged_dirs = build_session_entries(
-        [
-            {"source_path": "", "workspace_subdir": "x"},
-            {"source_path": "/p", "workspace_subdir": ""},
-        ]
+    assert (
+        build_bind_mounts(
+            [
+                {"source_path": "", "workspace_subdir": "x"},
+                {"source_path": "/p", "workspace_subdir": ""},
+            ]
+        )
+        == []
     )
-    assert entries == {}
-    assert bind_mounts == []
-    assert staged_dirs == []
 
 
-def test_symlink_tree_is_staged(tmp_path: Path) -> None:
-    repo = tmp_path / "repo"
-    repo.mkdir()
-    (repo / "real.txt").write_text("content")
-    (repo / "link.txt").symlink_to(repo / "real.txt")
+def test_manifest_entries_upload_sources_for_backends_without_bind_mounts(
+    tmp_path: Path,
+) -> None:
+    entries = build_manifest_entries([_source("repo", str(tmp_path), protect_metadata=True)])
 
-    entries, _mounts, staged_dirs = build_session_entries([_source("repo", str(repo))])
-
-    assert len(staged_dirs) == 1
+    assert set(entries) == {"repo"}
     entry = entries["repo"]
     assert isinstance(entry, LocalDir)
-    assert entry.src == staged_dirs[0]
-    assert not (staged_dirs[0] / "link.txt").is_symlink()
-    assert (staged_dirs[0] / "link.txt").read_text() == "content"
+    assert entry.src == tmp_path.resolve()
+
+
+def test_manifest_entries_skip_incomplete_sources() -> None:
+    assert (
+        build_manifest_entries(
+            [
+                {"source_path": "", "workspace_subdir": "x"},
+                {"source_path": "/p", "workspace_subdir": ""},
+            ]
+        )
+        == {}
+    )
+
+
+def test_only_bind_mount_capable_backends_are_registered_as_such() -> None:
+    assert backend_supports_bind_mounts("docker")
+    assert not backend_supports_bind_mounts("e2b")
+
+    async def _remote_backend(**_kwargs: Any) -> tuple[Any, Any]:
+        return object(), object()
+
+    try:
+        register_backend("e2b", _remote_backend)
+        assert not backend_supports_bind_mounts("e2b")
+        register_backend("e2b", _remote_backend, supports_bind_mounts=True)
+        assert backend_supports_bind_mounts("e2b")
+    finally:
+        _BACKENDS.pop("e2b", None)
+        _BIND_MOUNT_BACKENDS.discard("e2b")