fix(tools): declare filesystem path arguments and canonicalize all paths

tests/test_tool_path_contracts.py

@@ -0,0 +1,124 @@
+"""Tool path declarations are independent of tool names and validated twice."""
+import os
+
+import pytest
+
+from tiny_harness._guard import FilesystemGuard
+from tiny_harness._hitl import ApprovalDecision, ApprovalGate
+from tiny_harness._tools import ToolDef, ToolExecutor, ToolRegistry
+
+
+def _registry(path_args=("filename",), props=None):
+    registry = ToolRegistry()
+    observed = []
+    schema = {"type": "object", "properties": props or {
+        "filename": {"type": "string"},
+        "path": {"type": "string"},
+        "source": {"type": "string"},
+        "destination": {"type": "string"},
+    }}
+    registry.register_from_def(
+        ToolDef("custom_import", "fixture", schema,
+                risk_level="read_only", path_args=path_args),
+        lambda args: observed.append(dict(args)) or str(args),
+    )
+    return registry, observed
+
+
+def test_path_args_require_explicit_string_fields():
+    for path_args in [("missing",), ("filename", "filename"), ("filename", "integer")]:
+        with pytest.raises(ValueError, match="path_args"):
+            ToolDef(
+                "bad", "fixture",
+                {"type": "object", "properties": {
+                    "filename": {"type": "string"},
+                    "integer": {"type": "integer"},
+                }},
+                path_args=path_args,
+            )
+
+
+@pytest.mark.asyncio
+async def test_custom_path_argument_uses_workspace_not_process_cwd(tmp_path, monkeypatch):
+    workspace = tmp_path / "workspace"
+    other = tmp_path / "other"
+    workspace.mkdir()
+    other.mkdir()
+    monkeypatch.chdir(other)
+    registry, observed = _registry()
+    result = await ToolExecutor(registry, FilesystemGuard(str(workspace))).execute(
+        "custom_import", {"filename": "data.txt"}, "call-1",
+    )
+    assert result.success
+    assert observed == [{"filename": str(workspace / "data.txt")}]
+    assert not (other / "data.txt").exists()
+
+
+@pytest.mark.asyncio
+async def test_unknown_tool_checks_every_conventional_field(tmp_path):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    external = tmp_path / "external.txt"
+    external.write_text("private")
+    registry, observed = _registry(path_args=())
+    executor = ToolExecutor(registry, FilesystemGuard(str(workspace)))
+    denied = await executor.execute(
+        "custom_import",
+        {"path": "safe.txt", "source": str(external), "destination": "ok.txt"},
+        "call-2",
+    )
+    assert not denied.success
+    assert "outside" in denied.content
+    assert observed == []
+
+
+@pytest.mark.asyncio
+async def test_all_declared_paths_block_symlink_escape(tmp_path):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    secret = tmp_path / "outside.txt"
+    secret.write_text("private")
+    (workspace / "alias").symlink_to(secret)
+    registry, observed = _registry(path_args=("filename",))
+    executor = ToolExecutor(registry, FilesystemGuard(str(workspace)))
+    denied = await executor.execute("custom_import", {"filename": "alias"}, "call-3")
+    assert not denied.success
+    assert observed == []
+
+
+@pytest.mark.asyncio
+async def test_approval_modified_arguments_are_validated_and_recannonicalized(tmp_path):
+    workspace = tmp_path / "workspace"
+    workspace.mkdir()
+    outside = tmp_path / "outside.txt"
+    outside.write_text("secret")
+    registry, observed = _registry(path_args=("filename",))
+
+    async def approve(request):
+        return ApprovalDecision(approved=True, modified_args={"filename": str(outside)})
+
+    registry.get("custom_import").definition.risk_level = "mutation"
+    executor = ToolExecutor(
+        registry, FilesystemGuard(str(workspace)),
+        approval_gate=ApprovalGate(handler=approve),
+    )
+    result = await executor.execute(
+        "custom_import", {"filename": "safe.txt"}, "call-4",
+    )
+    assert not result.success and "outside" in result.content
+    assert observed == []
+
+
+@pytest.mark.asyncio
+async def test_registry_helper_accepts_custom_path_args(tmp_path):
+    registry = ToolRegistry()
+    registry.register_tool(
+        "artifact_read", "reads a fixture",
+        {"type": "object", "properties": {"asset_path": {"type": "string"}}},
+        lambda args: args["asset_path"],
+        risk_level="read_only", path_args=("asset_path",),
+    )
+    result = await ToolExecutor(registry, FilesystemGuard(str(tmp_path))).execute(
+        "artifact_read", {"asset_path": "nested/file.txt"}, "call-5",
+    )
+    assert result.success and result.content == str(tmp_path / "nested/file.txt")

