fix(prompt): trim demo-environment check to a short addendum

The sandbox never creates /workspace/scratch, so guidance pointing agents
there failed on first write. Make the Python/exec_command and recon
output-hygiene guidance path-agnostic (write to a file, relative to the
working dir) instead of naming a directory that may not exist.

strix/agents/prompts/system_prompt.jinja

@@ -168,9 +168,9 @@ EFFICIENCY TACTICS:
 - Download additional tools as needed for specific tasks
 - Run multiple scans in parallel when possible
 - Load the most relevant skill before starting a specialized testing workflow if doing so will improve accuracy, speed, or tool usage
-- Use `exec_command` for Python code: write reusable scripts under
-  `/workspace/scratch/` and run them with `python3`. For one-off snippets,
-  `python3 -c` or a here-document is acceptable.
+- Use `exec_command` for Python code: write reusable scripts to a file and
+  run them with `python3`. For one-off snippets, `python3 -c` or a
+  here-document is acceptable.
 - For Caido proxy automation inside Python, explicitly import from
   `caido_api`:
   `from caido_api import list_requests, view_request, repeat_request, list_sitemap, view_sitemap_entry, scope_rules`
@@ -186,7 +186,7 @@ EFFICIENCY TACTICS:
 VALIDATION REQUIREMENTS:
 - Full validation required - no assumptions
 - Demonstrate concrete impact with evidence
-- Consider business context for severity assessment
+- Consider business context for severity assessment — check whether the target is a demo/sandbox environment or content meant to be public, and factor that in
 - Independent verification through subagent
 - Document complete attack chain
 - Keep going until you find something that matters
@@ -240,12 +240,18 @@ AGENT ISOLATION & SANDBOXING:
 - All agents share the same /workspace directory and proxy history
 - Agents can see each other's files and proxy traffic for better collaboration
 
+DISK & SCRATCH HYGIENE:
+- /workspace is a shared, finite disk used by all agents at once — be a considerate tenant
+- Prefer bounded recon: scope crawls and scans by depth, duration, and target rather than "collect everything"
+- Redirect large tool output to a file, and once you've extracted what you need (e.g. a URL/endpoint list), remove the raw output
+- If disk gets tight or a write fails for space, check what's large under /workspace and clean up files from your own task; leave another agent's files unless you've confirmed they're no longer in use
+
 MANDATORY INITIAL PHASES:
 
 BLACK-BOX TESTING - PHASE 1 (RECON & MAPPING):
 - COMPLETE full reconnaissance: subdomain enumeration, port scanning, service detection
 - MAP entire attack surface: all endpoints, parameters, APIs, forms, inputs
-- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files
+- CRAWL thoroughly: spider all pages (authenticated and unauthenticated), discover hidden paths, analyze JS files — keep each crawl bounded by depth/duration, and tidy up raw output once endpoints are extracted
 - ENUMERATE technologies: frameworks, libraries, versions, dependencies
 - Reconnaissance should normally happen before targeted vulnerability discovery unless the correct next move is already obvious or the user/system explicitly asks to prioritize a specific area first
 - ONLY AFTER comprehensive mapping → proceed to vulnerability testing

strix/config/settings.py

@@ -56,6 +56,8 @@ class RuntimeSettings(BaseSettings):
     # 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")
 
 
 class TelemetrySettings(BaseSettings):

strix/core/agents.py

@@ -10,6 +10,8 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import TYPE_CHECKING, Any, Literal, cast
 
+from strix.core.sessions import session_write_lock
+
 
 if TYPE_CHECKING:
     from agents.items import TResponseInputItem
@@ -137,7 +139,8 @@ class AgentCoordinator:
             )
             return False
         try:
