Upload folder using huggingface_hub

Dockerfile

@@ -0,0 +1,81 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+# Multi-stage build using openenv-base
+# This Dockerfile is flexible and works for both:
+# - In-repo environments (with local OpenEnv sources)
+# - Standalone environments (with openenv from PyPI/Git)
+# The build script (openenv build) handles context detection and sets appropriate build args.
+
+ARG BASE_IMAGE=ghcr.io/meta-pytorch/openenv-base:latest
+FROM ${BASE_IMAGE} AS builder
+
+WORKDIR /app
+
+# Ensure git is available (required for installing dependencies from VCS)
+RUN apt-get update && \
+    apt-get install -y --no-install-recommends git && \
+    rm -rf /var/lib/apt/lists/*
+
+# Build argument to control whether we're building standalone or in-repo
+ARG BUILD_MODE=in-repo
+ARG ENV_NAME=terminus_env
+
+# Copy environment code (always at root of build context)
+COPY . /app/env
+
+# For in-repo builds, openenv is already vendored in the build context
+# For standalone builds, openenv will be installed via pyproject.toml
+WORKDIR /app/env
+
+# Ensure uv is available (for local builds where base image lacks it)
+RUN if ! command -v uv >/dev/null 2>&1; then \
+        curl -LsSf https://astral.sh/uv/install.sh | sh && \
+        mv /root/.local/bin/uv /usr/local/bin/uv && \
+        mv /root/.local/bin/uvx /usr/local/bin/uvx; \
+    fi
+    
+# Install dependencies using uv sync
+# If uv.lock exists, use it; otherwise resolve on the fly
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --frozen --no-install-project --no-editable; \
+    else \
+        uv sync --no-install-project --no-editable; \
+    fi
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    if [ -f uv.lock ]; then \
+        uv sync --frozen --no-editable; \
+    else \
+        uv sync --no-editable; \
+    fi
+
+# Final runtime stage
+FROM ${BASE_IMAGE}
+
+WORKDIR /app
+
+# Copy the virtual environment from builder
+COPY --from=builder /app/env/.venv /app/.venv
+
+# Copy the environment code
+COPY --from=builder /app/env /app/env
+
+# Set PATH to use the virtual environment
+ENV PATH="/app/.venv/bin:$PATH"
+
+# Set PYTHONPATH so imports work correctly
+ENV PYTHONPATH="/app/env:$PYTHONPATH"
+ENV ENABLE_WEB_INTERFACE=true
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
+    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health', timeout=2).read()" || exit 1
+
+# Run the FastAPI server
+# The module path is constructed to work with the /app/env structure
+CMD ["sh", "-c", "cd /app/env && uvicorn server.app:app --host 0.0.0.0 --port 8000"]

README.md

@@ -1,10 +1,97 @@
 ---
-title: Terminus Env
-emoji: 🌖
+title: Terminus Environment Server
+emoji: 🛠️
 colorFrom: green
-colorTo: indigo
+colorTo: blue
 sdk: docker
 pinned: false
+app_port: 8000
+base_path: /web
+tags:
+  - openenv
+  - terminus
+  - harbor
+  - coding
+short_description: Single-tool Terminus-2 coding environment via Harbor
 ---
 
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Terminus Environment
+
+`terminus_env` exposes Harbor's Terminus-2 software-engineering agent through
+one MCP tool. Each call materializes a temporary Harbor task from the supplied
+instruction, setup commands, and verification commands, then runs:
+
+```bash
+harbor run --agent terminus-2 --model <model> --path <tasks> --task-name <task>
+```
+
+The tool returns a JSON rollout result with the generated command, process exit
+code, output tails, captured task files, reward if one is printed, and any
+error.
+
+## Tool
+
+- `run_rollout(...)`: create and run one Harbor Terminus-2 task.
+
+Key arguments:
+
+- `instruction`: coding task prompt.
+- `model`: LiteLLM-style model name passed to Harbor.
+- `setup`: shell commands baked into the generated task Dockerfile.
+- `verify`: shell commands written to `tests/test.sh`.
+- `api_base`: optional model API base URL.
+- `api_key`: optional provider API key exported as common provider env vars.
+- `env`: optional Harbor execution provider, for example `daytona`.
+- `max_turns`, `temperature`, `collect_rollout_details`: Terminus-2 settings.
+
+## Quick Start
+
+```python
+from terminus_env import TerminusEnv
+
+with TerminusEnv(base_url="http://localhost:8000").sync() as env:
+    env.reset()
+    result = env.call_tool(
+        "run_rollout",
+        instruction="Fix the failing Flask endpoint and make tests pass.",
+        model="openai/gpt-4.1",
+        setup=["pip install -q flask pytest"],
+        verify=["pytest -q"],
+        max_turns=100,
+    )
+    print(result)
+```
+
+## Local Server
+
+```bash
+cd envs/terminus_env
+uv run --project . server
+```
+
+The API and custom web UI are served on port 8000. The UI is mounted at `/web`.
+
+## Docker
+
+```bash
+cd envs/terminus_env
+openenv build -t terminus-env
+docker run -p 8000:8000 terminus-env
+```
+
+## Configuration
+
+- `MAX_CONCURRENT_ENVS`: maximum concurrent WebSocket sessions. Defaults to `4`.
+- Provider-specific model keys can be passed per call with `api_key` or mounted
+  into the server environment.
+- Harbor task execution may require an execution provider and provider secrets
+  depending on where the server is hosted. Pass the provider with `env`, for
+  example `env="daytona"`, and configure the corresponding Harbor credentials
+  in the host environment.
+
+## Notes
+
+This environment intentionally follows a single-tool pattern: OpenEnv owns the
+environment transport and Harbor owns the Terminus-2 rollout loop. It does not
+introduce a shared OpenEnv sandbox abstraction; sandbox provider unification can
+come later behind an RFC-backed API.

__init__.py

@@ -0,0 +1,21 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Terminus Environment for OpenEnv."""
+
+from openenv.core.env_server.mcp_types import CallToolAction, ListToolsAction
+
+from .client import TerminusEnv
+from .models import CommandResult, TerminusRolloutResult, TerminusState
+
+__all__ = [
+    "TerminusEnv",
+    "TerminusState",
+    "TerminusRolloutResult",
+    "CommandResult",
+    "CallToolAction",
+    "ListToolsAction",
+]

client.py

@@ -0,0 +1,15 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Client for the Terminus environment."""
+
+from openenv.core.mcp_client import MCPToolClient
+
+
+class TerminusEnv(MCPToolClient):
+    """MCP client for calling the Terminus single-rollout tool."""
+
+    pass

models.py

@@ -0,0 +1,51 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Models for the Terminus environment."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from openenv.core.env_server.types import State
+from pydantic import BaseModel, Field
+
+
+class CommandResult(BaseModel):
+    """Outcome of one shell command used by setup or verification."""
+
+    command: str
+    exit_code: int | None = None
+    stdout: str = ""
+    stderr: str = ""
+
+
+class TerminusRolloutResult(BaseModel):
+    """JSON payload returned by the ``run_rollout`` MCP tool."""
+
+    task_id: str = ""
+    reward: float | None = None
+    done: bool = True
+    exit_code: int | None = None
+    command: list[str] = Field(default_factory=list)
+    task_dir: str = ""
+    setup: list[str] = Field(default_factory=list)
+    verify: list[str] = Field(default_factory=list)
+    stdout_tail: str = ""
+    stderr_tail: str = ""
+    files: dict[str, str] = Field(default_factory=dict)
+    error: str | None = None
+    metadata: dict[str, Any] = Field(default_factory=dict)
+
+
+class TerminusState(State):
+    """Per-session state for Terminus rollout calls."""
+
+    rollouts_completed: int = 0
+    last_task_id: str | None = None
+    last_reward: float | None = None
+    last_exit_code: int | None = None
+    last_error: str | None = None

openenv.yaml

@@ -0,0 +1,7 @@
+spec_version: 1
+name: terminus_env
+type: space
+runtime: fastapi
+app: server.app:app
+port: 8000
+

pyproject.toml

@@ -0,0 +1,42 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+[build-system]
+requires = ["setuptools>=45", "wheel"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "openenv-terminus-env"
+version = "0.1.0"
+description = "Single-tool Terminus-2 coding environment for OpenEnv via Harbor"
+requires-python = ">=3.12"
+dependencies = [
+    "openenv-core[core]>=0.2.2",
+    "fastapi>=0.115.0",
+    "fastmcp>=3.0.0",
+    "gradio>=4.0.0",
+    "harbor>=0.3.0",
+    "pydantic>=2.0.0",
+    "requests>=2.31.0",
+    "uvicorn>=0.24.0",
+]
+
+[project.optional-dependencies]
+dev = [
+    "pytest>=8.0.0",
+    "pytest-cov>=4.0.0",
+]
+
+[project.scripts]
+server = "terminus_env.server.app:main"
+
+[tool.setuptools]
+include-package-data = true
+packages = ["terminus_env", "terminus_env.server"]
+package-dir = { "terminus_env" = ".", "terminus_env.server" = "server" }
+
+[tool.setuptools.package-data]
+terminus_env = ["**/*.txt", "**/*.yaml"]

server/__init__.py

@@ -0,0 +1,11 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Terminus Env environment server components."""
+
+from .terminus_env_environment import TerminusEnvironment
+
+__all__ = ["TerminusEnvironment"]

server/app.py

@@ -0,0 +1,71 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""FastAPI app for the Terminus environment."""
+
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+from typing import Any, Dict
+
+from openenv.core.env_server.http_server import create_app
+from openenv.core.env_server.mcp_types import CallToolAction, CallToolObservation
+from pydantic import field_validator
+
+try:
+    from .terminus_env_environment import TerminusEnvironment
+except ImportError:  # pragma: no cover
+    from server.terminus_env_environment import TerminusEnvironment  # type: ignore
+
+
+def _load_env_file() -> None:
+    candidate = Path(__file__).resolve().parents[1] / ".env"
+    if not candidate.exists():
+        return
+    for raw in candidate.read_text(encoding="utf-8").splitlines():
+        line = raw.strip()
+        if not line or line.startswith("#") or "=" not in line:
+            continue
+        key, _, value = line.partition("=")
+        key = key.strip()
+        value = value.strip().strip('"').strip("'")
+        if key and key not in os.environ:
+            os.environ[key] = value
+
+
+class TerminusCallToolAction(CallToolAction):
+    """CallToolAction that accepts JSON strings for web UI arguments."""
+
+    @field_validator("arguments", mode="before")
+    @classmethod
+    def parse_arguments(cls, value: Any) -> Dict[str, Any]:
+        if isinstance(value, str):
+            return json.loads(value)
+        return value
+
+
+_load_env_file()
+os.environ.setdefault("ENABLE_WEB_INTERFACE", "true")
+
+app = create_app(
+    TerminusEnvironment,
+    TerminusCallToolAction,
+    CallToolObservation,
+    env_name="terminus_env",
+    max_concurrent_envs=int(os.getenv("MAX_CONCURRENT_ENVS", "4")),
+)
+
+
+def main(host: str = "0.0.0.0", port: int = 8000) -> None:
+    import uvicorn
+
+    uvicorn.run(app, host=host, port=port)
+
+
+if __name__ == "__main__":
+    main()

server/requirements.txt

@@ -0,0 +1,10 @@
+openenv-core[core]>=0.2.2
+fastapi>=0.115.0
+fastmcp>=3.0.0
+gradio>=4.0.0
+harbor>=0.3.0
+pydantic>=2.0.0
+requests>=2.31.0
+uvicorn>=0.24.0
+
+

server/terminus_env_environment.py

@@ -0,0 +1,379 @@
+# Copyright (c) Meta Platforms, Inc. and affiliates.
+# All rights reserved.
+#
+# This source code is licensed under the BSD-style license found in the
+# LICENSE file in the root directory of this source tree.
+
+"""Terminus environment implementation.
+
+The environment exposes one MCP tool, ``run_rollout``. The tool materializes a
+temporary Harbor task, invokes ``harbor run --agent terminus-2`` on that task,
+and returns a compact JSON rollout result.
+"""
+
+from __future__ import annotations
+
+import os
+import re
+import shutil
+import subprocess
+import tempfile
+from pathlib import Path
+from typing import Any, Optional
+from uuid import uuid4
+
+from fastmcp import FastMCP
+from openenv.core.env_server.mcp_environment import MCPEnvironment
+from openenv.core.env_server.types import Action, Observation
+
+try:
+    from ..models import TerminusRolloutResult, TerminusState
+except ImportError:  # pragma: no cover
+    from models import TerminusRolloutResult, TerminusState
+
+
+_TAIL_CHARS = 8000
+_DEFAULT_TIMEOUT_S = 900
+
+
+class TerminusEnvironment(MCPEnvironment):
+    """Single-tool MCP wrapper around Harbor's Terminus-2 agent."""
+
+    SUPPORTS_CONCURRENT_SESSIONS = True
+
+    def __init__(self):
+        self._state = TerminusState(episode_id=str(uuid4()), step_count=0)
+
+        mcp = FastMCP("terminus_env")
+
+        @mcp.tool
+        def run_rollout(
+            instruction: str,
+            model: str = "",
+            api_base: str = "",
+            api_key: str = "",
+            setup: Optional[list[str]] = None,
+            verify: Optional[list[str]] = None,
+            task_id: str = "",
+            env: str = "",
+            max_turns: int = 100,
+            temperature: float = 0.7,
+            collect_rollout_details: bool = False,
+            timeout_s: int = _DEFAULT_TIMEOUT_S,
+            keep_task_dir: bool = False,
+        ) -> str:
+            """Run one Terminus-2 rollout on a generated Harbor task.
+
+            Args:
+                instruction: Task instruction for Terminus-2.
+                model: LiteLLM-style model name passed to Harbor.
+                api_base: Optional custom model API base URL.
+                api_key: Optional API key exported for common providers.
+                setup: Shell commands baked into the task environment.
+                verify: Shell commands written into ``tests/test.sh``.
+                task_id: Optional identifier for bookkeeping.
+                env: Optional Harbor environment provider, e.g. ``daytona``.
+                max_turns: Terminus-2 turn limit.
+                temperature: Sampling temperature.
+                collect_rollout_details: Ask Terminus-2 to collect rollout details.
+                timeout_s: Wall-clock timeout for the Harbor process.
+                keep_task_dir: Keep generated task files for debugging.
+
+            Returns:
+                JSON-serialized TerminusRolloutResult.
+            """
+            return self._run_rollout(
+                instruction=instruction,
+                model=model,
+                api_base=api_base,
+                api_key=api_key,
+                setup=list(setup or []),
+                verify=list(verify or []),
+                task_id=task_id,
+                env=env,
+                max_turns=max_turns,
+                temperature=temperature,
+                collect_rollout_details=collect_rollout_details,
+                timeout_s=timeout_s,
+                keep_task_dir=keep_task_dir,
+            ).model_dump_json()
+
+        super().__init__(mcp)
+
+    def reset(
+        self,
+        seed: Optional[int] = None,
+        episode_id: Optional[str] = None,
+        **_: Any,
+    ) -> Observation:
+        self._state = TerminusState(
+            episode_id=episode_id or str(uuid4()),
+            step_count=0,
+        )
+        return Observation(
+            done=False,
+            reward=None,
+            metadata={
+                "status": "ready",
+                "message": "terminus_env ready. Call run_rollout(...).",
+            },
+        )
+
+    def _step_impl(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **_: Any,
+    ) -> Observation:
+        return Observation(
+            done=False,
+            reward=None,
+            metadata={
+                "error": (
+                    f"Unknown action type: {type(action).__name__}. "
+                    "Use ListToolsAction or CallToolAction for MCP interactions."
+                )
+            },
+        )
+
+    def step(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        self._state.step_count += 1
+        return super().step(action, timeout_s=timeout_s, **kwargs)
+
+    async def step_async(
+        self,
+        action: Action,
+        timeout_s: Optional[float] = None,
+        **kwargs: Any,
+    ) -> Observation:
+        self._state.step_count += 1
+        return await super().step_async(action, timeout_s=timeout_s, **kwargs)
+
+    @property
+    def state(self) -> TerminusState:
+        return self._state
+
+    def _run_rollout(
+        self,
+        *,
+        instruction: str,
+        model: str,
+        api_base: str,
+        api_key: str,
+        setup: list[str],
+        verify: list[str],
+        task_id: str,
+        env: str,
+        max_turns: int,
+        temperature: float,
+        collect_rollout_details: bool,
+        timeout_s: int,
+        keep_task_dir: bool,
+    ) -> TerminusRolloutResult:
+        resolved_task_id = task_id or f"terminus-{uuid4()}"
+        result = TerminusRolloutResult(
+            task_id=resolved_task_id,
+            setup=setup,
+            verify=verify,
+        )
+
+        if not instruction.strip():
+            result.error = "instruction is required"
+            self._record_result(result)
+            return result
+        if not model.strip():
+            result.error = "model is required"
+            self._record_result(result)
+            return result
+        if shutil.which("harbor") is None:
+            result.error = (
+                "harbor CLI is not installed. Install the terminus_env "
+                "dependencies or run `uv tool install harbor`."
+            )
+            self._record_result(result)
+            return result
+
+        task_root = Path(tempfile.mkdtemp(prefix=f"{resolved_task_id}-"))
+        result.task_dir = str(task_root)
+        try:
+            _write_harbor_task(task_root, instruction, setup, verify)
+            command = _build_harbor_command(
+                task_root=task_root,
+                task_id=task_root.name,
+                model=model,
+                api_base=api_base,
+                env=env,
+                max_turns=max_turns,
+                temperature=temperature,
+                collect_rollout_details=collect_rollout_details,
+            )
+            result.command = command
+
+            completed = _run_command(
+                command,
+                timeout_s=timeout_s,
+                api_key=api_key,
+            )
+            result.exit_code = completed.returncode
+            result.stdout_tail = (completed.stdout or "")[-_TAIL_CHARS:]
+            result.stderr_tail = (completed.stderr or "")[-_TAIL_CHARS:]
+            result.reward = _extract_reward(completed.stdout, completed.stderr)
+            result.error = None if completed.returncode == 0 else "harbor run failed"
+            result.files = _collect_small_files(task_root)
+        except subprocess.TimeoutExpired as exc:
+            result.exit_code = None
+            result.stdout_tail = (exc.stdout or "")[-_TAIL_CHARS:]
+            result.stderr_tail = (exc.stderr or "")[-_TAIL_CHARS:]
+            result.error = f"harbor run timed out after {timeout_s}s"
+        except Exception as exc:  # noqa: BLE001
+            result.error = f"{type(exc).__name__}: {exc}"
+        finally:
+            if not keep_task_dir:
+                shutil.rmtree(task_root, ignore_errors=True)
+
+        self._record_result(result)
+        return result
+
+    def _record_result(self, result: TerminusRolloutResult) -> None:
+        self._state.rollouts_completed += 1
+        self._state.last_task_id = result.task_id
+        self._state.last_reward = result.reward
+        self._state.last_exit_code = result.exit_code
+        self._state.last_error = result.error
+
+
+def _write_harbor_task(
+    task_root: Path,
+    instruction: str,
+    setup: list[str],
+    verify: list[str],
+) -> None:
+    (task_root / "environment").mkdir(parents=True, exist_ok=True)
+    (task_root / "tests").mkdir(parents=True, exist_ok=True)
+    (task_root / "solution").mkdir(parents=True, exist_ok=True)
+
+    (task_root / "instruction.md").write_text(instruction, encoding="utf-8")
+    (task_root / "task.toml").write_text(
+        'name = "openenv-terminus-task"\n',
+        encoding="utf-8",
+    )
+    (task_root / "environment" / "Dockerfile").write_text(
+        _build_dockerfile(setup),
+        encoding="utf-8",
+    )
+    test_script = _build_test_script(verify)
+    test_path = task_root / "tests" / "test.sh"
+    test_path.write_text(test_script, encoding="utf-8")
+    test_path.chmod(0o755)
+
+
+def _build_dockerfile(setup: list[str]) -> str:
+    setup_block = "\n".join(setup)
+    run_setup = ""
+    if setup_block.strip():
+        run_setup = "\nRUN <<'EOF'\nset -eux\n" + setup_block + "\nEOF\n"
+    return (
+        "FROM python:3.12-slim\n"
+        "WORKDIR /workspace\n"
+        "RUN apt-get update && apt-get install -y --no-install-recommends "
+        "bash ca-certificates curl git tmux && rm -rf /var/lib/apt/lists/*\n"
+        f"{run_setup}"
+    )
+
+
+def _build_test_script(verify: list[str]) -> str:
+    if not verify:
+        return "#!/usr/bin/env bash\nset -euo pipefail\nexit 0\n"
+    return "#!/usr/bin/env bash\nset -euo pipefail\n" + "\n".join(verify) + "\n"
+
+
+def _build_harbor_command(
+    *,
+    task_root: Path,
+    task_id: str,
+    model: str,
+    api_base: str,
+    env: str,
+    max_turns: int,
+    temperature: float,
+    collect_rollout_details: bool,
+) -> list[str]:
+    command = [
+        "harbor",
+        "run",
+        "--agent",
+        "terminus-2",
+        "--model",
+        model,
+        "--path",
+        str(task_root.parent),
+        "--task-name",
+        task_id,
+    ]
+    if env:
+        command.extend(["--env", env])
+    if api_base:
+        command.extend(["--agent-kwarg", f"api_base={api_base}"])
+    command.extend(
+        [
+            "--agent-kwarg",
+            f"max_turns={max_turns}",
+            "--agent-kwarg",
+            f"temperature={temperature}",
+        ]
+    )
+    if collect_rollout_details:
+        command.extend(["--agent-kwarg", "collect_rollout_details=true"])
+    return command
+
+
+def _run_command(
+    command: list[str],
+    *,
+    timeout_s: int,
+    api_key: str,
+) -> subprocess.CompletedProcess[str]:
+    env = os.environ.copy()
+    if api_key:
+        env.setdefault("OPENAI_API_KEY", api_key)
+        env.setdefault("ANTHROPIC_API_KEY", api_key)
+    return subprocess.run(
+        command,
+        check=False,
+        capture_output=True,
+        text=True,
+        timeout=timeout_s,
+        env=env,
+    )
+
+
+def _extract_reward(*texts: str) -> float | None:
+    joined = "\n".join(texts)
+    for pattern in [
+        r'"reward"\s*:\s*(-?\d+(?:\.\d+)?)',
+        r"reward['\"]?\s*[:=]\s*(-?\d+(?:\.\d+)?)",
+    ]:
+        match = re.search(pattern, joined, flags=re.IGNORECASE)
+        if match:
+            return float(match.group(1))
+    return None
+
+
+def _collect_small_files(root: Path) -> dict[str, str]:
+    files: dict[str, str] = {}
+    for path in root.rglob("*"):
+        if not path.is_file() or path.stat().st_size > 64_000:
+            continue
+        rel = str(path.relative_to(root))
+        if rel.startswith("."):
+            continue
+        try:
+            files[rel] = path.read_text(encoding="utf-8")[:_TAIL_CHARS]
+        except UnicodeDecodeError:
+            continue
+    return files