tiny_harness/_tools.py

@@ -21,10 +21,19 @@ class ToolDef:
     description: str
     parameters: dict
     risk_level: str = "mutation"
+    # Filesystem locations in the schema, independent of a tool's Python name.
+    # All fields listed here are canonicalized before the handler is invoked.
+    path_args: tuple[str, ...] = ()
 
     def __post_init__(self) -> None:
         if self.risk_level not in ("safe", "read_only", "mutation", "destructive", "dangerous"):
             raise ValueError(f"Unknown tool risk_level: {self.risk_level!r}")
+        if not isinstance(self.path_args, tuple) or len(set(self.path_args)) != len(self.path_args):
+            raise ValueError("path_args must be a tuple of distinct schema property names")
+        props = self.parameters.get("properties", {})
+        for field in self.path_args:
+            if not isinstance(field, str) or not field or props.get(field, {}).get("type") != "string":
+                raise ValueError(f"path_args entry {field!r} must be a declared string property")
 
 
 @dataclass
@@ -63,8 +72,11 @@ def register(self, tool: Tool) -> None:
     def register_from_def(self, def_: ToolDef, handler: Callable) -> None:
         self.register(Tool(definition=def_, handler=handler))
 
-    def register_tool(self, name: str, description: str, parameters: dict, handler: Callable, risk_level: str = "mutation") -> None:
-        self.register_from_def(ToolDef(name=name, description=description, parameters=parameters, risk_level=risk_level), handler)
+    def register_tool(self, name: str, description: str, parameters: dict, handler: Callable, risk_level: str = "mutation", path_args: tuple[str, ...] = ()) -> None:
+        self.register_from_def(ToolDef(
+            name=name, description=description, parameters=parameters,
+            risk_level=risk_level, path_args=path_args,
+        ), handler)
 
     def get(self, name: str) -> Tool | None:
         return self._tools.get(name)
@@ -76,13 +88,22 @@ def names(self) -> list[str]:
         return list(self._tools.keys())
 
 
-# Built-in filesystem tools have explicit path contracts. A plugin that uses
-# other argument names must implement its own filesystem confinement.
-_FILE_TOOLS = frozenset({
-    "read_file", "write_file", "list_directory", "find_files",
-    "delete_file", "create_directory", "move_file", "search_content",
-})
-
+# Compatibility for third-party tools registered before path_args existed.
+# New tools should declare ToolDef(path_args=("filename", ...)).
+_LEGACY_FILE_PATH_KEYS = {
+    "read_file": ("path",),
+    "write_file": ("path",),
+    "list_directory": ("path",),
+    "find_files": ("path",),
+    "delete_file": ("path",),
+    "create_directory": ("path",),
+    "move_file": ("source", "destination"),
+    "search_content": ("path",),
+}
+_DEFAULT_WORKSPACE_PATH_TOOLS = frozenset((
+    "list_directory", "find_files", "search_content", "run_command",
+))
+_CONVENTIONAL_PATH_KEYS = ("path", "source", "destination", "cwd")
 
 class ToolExecutor:
     def __init__(self, registry: ToolRegistry, guard: "FilesystemGuard", timeout_ms: int = 30_000, max_output_chars: int = 50_000, approval_gate: "ApprovalGate | None" = None, event_bus: "EventBus | None" = None, allow_unsafe_shell: bool = False):