-            await session.add_items([self._message_to_session_item(message)])
+            async with session_write_lock(session):
+                await session.add_items([self._message_to_session_item(message)])
         except Exception:
             logger.exception(
                 "agent.send failed to append to SDK session target=%s",

strix/core/execution.py

@@ -17,7 +17,11 @@ from openai import APIError
 
 from strix.core.hooks import BudgetExceededError
 from strix.core.inputs import child_initial_input
-from strix.core.sessions import open_agent_session, strip_all_images_from_session
+from strix.core.sessions import (
+    enforce_image_budget,
+    open_agent_session,
+    strip_all_images_from_session,
+)
 
 
 if TYPE_CHECKING:
@@ -349,6 +353,13 @@ async def _run_cycle(  # noqa: PLR0912, PLR0915
     while True:
         try:
             await coordinator.mark_running(agent_id)
+            if session is not None:
+                max_images = context.get("max_context_images")
+                if isinstance(max_images, int):
+                    try:
+                        await enforce_image_budget(session, max_images)
+                    except Exception:
+                        logger.exception("image-budget enforcement failed for %s", agent_id)
             stream = Runner.run_streamed(
                 agent,
                 input=input_data,

strix/core/inputs.py

@@ -13,6 +13,7 @@ from strix.config.models import (
     is_known_openai_bare_model,
     model_supports_reasoning,
 )
+from strix.core.sessions import scrub_images_from_items
 
 
 if TYPE_CHECKING:
@@ -161,7 +162,11 @@ def child_initial_input(
     """
     parts: list[str] = []
     if parent_history:
-        rendered = json.dumps(parent_history, ensure_ascii=False, default=str)
+        rendered = json.dumps(
+            scrub_images_from_items(parent_history),
+            ensure_ascii=False,
+            default=str,
+        )
         parts.append(
             "== Inherited context from parent (background only) ==\n"
             f"{rendered}\n"

strix/core/runner.py

@@ -287,6 +287,7 @@ async def run_strix_scan(
             "parent_id": None,
             "interactive": interactive,
             "spawn_child_agent": spawn_child_agent,
+            "max_context_images": settings.runtime.max_context_images,
         }
 
         root_session = open_agent_session(root_id, agents_db)

strix/core/sessions.py

@@ -2,64 +2,149 @@
 
 from __future__ import annotations
 
-import contextlib
+import asyncio
+import logging
 from typing import TYPE_CHECKING, Any, cast
+from weakref import WeakKeyDictionary
 
 from agents.memory import SQLiteSession
 
 
 if TYPE_CHECKING:
+    from collections.abc import Callable
     from pathlib import Path
 
     from agents.items import TResponseInputItem
     from agents.memory import Session
 
 
+logger = logging.getLogger(__name__)
+
+
 def open_agent_session(agent_id: str, path: Path) -> SQLiteSession:
     path.parent.mkdir(parents=True, exist_ok=True)
     return SQLiteSession(session_id=agent_id, db_path=path)
 
 
 _IMAGE_REJECTED_TEXT = "[image rejected by the model]"
+_IMAGE_ELIDED_TEXT = "[older screenshot elided to bound context memory]"
+_INHERITED_IMAGE_TEXT = "[screenshot omitted from inherited context]"
+
+
+def _output_has_image(item_dict: dict[str, Any]) -> bool:
+    return (
+        item_dict.get("type") == "function_call_output"
+        and isinstance(item_dict.get("output"), list)
+        and any(isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"])
+    )
+
+
+def _elided_output(item_dict: dict[str, Any], text: str) -> dict[str, Any]:
+    # Replace only image blocks; sibling text blocks are preserved.
+    output = item_dict.get("output")
+    blocks = output if isinstance(output, list) else []
+    return {
+        "type": "function_call_output",
+        "call_id": item_dict.get("call_id"),
+        "output": [
+            {"type": "input_text", "text": text}
+            if isinstance(block, dict) and block.get("type") == "input_image"
+            else block
+            for block in blocks
+        ],
+    }
+
+
+_session_write_locks: WeakKeyDictionary[Session, asyncio.Lock] = WeakKeyDictionary()
+
+
+def session_write_lock(session: Session) -> asyncio.Lock:
+    """Lock serialising all out-of-band writes to ``session``."""
+    lock = _session_write_locks.get(session)
+    if lock is None:
+        lock = asyncio.Lock()
+        _session_write_locks[session] = lock
+    return lock
+
+
+async def _rewrite_session(
+    session: Session,
+    transform: Callable[[list[Any]], tuple[list[Any], bool]],
+) -> bool:
+    """Read-modify-write a session under its write lock, restoring on failure."""
+    async with session_write_lock(session):
+        items = await session.get_items()
+        if not items:
+            return False
+        rebuilt, changed = transform(list(items))
+        if not changed:
+            return False
+        rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
+        original_items = cast("list[TResponseInputItem]", list(items))
+        await session.clear_session()
+        try:
+            await session.add_items(rebuilt_items)
+        except Exception:
+            logger.exception("session rewrite failed; restoring original items")
+            await session.clear_session()
+            await session.add_items(original_items)
+            raise
+        return True
 
 
 async def strip_all_images_from_session(session: Session) -> bool:
-    items = await session.get_items()
-    if not items:
+    """Replace every image tool output with a text placeholder (rejection recovery)."""
+
+    def _transform(items: list[Any]) -> tuple[list[Any], bool]:
+        rebuilt: list[Any] = []
+        changed = False
+        for item in items:
+            item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
+            if item_dict is not None and _output_has_image(item_dict):
+                rebuilt.append(_elided_output(item_dict, _IMAGE_REJECTED_TEXT))
+                changed = True
+            else:
+                rebuilt.append(item)
+        return rebuilt, changed
+
+    return await _rewrite_session(session, _transform)
+
+
+async def enforce_image_budget(session: Session, max_images: int) -> bool:
+    """Keep only the most recent ``max_images`` image outputs; elide older ones."""
+    if max_images < 0:
         return False
 
-    rebuilt: list[Any] = []
-    changed = False
-    for item in items:
-        item_dict = cast("dict[str, Any]", item) if isinstance(item, dict) else None
-        if (
-            item_dict is not None
-            and item_dict.get("type") == "function_call_output"
-            and isinstance(item_dict.get("output"), list)
-            and any(
-                isinstance(b, dict) and b.get("type") == "input_image" for b in item_dict["output"]
-            )
-        ):
-            rebuilt.append(
-                {
-                    "type": "function_call_output",
-                    "call_id": item_dict.get("call_id"),
-                    "output": [{"type": "input_text", "text": _IMAGE_REJECTED_TEXT}],
-                },
-            )
-            changed = True
-        else:
-            rebuilt.append(item)
+    def _transform(items: list[Any]) -> tuple[list[Any], bool]:
+        image_indices = [
+            i
+            for i, item in enumerate(items)
+            if isinstance(item, dict) and _output_has_image(cast("dict[str, Any]", item))
+        ]
+        if len(image_indices) <= max_images:
+            return items, False
+        to_elide = set(image_indices[: len(image_indices) - max_images])
+        rebuilt = [
+            _elided_output(cast("dict[str, Any]", item), _IMAGE_ELIDED_TEXT)
+            if i in to_elide
+            else item
+            for i, item in enumerate(items)
+        ]
+        return rebuilt, True
 
-    if not changed:
-        return False
+    return await _rewrite_session(session, _transform)
 
-    rebuilt_items = cast("list[TResponseInputItem]", rebuilt)
-    await session.clear_session()
-    try:
-        await session.add_items(rebuilt_items)
-    except Exception:
-        with contextlib.suppress(Exception):
-            await session.add_items(rebuilt_items)
-        raise
-    return True
+
+def scrub_images_from_items(items: list[Any]) -> list[Any]:
+    """Return a copy of ``items`` with every image block replaced by text."""
+
+    def _scrub(obj: Any) -> Any:
+        if isinstance(obj, dict):
+            if obj.get("type") == "input_image":
+                return {"type": "input_text", "text": _INHERITED_IMAGE_TEXT}
+            return {k: _scrub(v) for k, v in obj.items()}
+        if isinstance(obj, list):
+            return [_scrub(v) for v in obj]
+        return obj
+
+    return [_scrub(item) for item in items]

strix/runtime/docker_client.py

@@ -24,20 +24,25 @@ from __future__ import annotations
 
 import contextlib
 import logging
+import os
 import uuid
-from typing import Any
+from typing import Any, cast
 
+from agents.sandbox.errors import ExposedPortUnavailableError
 from agents.sandbox.manifest import Manifest
 from agents.sandbox.sandboxes.docker import (
     DockerSandboxClient,
+    DockerSandboxSession,
     _build_docker_volume_mounts,
     _docker_port_key,
     _manifest_requires_fuse,
     _manifest_requires_sys_admin,
 )
 from agents.sandbox.session.sandbox_session import SandboxSession
+from agents.sandbox.types import ExposedPortEndpoint
 from docker import errors as docker_errors  # type: ignore[import-untyped, unused-ignore]
 from docker.models.containers import Container  # type: ignore[import-untyped, unused-ignore]
+from docker.types import LogConfig  # type: ignore[import-untyped, unused-ignore]
 from docker.types import Mount as DockerSDKMount  # type: ignore[import-untyped, unused-ignore]
 from docker.utils import parse_repository_tag  # type: ignore[import-untyped, unused-ignore]
 from requests.exceptions import RequestException
@@ -46,6 +51,103 @@ from requests.exceptions import RequestException
 logger = logging.getLogger(__name__)
 
 
+_SANDBOX_NETWORK_ENV = "STRIX_DOCKER_SANDBOX_NETWORK"
+
+
+def _sandbox_network() -> str | None:
+    value = os.environ.get(_SANDBOX_NETWORK_ENV, "").strip()
+    return value or None
+
+
+def _apply_sandbox_network(create_kwargs: dict[str, Any]) -> None:
+    network = _sandbox_network()
+    if network:
+        create_kwargs["network"] = network
+        create_kwargs.pop("ports", None)
+
+
+def _apply_resource_limits(create_kwargs: dict[str, Any]) -> None:
+    """Apply optional cgroup resource caps from the environment. Unset/blank
+    values leave docker's default (unbounded), so this is opt-in per host."""
+    mem_limit = os.environ.get("STRIX_SANDBOX_MEM_LIMIT", "").strip()
+    if mem_limit:
+        create_kwargs["mem_limit"] = mem_limit
+
+    shm_size = os.environ.get("STRIX_SANDBOX_SHM_SIZE", "").strip()
+    if shm_size:
+        create_kwargs["shm_size"] = shm_size
+
+    cpus = os.environ.get("STRIX_SANDBOX_CPUS", "").strip()
+    if cpus:
+        with contextlib.suppress(ValueError, OverflowError):
+            nano_cpus = int(float(cpus) * 1_000_000_000)
+            if 0 < nano_cpus <= 2**63 - 1:
+                create_kwargs["nano_cpus"] = nano_cpus
+
+    pids_limit = os.environ.get("STRIX_SANDBOX_PIDS_LIMIT", "").strip()
+    if pids_limit:
+        with contextlib.suppress(ValueError):
+            create_kwargs["pids_limit"] = int(pids_limit)
+
+
+def _apply_log_limits(create_kwargs: dict[str, Any]) -> None:
+    """Bound the container's json-file log so a runaway process in the sandbox
+    (e.g. a tool that busy-loops writing to stdout) cannot fill the host disk
+    and take the Docker daemon down with it.
+
+    Unlike the cgroup caps above, this defaults **on** — docker's own default
+    is an unbounded json-file, which is unsafe for an autonomous agent that
+    executes arbitrary commands. ``max-file`` rotation means the on-disk cap is
+    ``max-size * max-file``. Set ``STRIX_SANDBOX_LOG_MAX_SIZE`` to ``0``/``off``
+    to opt back out to docker's default."""
+    max_size = os.environ.get("STRIX_SANDBOX_LOG_MAX_SIZE", "50m").strip()
+    if max_size.lower() in ("0", "off", "none", "unlimited"):
+        return
+    max_file = os.environ.get("STRIX_SANDBOX_LOG_MAX_FILE", "3").strip() or "3"
+    create_kwargs["log_config"] = LogConfig(
+        type=LogConfig.types.JSON,
+        config={"max-size": max_size, "max-file": max_file},
+    )
+
+
+class StrixDockerSandboxSession(DockerSandboxSession):
+    sandbox_network: str = ""
+
+    async def _resolve_exposed_port(self, port: int) -> ExposedPortEndpoint:
+        try:
+            self._container.reload()
+        except docker_errors.APIError as e:
+            raise ExposedPortUnavailableError(
+                port=port,
+                exposed_ports=self.state.exposed_ports,
+                reason="backend_unavailable",
+                context={
+                    "backend": "docker",
+                    "detail": "container_reload_failed",
+                    "network": self.sandbox_network,
+                },
+                cause=e,
+            ) from e
+
+        attrs = getattr(self._container, "attrs", {}) or {}
+        networks = attrs.get("NetworkSettings", {}).get("Networks", {})
+        endpoint = networks.get(self.sandbox_network) or {}
+        ip = endpoint.get("IPAddress") or endpoint.get("GlobalIPv6Address")
+        if not isinstance(ip, str) or not ip:
+            raise ExposedPortUnavailableError(
+                port=port,
+                exposed_ports=self.state.exposed_ports,
+                reason="backend_unavailable",
+                context={
+                    "backend": "docker",
+                    "detail": "container_not_on_network",
+                    "network": self.sandbox_network,
+                },
+            )
+        host = f"[{ip}]" if ":" in ip else ip
+        return ExposedPortEndpoint(host=host, port=port, tls=False)
+
+
 class StrixDockerSandboxClient(DockerSandboxClient):
     # Host directories to bind-mount into the container, set by the docker
     # backend before ``create()``. Each item is ``{source, target, read_only}``.
@@ -117,6 +219,10 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         extra_hosts = create_kwargs.setdefault("extra_hosts", {})
         extra_hosts["host.docker.internal"] = "host-gateway"
 
+        _apply_sandbox_network(create_kwargs)
+        _apply_resource_limits(create_kwargs)
+        _apply_log_limits(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", ())
@@ -146,6 +252,15 @@ class StrixDockerSandboxClient(DockerSandboxClient):
         )
         return container
 
+    async def create(self, **kwargs: Any) -> SandboxSession:
+        session = await super().create(**kwargs)
+        network = _sandbox_network()
+        inner = session._inner
+        if network and isinstance(inner, DockerSandboxSession):
+            inner.__class__ = StrixDockerSandboxSession
+            cast("StrixDockerSandboxSession", inner).sandbox_network = network
+        return session
+
     async def delete(self, session: SandboxSession) -> SandboxSession:
         container_id = getattr(getattr(session._inner, "state", None), "container_id", None)
         if container_id:

strix/runtime/local_dir_staging.py

@@ -0,0 +1,120 @@
+"""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))
+    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,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import shutil
 from pathlib import Path
 from typing import Any
 
@@ -12,6 +13,7 @@ from agents.sandbox.manifest import Environment, Manifest
 from strix.config import load_settings
 from strix.runtime.backends import get_backend
 from strix.runtime.caido_bootstrap import bootstrap_caido
+from strix.runtime.local_dir_staging import stage_symlink_safe_dir
 
 
 logger = logging.getLogger(__name__)
@@ -29,16 +31,20 @@ _WORKSPACE_ROOT = "/workspace"
 
 def build_session_entries(
     local_sources: list[dict[str, Any]],
-) -> tuple[dict[str | Path, BaseEntry], 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.
+    ``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] = {}
     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 ""
@@ -54,8 +60,11 @@ def build_session_entries(
                 }
             )
         else:
-            entries[ws_subdir] = LocalDir(src=resolved)
-    return entries, bind_mounts
+            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
 
 
 async def create_or_reuse(
@@ -75,7 +84,7 @@ async def create_or_reuse(
         logger.info("Reusing existing sandbox session for scan %s", scan_id)
         return cached
 
-    entries, bind_mounts = build_session_entries(local_sources)
+    entries, bind_mounts, staged_dirs = build_session_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.)
@@ -106,12 +115,16 @@ async def create_or_reuse(
         backend_name,
         image,
     )
-    client, session = await backend(
-        image=image,
-        manifest=manifest,
-        exposed_ports=(_CONTAINER_CAIDO_PORT,),
-        bind_mounts=bind_mounts,
-    )
+    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)
 
     caido_endpoint = await session.resolve_exposed_port(_CONTAINER_CAIDO_PORT)
     scheme = "https" if caido_endpoint.tls else "http"
@@ -154,11 +167,19 @@ async def cleanup(scan_id: str) -> None:
         except Exception:  # noqa: BLE001
             logger.debug("cleanup(%s): caido_client.aclose() raised", scan_id, exc_info=True)
 
+    client = bundle["client"]
     try:
-        await bundle["client"].delete(bundle["session"])
+        await client.delete(bundle["session"])
         logger.info("Cleaned up sandbox session for scan %s", scan_id)
     except Exception:
         logger.exception(
             "cleanup(%s): client.delete raised; container may need manual reaping",
             scan_id,
         )
+
+    docker_client = getattr(client, "docker_client", None)
+    if docker_client is not None:
+        try:
+            docker_client.close()
+        except Exception:  # noqa: BLE001
+            logger.debug("cleanup(%s): docker_client.close() raised", scan_id, exc_info=True)

strix/skills/tooling/katana.md

@@ -24,7 +24,15 @@ High-signal flags:
 - `-p, -parallelism <n>` concurrent input targets
 - `-rl, -rate-limit <n>` request rate limit
 - `-timeout <seconds>` request timeout
+- `-ct, -crawl-duration <s|m|h|d>` maximum time to crawl the target
 - `-retry <n>` retry count
+- `-mdp, -max-domain-pages <n>` cap pages crawled per domain (default: unlimited)
+- `-fsu, -filter-similar` collapse similar URLs (e.g. /users/123 and /users/456)
+- `-fs, -field-scope <dn|rdn|fqdn|regex>` crawl scope (default `rdn` = root domain + ALL subdomains)
+- `-f, -field <url|path|...>` emit only one field (e.g. `-f url` for a plain URL list)
+- `-or, -omit-raw` omit raw request/response from JSONL output
+- `-ob, -omit-body` omit response body from JSONL output
+- `-mrs, -max-response-size <bytes>` cap per-response bytes read (default 4194304)
 - `-ef, -extension-filter <list>` extension exclusions
 - `-tlsi, -tls-impersonate` experimental JA3/TLS impersonation
 - `-hl, -headless` enable hybrid headless crawling
@@ -37,13 +45,13 @@ High-signal flags:
 - `-silent`, `-j, -jsonl`, `-o <file>` output controls
 
 Agent-safe baseline for automation:
-`mkdir -p crawl && katana -u https://target.tld -d 3 -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
+`mkdir -p crawl && katana -u https://target.tld -d 3 -ct 10m -mdp 2000 -fsu -jc -kf robotstxt -c 10 -p 10 -rl 50 -timeout 10 -retry 1 -ef png,jpg,jpeg,gif,svg,css,woff,woff2,ttf,eot,map -silent -j -o crawl/katana.jsonl`
 
 Common patterns:
 - Fast crawl baseline:
   `katana -u https://target.tld -d 3 -jc -silent`
-- Deeper JS-aware crawl:
-  `katana -u https://target.tld -d 5 -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
+- Deeper JS-aware crawl (narrowed target; keep it time-bounded):
+  `katana -u https://target.tld -d 5 -ct 15m -jc -jsl -kf all -c 10 -p 10 -rl 50 -o katana_urls.txt`
 - Multi-target run with JSONL output:
   `katana -list urls.txt -d 3 -jc -silent -j -o katana.jsonl`
 - Headless crawl with local Chrome:
@@ -59,6 +67,13 @@ Critical correctness rules:
 - For `-kf`, keep depth at least `-d 3` so known files are fully covered.
 - If writing to a file, ensure parent directory exists before `-o`.
 
+Keeping output small (katana has NO default page cap, so plan for volume):
+- Bound scope and volume: `-fs fqdn` (or `-cs`/`-cos` regex) so the crawl doesn't wander across every subdomain, `-mdp <n>` to cap pages per domain, `-fsu` to collapse near-identical URLs, and `-ct`/`-d` to bound time and depth.
+- Shrink each record: default JSONL is verbose. If you only need endpoints, emit a plain URL list with `-f url` instead of `-j`. If you need JSONL, drop the heavy parts with `-or` (omit raw) and `-ob` (omit body), and lower `-mrs` to cap per-response bytes.
+- Reserve `-jsl` / `-kf all` / higher `-d` for a specific narrowed target — they multiply output fast on large sites.
+- Reduce, then delete: once the crawl finishes, extract just what you need (e.g. `katana ... -f url -o urls.txt` or `sort -u` a URL list, or a short note of interesting paths) and remove the raw crawl file/dir. Don't keep large raw crawls around after you've distilled them.
+- Sanity-check size (`du -sh <out>`); if it's outsized for the scope, tighten `-fs`/`-mdp`/`-fsu`/`-d`/`-ct` and re-run rather than keeping it.
+
 Usage rules:
 - Keep `-d`, `-c`, `-p`, and `-rl` explicit for reproducible runs.
 - Use `-ef` early to reduce static-file noise before fuzzing.

strix/skills/tooling/python.md

@@ -7,9 +7,9 @@ description: Run Python through exec_command in the SDK sandbox. Use the image-b
 
 Use `exec_command` for Python. There is no separate Strix Python executor.
 
-Prefer writing reusable scripts to `/workspace/scratch/<name>.py` and
-running them with `python3 /workspace/scratch/<name>.py`. For short
-one-off transformations, `python3 -c` or a small here-document is fine.
+Prefer writing reusable scripts to a `.py` file and running them with
+`python3 <name>.py`. For short one-off transformations, `python3 -c` or a
+small here-document is fine.
 
 The `shell` parameter on `exec_command` is for swapping POSIX shells
 (`bash`/`zsh`/`sh`), not for picking interpreters. Put the interpreter
@@ -84,8 +84,9 @@ automatically, so it shows up in `list_requests` and you can use
 For iterative exploit work, put code in a file:
 
 ```text
-1. Create or edit `/workspace/scratch/exploit.py` with `apply_patch`.
-2. Run it with `exec_command`: `python3 /workspace/scratch/exploit.py`.
+1. Create or edit a task-unique script (e.g. `poc_<task-id>.py`, so it can't
+   clobber a project file or another agent's script) with `apply_patch`.
+2. Run it with `exec_command`: `python3 poc_<task-id>.py`.
 3. Edit and rerun until the proof-of-concept is reliable.
 ```
 

strix/tools/agents_graph/tools.py

@@ -229,7 +229,8 @@ async def wait_for_message(  # noqa: PLR0911
     Use when you have nothing useful to do until a child/peer responds
     — typically after spawning subagents and you want to wait for
     their completion reports. The agent automatically resumes when any
-    message arrives.
+    message arrives, so pick a ``timeout_seconds`` proportional to the
+    work you're awaiting.
 
     **Critical caveats:**
 
@@ -246,9 +247,19 @@ async def wait_for_message(  # noqa: PLR0911
         reason: One-line note shown in graph snapshots while you're
             waiting (helps a human or sibling agent debug who's stuck
             on what).
-        timeout_seconds: Hard cap (default 600s). On timeout the tool
-            returns and you decide whether to keep working or wait
-            again.
+        timeout_seconds: Max seconds to wait (default 600). This is only
+            a cap — the tool returns the INSTANT a message arrives, so a
+            larger value never makes you wait longer when the reply does
+            come. Right-size it to what you're waiting on: a short wait
+            (e.g. 10-60s) for a quick ack or a small/fast subtask, and a
+            longer one (e.g. ~100-200s) only for genuinely long-running
+            work (deep recon, exploitation, a full sub-scan). The cap only
+            bites when the expected message never arrives — so an oversized
+            timeout on a trivial wait just strands you idle until it
+            elapses. On timeout the tool returns and you decide whether to
+            keep working or wait again. (Applies to autonomous multi-agent
+            runs; in interactive/chat sessions the agent instead parks until
+            a message arrives and this cap is not enforced.)
     """
     inner = _ctx(ctx)
     coordinator = coordinator_from_context(inner)

tests/test_local_dir_staging.py

@@ -0,0 +1,108 @@
+"""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()

tests/test_runner_rate_limit.py

@@ -37,7 +37,8 @@ async def test_persistent_rate_limit_stops_gracefully(
             model="openai/gpt-4o",
             reasoning_effort="high",
             force_required_tool_choice=False,
-        )
+        ),
+        runtime=types.SimpleNamespace(max_context_images=3),
     )
     monkeypatch.setattr(runner, "load_settings", lambda: settings)
     monkeypatch.setattr(runner, "configure_sdk_model_defaults", lambda _settings: None)
@@ -54,8 +55,8 @@ async def test_persistent_rate_limit_stops_gracefully(
     async def _cleanup(*_args: Any, **_kwargs: Any) -> None:
         return None
 
-    monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)
-    monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)
+    monkeypatch.setattr(runner.session_manager, "create_or_reuse", _create_or_reuse)  # type: ignore[attr-defined]
+    monkeypatch.setattr(runner.session_manager, "cleanup", _cleanup)  # type: ignore[attr-defined]
 
     monkeypatch.setattr(runner, "build_root_task", lambda _scan_config: "task")
     monkeypatch.setattr(runner, "build_scope_context", lambda _scan_config: "")

tests/test_session_entries.py

@@ -18,15 +18,18 @@ def _source(subdir: str, path: str, *, mount: bool = False) -> dict[str, Any]:
 
 
 def test_copied_source_becomes_localdir_entry(tmp_path: Path) -> None:
-    entries, bind_mounts = build_session_entries([_source("repo", str(tmp_path))])
+    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 = build_session_entries([_source("repo", str(tmp_path), mount=True)])
+    entries, bind_mounts, _staged = build_session_entries(
+        [_source("repo", str(tmp_path), mount=True)]
+    )
 
     assert entries == {}
     assert bind_mounts == [
@@ -44,7 +47,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
     copied.mkdir()
     mounted.mkdir()
 
-    entries, bind_mounts = build_session_entries(
+    entries, bind_mounts, _staged = build_session_entries(
         [
             _source("copied", str(copied)),
             _source("mounted", str(mounted), mount=True),
@@ -57,7 +60,7 @@ def test_mixed_sources_split_correctly(tmp_path: Path) -> None:
 
 
 def test_incomplete_sources_are_skipped() -> None:
-    entries, bind_mounts = build_session_entries(
+    entries, bind_mounts, staged_dirs = build_session_entries(
         [
             {"source_path": "", "workspace_subdir": "x"},
             {"source_path": "/p", "workspace_subdir": ""},
@@ -65,3 +68,20 @@ def test_incomplete_sources_are_skipped() -> None:
     )
     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")
+
+    entries, _mounts, staged_dirs = build_session_entries([_source("repo", str(repo))])
+
+    assert len(staged_dirs) == 1
+    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"