@@ -97,33 +118,38 @@ def __init__(self, registry: ToolRegistry, guard: "FilesystemGuard", timeout_ms:
     def get_definitions(self) -> list[dict]:
         return self._registry.get_definitions()
 
-    def _guard_args(self, name: str, args: dict, risk_level: str) -> dict:
+    def _guard_args(self, tool: Tool, args: dict) -> dict:
         if not self._guard:
-            return args
+            return dict(args)
+        name = tool.definition.name
         checked = dict(args)
+
+        # Legacy conventional fields are checked for compatibility. A ToolDef
+        # may additionally declare arbitrary path fields such as "filename".
+        # Resolve EVERY field, not just the first nonempty one.
+        fields = set(tool.definition.path_args)
+        fields.update(_LEGACY_FILE_PATH_KEYS.get(name, ()))
+        fields.update(k for k in _CONVENTIONAL_PATH_KEYS if k in checked)
         if name == "run_command":
-            # The executor owns the workspace contract. Never let subprocess
-            # inherit an unrelated process working directory.
-            checked["cwd"] = self._guard.guard(checked.get("cwd") or ".", risk_level)
-        elif name in _FILE_TOOLS:
-            keys = ("source", "destination") if name == "move_file" else ("path",)
-            for key in keys:
-                if key in checked:
-                    checked[key] = self._guard.guard(checked[key], risk_level)
-            if name in ("list_directory", "find_files", "search_content") and "path" not in checked:
-                checked["path"] = self._guard.guard(".", risk_level)
-            if name == "find_files":
-                # Glob patterns require separate validation from plain paths.
-                import os
-                pattern = checked.get("pattern", "")
-                if os.path.isabs(pattern) or ".." in pattern.replace("\\", "/").split("/"):
-                    raise ValueError("Glob pattern must remain within the workspace")
-        else:
-            # Legacy protection for third-party tools with a conventional
-            # path argument; this does NOT sandbox arbitrary shell commands.
-            path = checked.get("path") or checked.get("source") or checked.get("destination") or checked.get("cwd")
-            if path:
-                self._guard.guard(path, risk_level)
+            fields.add("cwd")
+
+        for field in fields:
+            if field in checked:
+                value = checked[field]
+                if not isinstance(value, str):
+                    raise ValueError(f"Filesystem path {field!r} must be a string")
+                checked[field] = self._guard.guard(value, tool.definition.risk_level)
+            elif field == "cwd" and name == "run_command":
+                checked["cwd"] = self._guard.guard(".", tool.definition.risk_level)
+            elif field == "path" and name in _DEFAULT_WORKSPACE_PATH_TOOLS:
+                checked["path"] = self._guard.guard(".", tool.definition.risk_level)
+
+        if name == "find_files":
+            # Glob patterns need validation beyond ordinary path arguments.
+            import os
+            pattern = checked.get("pattern", "")
+            if os.path.isabs(pattern) or ".." in pattern.replace("\\", "/").split("/"):
+                raise ValueError("Glob pattern must remain within the workspace")
         return checked
 
     async def execute(self, name: str, args: dict, call_id: str) -> ToolResult:
@@ -147,7 +173,7 @@ async def execute(self, name: str, args: dict, call_id: str) -> ToolResult:
             return ToolResult.error(call_id, f"Invalid arguments for '{name}':\n" + "\n".join(f"  - {e}" for e in errors))
 
         try:
-            self._guard_args(name, args, tool.definition.risk_level)
+            self._guard_args(tool, args)
         except Exception as e:
             return ToolResult.error(call_id, str(e))
 
@@ -168,7 +194,7 @@ async def execute(self, name: str, args: dict, call_id: str) -> ToolResult:
         if errors:
             return ToolResult.error(call_id, f"Invalid approved arguments for '{name}':\n" + "\n".join(f"  - {e}" for e in errors))
         try:
-            args = self._guard_args(name, args, tool.definition.risk_level)
+            args = self._guard_args(tool, args)
         except Exception as e:
             return ToolResult.error(call_id, str(e))
 

tiny_harness/skills/files.py

@@ -77,31 +77,31 @@
 
 def register(agent) -> None:
     agent.tools.register_from_def(
-        ToolDef(name="read_file", description="Read file contents from the filesystem. Returns text content. For large files, use offset and limit to read specific sections. Binary files return a size/type description instead of raw content.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to read."}, "offset": {"type": "integer", "description": "Line number to start reading from (1-indexed). Default: 1."}, "limit": {"type": "integer", "description": "Maximum number of lines to read. Omit to read the entire file."}}, "required": ["path"]}, risk_level="read_only"),
+        ToolDef(name="read_file", description="Read file contents from the filesystem. Returns text content. For large files, use offset and limit to read specific sections. Binary files return a size/type description instead of raw content.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to read."}, "offset": {"type": "integer", "description": "Line number to start reading from (1-indexed). Default: 1."}, "limit": {"type": "integer", "description": "Maximum number of lines to read. Omit to read the entire file."}}, "required": ["path"]}, risk_level="read_only", path_args=("path",)),
         partial(read_file, structured=True),
     )
     agent.tools.register_from_def(
-        ToolDef(name="write_file", description="Create a new file or overwrite an existing one. Auto-creates parent directories. WARNING: Overwrites existing files without confirmation. Verify with read_file if unsure about current contents.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to create or overwrite."}, "content": {"type": "string", "description": "The text content to write to the file."}}, "required": ["path", "content"]}, risk_level="mutation"),
+        ToolDef(name="write_file", description="Create a new file or overwrite an existing one. Auto-creates parent directories. WARNING: Overwrites existing files without confirmation. Verify with read_file if unsure about current contents.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to create or overwrite."}, "content": {"type": "string", "description": "The text content to write to the file."}}, "required": ["path", "content"]}, risk_level="mutation", path_args=("path",)),
         partial(write_file, structured=True),
     )
     agent.tools.register_from_def(
-        ToolDef(name="list_directory", description="List files and subdirectories in a directory. Shows names, types (file/dir), and sizes. Use to understand project structure before making changes.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Directory to list. Default: current directory."}, "pattern": {"type": "string", "description": "Glob pattern to filter (e.g., '*.py')."}, "recursive": {"type": "boolean", "description": "List recursively if true. Default: false."}}}, risk_level="read_only"),
+        ToolDef(name="list_directory", description="List files and subdirectories in a directory. Shows names, types (file/dir), and sizes. Use to understand project structure before making changes.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Directory to list. Default: current directory."}, "pattern": {"type": "string", "description": "Glob pattern to filter (e.g., '*.py')."}, "recursive": {"type": "boolean", "description": "List recursively if true. Default: false."}}}, risk_level="read_only", path_args=("path",)),
         partial(list_directory, structured=True),
     )
     agent.tools.register_from_def(
-        ToolDef(name="find_files", description="Find files matching a glob pattern. Supports recursive wildcards like '**/*.py'. Use to discover file locations before reading them.", parameters={"type": "object", "properties": {"pattern": {"type": "string", "description": "Glob pattern (e.g., '**/*.py', 'src/**/*test*')."}, "path": {"type": "string", "description": "Directory to search in. Default: current directory."}, "max_results": {"type": "integer", "description": "Maximum number of results. Default: 200."}}, "required": ["pattern"]}, risk_level="read_only"),
+        ToolDef(name="find_files", description="Find files matching a glob pattern. Supports recursive wildcards like '**/*.py'. Use to discover file locations before reading them.", parameters={"type": "object", "properties": {"pattern": {"type": "string", "description": "Glob pattern (e.g., '**/*.py', 'src/**/*test*')."}, "path": {"type": "string", "description": "Directory to search in. Default: current directory."}, "max_results": {"type": "integer", "description": "Maximum number of results. Default: 200."}}, "required": ["pattern"]}, risk_level="read_only", path_args=("path",)),
         partial(find_files, structured=True),
     )
     agent.tools.register_from_def(
-        ToolDef(name="delete_file", description="Permanently delete a file. IRREVERSIBLE — cannot be undone. Prefer move_file to archive. Always confirm what you're deleting by reading or listing first.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to delete permanently."}}, "required": ["path"]}, risk_level="destructive"),
+        ToolDef(name="delete_file", description="Permanently delete a file. IRREVERSIBLE — cannot be undone. Prefer move_file to archive. Always confirm what you're deleting by reading or listing first.", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the file to delete permanently."}}, "required": ["path"]}, risk_level="destructive", path_args=("path",)),
         partial(delete_file, structured=True),
     )
     agent.tools.register_from_def(
-        ToolDef(name="create_directory", description="Create a directory and any needed parent directories. Safe to call on existing directories (no error).", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the directory to create."}}, "required": ["path"]}, risk_level="mutation"),
+        ToolDef(name="create_directory", description="Create a directory and any needed parent directories. Safe to call on existing directories (no error).", parameters={"type": "object", "properties": {"path": {"type": "string", "description": "Path to the directory to create."}}, "required": ["path"]}, risk_level="mutation", path_args=("path",)),
         partial(create_directory, structured=True),
     )
     agent.tools.register_from_def(
-        ToolDef(name="move_file", description="Move or rename a file or directory. Overwrites the destination if it already exists. Safer alternative to delete_file for removing files.", parameters={"type": "object", "properties": {"source": {"type": "string", "description": "Current path of the file or directory."}, "destination": {"type": "string", "description": "New path for the file or directory."}}, "required": ["source", "destination"]}, risk_level="mutation"),
+        ToolDef(name="move_file", description="Move or rename a file or directory. Overwrites the destination if it already exists. Safer alternative to delete_file for removing files.", parameters={"type": "object", "properties": {"source": {"type": "string", "description": "Current path of the file or directory."}, "destination": {"type": "string", "description": "New path for the file or directory."}}, "required": ["source", "destination"]}, risk_level="mutation", path_args=("source", "destination")),
         partial(move_file, structured=True),
     )
     agent._prompt.append(FILES_PROMPT_SECTION)

tiny_harness/skills/search.py

@@ -88,6 +88,7 @@ def register(agent) -> None:
                 "required": ["pattern"],
             },
             risk_level="read_only",
+            path_args=("path",),
         ),
         partial(search_content, structured=True),
     )