From bd77149876954c0aad8e85e84878ede2a36c2c77 Mon Sep 17 00:00:00 2001 From: Talos Maintainers <82345214+morningstarnasser@users.noreply.github.com> Date: Tue, 3 Nov 2026 12:06:02 +0100 Subject: [PATCH] feat: preview Hermes and OpenClaw imports --- CHANGELOG.md | 9 + CLAUDE.md | 8 +- README.md | 18 +- docs/import-agents.md | 40 +++ site/console.html | 4 +- site/docs/index.html | 2 +- site/index.html | 2 +- site/llms.txt | 2 +- site/redteam/index.html | 2 +- site/registry/index.html | 2 +- site/setup/index.html | 2 +- site/status.json | 2 +- site/vergleich/index.html | 2 +- talos/cli.py | 63 ++++- talos/import_agents.py | 535 +++++++++++++++++++++++++++++++++++ tests/test_cli_onboarding.py | 35 ++- tests/test_import_agents.py | 304 ++++++++++++++++++++ 17 files changed, 1015 insertions(+), 17 deletions(-) create mode 100644 docs/import-agents.md create mode 100644 talos/import_agents.py create mode 100644 tests/test_import_agents.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 9765334..bec7f8d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). The 0.20.0 beta line follows the frozen operator contract; earlier alpha releases remain historical snapshots. Beta does not mean error-free operation. +## [Unreleased] + +### Added + +- Add an explicit, read-only `talos import ... --dry-run` preview for portable + Hermes and OpenClaw profile text, memory and Agent Skills. The preview writes + nothing, never activates imported instructions, and excludes secrets, + permissions, executable configuration, schedules, sessions and databases. + ## [0.20.0-rc.2] — 2026-11-01 ### Security diff --git a/CLAUDE.md b/CLAUDE.md index 9108b2d..7345045 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -14,7 +14,7 @@ preferences live in `USER.md`. All three are operator-owned prompt state and rel |---|---| | Gate path | `policy.py`, **996 lines** — has to stay readable in one sitting | | Tools | **37**, every one gated | -| Suites | **3456** collected tests · **277** adversarial · 44 end-to-end | +| Suites | **3477** collected tests · **277** adversarial · 44 end-to-end | | Home | · docs at `/docs/` | | Repository | `talos-kernel/talos` is the public source tree | @@ -52,6 +52,10 @@ automatically; do not overwrite operator state with repository templates. Homebr installations update through `brew upgrade talos-kernel/tap/talos`; `talos update` refuses them. Source installations retain their own profile and update path. +`talos import hermes|openclaw --source --dry-run` can explicitly preview a +small portable allowlist from another agent home. It changes neither home, imports +no authority or secrets, and does not activate the persistent text or Agent Skills. + The source branch can contain fixes newer than the latest signed release. Check `CHANGELOG.md`, the release tag and `site/status.json` before claiming a fix is shipped; a source commit does not change an existing signed archive. @@ -313,7 +317,7 @@ repairs; corrections never replay an effect or grant permission. Keep those prop ```bash python3 -m venv .venv && . .venv/bin/activate && pip install --require-hashes -r requirements.lock -r requirements-dev.lock -python -m pytest tests/ -q # 3456 collected tests; platform-specific skips are explicit +python -m pytest tests/ -q # 3477 collected tests; platform-specific skips are explicit python redteam.py # 277 adversarial cases — mandatory for any kernel change python e2e.py # 44 cases against a real model (costs tokens and time) python -m talos --once # single cycle, for diagnosis diff --git a/README.md b/README.md index 1a5e095..8df16db 100644 --- a/README.md +++ b/README.md @@ -21,10 +21,10 @@

- - Tests + Tests Red team Gate path Tools @@ -258,6 +258,13 @@ section atomically with mode 0600. It prints the configuration path and the next Use `talos doctor` for missing prerequisites and `talos help ` for command help. A running service needs a restart after changing its file settings. +Coming from Hermes or OpenClaw? `talos import` produces a read-only, redacted plan for +portable profile files, memory documents and Agent Skills. It does not copy anything in +this release. Secrets, approvals, command allowlists, executable plugins and MCP commands, +schedules, sessions and databases are skipped without reading their values. Every proposed +skill remains inactive until a later, separately reviewed activation path passes Talos' +existing approval boundary. See [the migration preview contract](docs/import-agents.md). + Later, a single part can be redone without repeating the rest: ```bash @@ -271,6 +278,9 @@ python -m talos config set TALOS_MODEL_OVERRIDES '{"local-model": {"context_wind # a window or a price the catalogue does not know — # never a provider, url or key; shown as your word python -m talos models --refresh # force the same provider-catalogue sync now +python -m talos import list # supported agent-home previews +python -m talos import hermes --source "$HOME/.hermes" --dry-run +python -m talos import openclaw --source "$HOME/.openclaw" --dry-run python -m talos status # what it did last python -m talos chat # a session here; approvals at a terminal python -m talos ask "how many …?" # one turn from here, answer on stdout @@ -882,13 +892,14 @@ command centre. `/allowed` `/revoke` · `/log` `/undo` `/policy` `/autonomy` `/tools` `/whoami` `/version` · `/usage` `/model` `/eval` `/reasoning` `/debug` -**On the command line** — thirteen, each answering a question an operator actually asks: +**On the command line** — each answering a question an operator actually asks: | | | |---|---| | `chat` | a session here; approvals at a terminal | | `ask "…"` | one turn, for scripts and cron | | `setup` · `doctor` · `config` · `models` | set it up, and find out why something won't work | +| `import list` · `import --source … --dry-run` | preview portable agent data without writing or activating it | | `status` · `events` · `why ` | what it did, and why that was allowed or refused | | `report` · `review` | a record someone else can read · what this installation should change | | `update` · `version` | signature checked, tests first | @@ -925,6 +936,7 @@ sitting — a gate you cannot read is not a gate. | `vision.py` / `hearing.py` / `speech.py` / `frames.py` | reading a picture, hearing a recording, speaking, one still out of a video — ordinary READ/WRITE with a target | | `cli.py` / `doctor.py` / `configcli.py` / `schema.py` | the subcommands: diagnose, read and change settings — the schema decides what may be written, and the allowlist and the network exceptions never are | | `askcli.py` | `talos ask` — one turn from a script, as a channel with no special right | +| `import_agents.py` | bounded, redacted, zero-write migration previews; imported text remains inactive | | `models.py` / `model_catalog_sync.py` | provider-bound live model lists; complete snapshots may replace membership, incomplete ones only add, and failures keep the last good cache | | `updater.py` | update beside the old tree, both suites in the new one, switch only if green | | `eventlog.py` / `memory.py` / `usage.py` | durable log, conversation memory, metering | diff --git a/docs/import-agents.md b/docs/import-agents.md new file mode 100644 index 0000000..bf89cfa --- /dev/null +++ b/docs/import-agents.md @@ -0,0 +1,40 @@ +# Import preview from Hermes or OpenClaw + +`talos import` answers one question: which portable parts of an existing agent home could +be reviewed for Talos? The current command is deliberately a dry run. It creates no target +directory, backup, report or temporary file and never changes the source. + +```bash +talos import list +talos import hermes --source "$HOME/.hermes" --dry-run +talos import openclaw --source "$HOME/.openclaw" --dry-run +``` + +The source path is explicit. Talos accepts an absolute path after shell expansion and +refuses a source or target boundary containing a symlink. Input files must be bounded, +regular, singly linked UTF-8 files. The preview contains relative paths, byte counts and +SHA-256 digests, never file bodies. + +## What the preview considers + +- `SOUL.md`, `AGENTS.md`, `USER.md` and `MEMORY.md`, where that source stores them. +- Agent Skills with a valid `SKILL.md`, including one category-directory layer. +- Existing staging destinations as conflicts. The preview never overwrites them. + +Every candidate points below `data/import-staging//`. `staged` means “eligible for +operator review”, not active or trusted. Imported text can contain hostile instructions. +It grants no permission, and a skill must still pass Talos' managed-skill approval path +before it can become active. + +## What is always skipped + +- API keys, tokens, OAuth state, `.env`, `auth.json` and auth profiles. +- Command allowlists, approvals, standing permissions and autonomy settings. +- Plugins, hooks, MCP commands and other executable configuration. +- Schedules, cron jobs, plans, sessions, logs and databases. + +Credentials are configured again through Talos' own setup flow. Permissions are never +migrated. External connector grants must be paired separately for Talos and are never +copied from Hermes or OpenClaw. A future apply command needs its own backup, atomic +publication, exact-item confirmation, read-back and rollback contract; this preview does +not imply that contract already exists. diff --git a/site/console.html b/site/console.html index ab6726c..789b52d 100644 --- a/site/console.html +++ b/site/console.html @@ -378,7 +378,7 @@

Witness one deed from door to ledger.

This candidate, measured
- 3456automated tests + 3477automated tests 277/277adversarial 996lines · gate 37gated tools @@ -554,7 +554,7 @@

Numbers this candidate must ship.

0.20.0-rc.2
release candidate
public installer v0.20.0-rc.2
-
3456
automated tests, collected by pytest
+
3477
automated tests, collected by pytest
277
adversarial cases, all behaving
996
lines in policy.py, readable in one sitting
37
tools, every one through the gate
diff --git a/site/docs/index.html b/site/docs/index.html index e9d491c..18bd56e 100644 --- a/site/docs/index.html +++ b/site/docs/index.html @@ -244,7 +244,7 @@

Talos AI agent: the field manual

That one decision shapes the rest of this manual. It is why the command list is short, and why a few obvious conveniences are missing on purpose.

-
3456automated tests
+
3477automated tests
277/277adversarial
996gate lines
37gated tools
diff --git a/site/index.html b/site/index.html index b15d10d..30a2bd2 100644 --- a/site/index.html +++ b/site/index.html @@ -639,7 +639,7 @@

Measured, with the path stated

None of this asks for belief. The installer runs the unit and adversarial suites in front of you; the real-model cases are a separate opt-in E2E run.

Release-candidate source and signed release. Counts describe the source test suite, not speed or real-world success rates; signed release contents follow their tag.

-
0
automated tests collected from the source tree
+
0
automated tests collected from the source tree
0
adversarial cases trying to pass the kernel
0
opt-in end-to-end cases against a real model
0
lines in the file that decides every verdict
diff --git a/site/llms.txt b/site/llms.txt index 106b7ae..b3b49b7 100644 --- a/site/llms.txt +++ b/site/llms.txt @@ -6,7 +6,7 @@ - Permission kernel: 996 lines of Python, the only code that decides - 37 tools, all gated; unattended runs cannot escalate -- 3456 tests collected in the source tree; 277/277 red-team cases passed in source verification. Release contents follow their own signed tag. +- 3477 tests collected in the source tree; 277/277 red-team cases passed in source verification. Release contents follow their own signed tag. - MCP servers run only inside a UID-separated worker, from an operator-owned registry — no marketplace, no third-party code in the agent process - Channels: Telegram, mail (IMAP pull), CLI, WhatsApp via an operator-controlled broker — every channel fetches; no inbound port, no webhook - Updates: signed tarballs, verified before anything is unpacked; dependencies pinned and hashed, installed with --require-hashes diff --git a/site/redteam/index.html b/site/redteam/index.html index da14acd..2a2053a 100644 --- a/site/redteam/index.html +++ b/site/redteam/index.html @@ -224,7 +224,7 @@

277 ways to break an agent.

25attack surfaces
996kernel lines
37gated tools
-
3456automated tests
+
3477automated tests
0passes sold as safety
diff --git a/site/registry/index.html b/site/registry/index.html index 839e567..46aafe4 100644 --- a/site/registry/index.html +++ b/site/registry/index.html @@ -239,7 +239,7 @@

The MCP registry

guide with the full protocol is docs/claude-worker.md. A curated starting point ships as examples/mcp-servers.json.

-
3456automated tests
+
3477automated tests
277/277adversarial
996gate lines
37gated tools
diff --git a/site/setup/index.html b/site/setup/index.html index 0c6f14e..0ab5e9c 100644 --- a/site/setup/index.html +++ b/site/setup/index.html @@ -163,7 +163,7 @@

This address moved.

Return to the keep
-
3456automated tests
+
3477automated tests
277/277adversarial
996gate lines
37gated tools
diff --git a/site/status.json b/site/status.json index 3afb764..f0e29cd 100644 --- a/site/status.json +++ b/site/status.json @@ -3,7 +3,7 @@ "release_state": "released", "source_state": "release", "installer_version": "0.20.0-rc.2", - "tests": 3456, + "tests": 3477, "redteam_faelle": 277, "kernel_zeilen": 996, "werkzeuge": 37, diff --git a/site/vergleich/index.html b/site/vergleich/index.html index cc9c9d7..a5c7856 100644 --- a/site/vergleich/index.html +++ b/site/vergleich/index.html @@ -290,7 +290,7 @@

The others can do more.

37tools here, every one gated
100+skills there · counted honestly
-
3456automated tests
+
3477automated tests
277/277adversarial cases
996gate lines
0inbound ports
diff --git a/talos/cli.py b/talos/cli.py index 17a9fe9..85a8525 100644 --- a/talos/cli.py +++ b/talos/cli.py @@ -52,6 +52,7 @@ setup [terminal|identity|model|mail] guided setup; preserves other settings doctor [--online] what is missing — changes nothing config list|get|set|describe|validate find, explain and change settings + import list|hermes|openclaw preview portable data from another agent models [--refresh] which models a provider offers local-check [--provider ollama|lm-studio] probe local endpoint without operator data @@ -266,6 +267,65 @@ def cmd_chat(rest: list[str], out=None) -> int: return 0 +def cmd_import(rest: list[str], out=None) -> int: + """Preview portable data from another agent without changing either home. + + Import text is persistent, foreign instruction material. The first release + therefore stops at a redacted plan: activation needs a separate reviewed path, + especially for skills. Keeping this command before configuration loading also + lets a new Talos installation inspect its old home before setup. + """ + schreiben = (out or sys.stdout).write + usage = " usage: talos import --source --dry-run\n" + if rest in (["--help"], ["-h"]): + schreiben( + usage + + " talos import list\n" + + " Preview only. Secrets, permissions, executable hooks, schedules and sessions are skipped.\n" + ) + return 0 + if rest == ["list"]: + schreiben( + "\n hermes portable profile, memory and Agent Skills\n" + " openclaw portable workspace profile, memory and Agent Skills\n\n" + " Dry run only; imported text remains inactive and grants no permission.\n\n" + ) + return 0 + if not rest or rest[0] not in {"hermes", "openclaw"}: + schreiben(usage) + return 2 + source_type = rest[0] + source = "" + dry_run = False + index = 1 + while index < len(rest): + argument = rest[index] + if argument == "--source" and index + 1 < len(rest) and not source: + source = rest[index + 1] + index += 2 + continue + if argument == "--dry-run" and not dry_run: + dry_run = True + index += 1 + continue + schreiben(f" unknown or repeated import option: {argument}\n{usage}") + return 2 + if not source or not dry_run: + schreiben(usage + " This release requires an explicit source and supports preview only.\n") + return 2 + + from .import_agents import ImportPlanError, plan_import + from .paths import PROFILE_DIR + + try: + plan = plan_import(source, PROFILE_DIR, source_type=source_type) + except (ImportPlanError, OSError) as error: + schreiben(f"\n import preview refused: {error}\n\n") + return 1 + schreiben("\n" + plan.render() + "\n\n") + return 0 + + def _help(_rest: list[str]) -> int: from .terminalui import heading, help_text @@ -281,6 +341,7 @@ def _help(_rest: list[str]) -> int: "setup": _lazy("setup_wizard", "run_setup"), "doctor": _lazy("doctor", "run_doctor"), "config": _lazy("configcli", "run_config"), + "import": cmd_import, "models": _lazy("models", "run_models"), "local-check": _lazy("localcheck", "run_local_check"), "ask": cmd_ask, @@ -337,4 +398,4 @@ def dispatch(args: list[str]) -> int | None: COMMANDS = tuple(TABLE) -__all__ = ["COMMANDS", "HELP", "TABLE", "cmd_status", "cmd_verify", "cmd_version", "dispatch"] +__all__ = ["COMMANDS", "HELP", "TABLE", "cmd_import", "cmd_status", "cmd_verify", "cmd_version", "dispatch"] diff --git a/talos/import_agents.py b/talos/import_agents.py new file mode 100644 index 0000000..fb5721c --- /dev/null +++ b/talos/import_agents.py @@ -0,0 +1,535 @@ +"""Read-only migration preview for portable Hermes and OpenClaw user data. + +The importer deliberately stops at a plan. Source files are foreign, persistent +instructions; copying them into the active profile would bypass Talos' review and +``skill_manage`` approval boundaries. This module therefore reads a small allowlist, +labels every accepted item for inactive staging, and never creates or changes a file. + +Secrets, authority, executable configuration, schedules, sessions and databases are +listed as skipped categories without opening their source files. The plan contains +only sizes and hashes of portable files, never their contents. +""" +from __future__ import annotations + +import hashlib +import json +import os +import stat +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Iterable + +from . import skills +from .recall import looks_secret + +PLAN_VERSION = 1 +SOURCE_TYPES = ("hermes", "openclaw") +STAGING_ROOT = Path("data") / "import-staging" +MAX_PORTABLE_BYTES = 256 * 1024 +MAX_SKILL_ROOTS = 64 +MAX_SKILL_CANDIDATES = 500 +MAX_DIRECTORY_ENTRIES = 1_024 + +STATUS_STAGED = "staged" +STATUS_CONFLICT = "conflict" +STATUS_REJECTED = "rejected" +STATUS_SKIPPED = "skipped" + +_NOFOLLOW = getattr(os, "O_NOFOLLOW", 0) +_DIRECTORY = getattr(os, "O_DIRECTORY", 0) +_NONBLOCK = getattr(os, "O_NONBLOCK", 0) + + +class ImportPlanError(ValueError): + """The requested source/target boundary is not safe enough to inspect.""" + + +@dataclass(frozen=True) +class ImportItem: + """One reviewable mapping; content is intentionally absent.""" + + category: str + source: str + target: str + status: str + reason: str + bytes: int = 0 + sha256: str = "" + + def to_dict(self) -> dict[str, object]: + return asdict(self) + + +@dataclass(frozen=True) +class ImportPlan: + """A deterministic, serialisable dry-run result.""" + + version: int + source_type: str + items: tuple[ImportItem, ...] + + def to_dict(self) -> dict[str, object]: + return { + "version": self.version, + "source_type": self.source_type, + "items": [item.to_dict() for item in self.items], + } + + def render(self) -> str: + counts = {status: 0 for status in ( + STATUS_STAGED, STATUS_CONFLICT, STATUS_REJECTED, STATUS_SKIPPED + )} + for item in self.items: + counts[item.status] = counts.get(item.status, 0) + 1 + lines = [ + f"Talos import preview v{self.version} — {self.source_type}", + "Dry run only: no files were changed. Imported text is untrusted data and grants no permission.", + ] + for item in self.items: + suffix = f" ({_display(item.reason)})" if item.reason else "" + mapping = f" -> {_display(item.target)}" if item.target else "" + lines.append( + f"[{item.status}] {item.category}: {_display(item.source)}{mapping}{suffix}" + ) + lines.append( + "Summary: " + ", ".join( + f"{counts[status]} {status}" for status in ( + STATUS_STAGED, STATUS_CONFLICT, STATUS_REJECTED, STATUS_SKIPPED + ) + ) + ) + return "\n".join(lines) + + +@dataclass(frozen=True) +class _Candidate: + category: str + source: Path + target: Path + + +def detect_source(source: str | os.PathLike[str]) -> str: + """Detect a strong on-disk signature without opening configuration files.""" + root = _safe_root(source, "source", must_exist=True) + openclaw = any(_entry_kind(root / name) == "file" for name in ( + "openclaw.json", "clawdbot.json", "moltbot.json" + )) + hermes = _entry_kind(root / "config.yaml") == "file" or ( + _entry_kind(root / "memories") == "directory" + and _entry_kind(root / "skills") == "directory" + ) + if openclaw and hermes: + raise ImportPlanError("source matches both Hermes and OpenClaw; choose the source type explicitly") + if openclaw: + return "openclaw" + if hermes: + return "hermes" + raise ImportPlanError("source is not a recognised Hermes or OpenClaw home") + + +def plan_import( + source: str | os.PathLike[str], + target: str | os.PathLike[str], + *, + source_type: str = "auto", +) -> ImportPlan: + """Inspect allowlisted portable data and return a zero-write migration plan.""" + source_root = _safe_root(source, "source", must_exist=True) + target_root = _safe_root(target, "target", must_exist=False) + if _paths_overlap(source_root, target_root): + raise ImportPlanError("source and target directories must not overlap") + kind = detect_source(source_root) if source_type == "auto" else source_type + if kind not in SOURCE_TYPES: + raise ImportPlanError(f"source type must be one of: auto, {', '.join(SOURCE_TYPES)}") + + candidates, skill_roots = ( + _hermes_candidates(source_root, target_root) + if kind == "hermes" + else _openclaw_candidates(source_root, target_root) + ) + items = [_inspect_candidate(source_root, target_root, candidate) for candidate in candidates] + items.extend(_inspect_skills(source_root, target_root, kind, skill_roots)) + items.extend(_skipped_items(kind)) + ordered = tuple(sorted(items, key=lambda item: ( + item.status != STATUS_STAGED, item.category, item.source, item.target + ))) + return ImportPlan( + version=PLAN_VERSION, + source_type=kind, + items=ordered, + ) + + +def _hermes_candidates(source: Path, target: Path) -> tuple[list[_Candidate], tuple[Path, ...]]: + stage = target / STAGING_ROOT / "hermes" + candidates: list[_Candidate] = [] + _first(candidates, "identity", source, stage / "profile" / "SOUL.md", ("SOUL.md",)) + _first(candidates, "instructions", source, stage / "profile" / "AGENTS.md", ("AGENTS.md",)) + _first(candidates, "preferences", source, stage / "profile" / "USER.md", ( + "memories/USER.md", "USER.md" + )) + _first(candidates, "memory", source, stage / "memory" / "MEMORY.md", ( + "memories/MEMORY.md", "MEMORY.md" + )) + return candidates, (source / "skills",) + + +def _openclaw_candidates(source: Path, target: Path) -> tuple[list[_Candidate], tuple[Path, ...]]: + stage = target / STAGING_ROOT / "openclaw" + workspace = _first_directory(source, ("workspace", "workspace-main", "workspace.default")) + if workspace is None: + return [], () + candidates: list[_Candidate] = [] + _first(candidates, "identity", workspace, stage / "profile" / "SOUL.md", ("SOUL.md",)) + _first(candidates, "instructions", workspace, stage / "profile" / "AGENTS.md", ("AGENTS.md",)) + _first(candidates, "preferences", workspace, stage / "profile" / "USER.md", ("USER.md",)) + _first(candidates, "memory", workspace, stage / "memory" / "MEMORY.md", ("MEMORY.md",)) + return candidates, (workspace / "skills",) + + +def _first( + candidates: list[_Candidate], + category: str, + base: Path, + target: Path, + names: Iterable[str], +) -> None: + for name in names: + candidate = base / name + if _entry_kind(candidate) is not None: + candidates.append(_Candidate(category, candidate, target)) + return + + +def _first_directory(source: Path, names: Iterable[str]) -> Path | None: + for name in names: + candidate = source / name + if _entry_kind(candidate) == "directory": + return candidate + if _entry_kind(candidate) == "symlink": + raise ImportPlanError("OpenClaw workspace must be a real directory (no symlinks)") + return None + + +def _inspect_candidate(source_root: Path, target_root: Path, candidate: _Candidate) -> ImportItem: + source_label = _relative_label(source_root, candidate.source) + target_label = _relative_label(target_root, candidate.target) + try: + data = _read_portable(source_root, candidate.source) + text = data.decode("utf-8") + if not text.strip(): + raise ImportPlanError("empty portable file") + secret = looks_secret(text) + if secret: + raise ImportPlanError(f"secret-like content ({secret})") + except (ImportPlanError, UnicodeDecodeError, OSError) as error: + reason = "not valid UTF-8" if isinstance(error, UnicodeDecodeError) else str(error) + return ImportItem(candidate.category, source_label, target_label, + STATUS_REJECTED, reason) + digest = hashlib.sha256(data).hexdigest() + target_state = _target_state(target_root, candidate.target) + if target_state == "unsafe": + return ImportItem( + candidate.category, source_label, target_label, STATUS_REJECTED, + "staging target ancestry contains a symlink; target not inspected", + ) + conflict = target_state == "conflict" + return ImportItem( + candidate.category, + source_label, + target_label, + STATUS_CONFLICT if conflict else STATUS_STAGED, + "staging target already exists; no overwrite" if conflict else "review required before activation", + len(data), + digest, + ) + + +def _inspect_skills( + source_root: Path, + target_root: Path, + kind: str, + roots: Iterable[Path], +) -> list[ImportItem]: + items: list[ImportItem] = [] + seen: set[str] = set() + files, rejected = _skill_files(source_root, tuple(roots)) + items.extend(rejected) + for skill_file in files: + source_label = _relative_label(source_root, skill_file) + name = skill_file.parent.name + target = target_root / STAGING_ROOT / kind / "skills" / name / skills.SKILL_FILENAME + target_label = _relative_label(target_root, target) + if name in seen: + items.append(ImportItem("skill", source_label, target_label, + STATUS_REJECTED, "duplicate skill name")) + continue + seen.add(name) + item = _inspect_candidate( + source_root, target_root, _Candidate("skill", skill_file, target) + ) + if item.status in {STATUS_STAGED, STATUS_CONFLICT}: + try: + data = _read_portable(source_root, skill_file) + text = data.decode("utf-8") + head, _body = skills._split_frontmatter(text) + fields, _metadata = skills._parse_frontmatter(head) + skills._validated_identity(fields, name) + except (skills.SkillError, ImportPlanError, UnicodeDecodeError, OSError): + # Parser errors can contain attacker-controlled frontmatter. Keep the + # reason generic so malformed persistent text cannot leak through a plan. + item = ImportItem("skill", source_label, target_label, + STATUS_REJECTED, "malformed Agent Skill frontmatter") + items.append(item) + return items + + +def _skill_files( + source_root: Path, + roots: tuple[Path, ...], +) -> tuple[tuple[Path, ...], list[ImportItem]]: + """Find direct Agent Skills plus one category layer without opening any file.""" + files: list[Path] = [] + rejected: list[ImportItem] = [] + categories: list[Path] = [] + for root in roots: + kind = _entry_kind(root) + if kind is None: + continue + if kind != "directory": + rejected.append(ImportItem( + "skill", _relative_label(source_root, root), "", STATUS_REJECTED, + "skill root must be a real directory (no symlinks)", + )) + continue + children, overflow = _bounded_children(root) + if overflow: + rejected.append(ImportItem( + "skill", _relative_label(source_root, root), "", STATUS_REJECTED, + f"skill root exceeds {MAX_DIRECTORY_ENTRIES} visible entries", + )) + for child in children: + child_kind = _entry_kind(child) + if child_kind == "symlink": + rejected.append(ImportItem( + "skill", _relative_label(source_root, child), "", STATUS_REJECTED, + "skill directory is a symlink", + )) + continue + if child_kind != "directory": + continue + descriptor = child / skills.SKILL_FILENAME + if _entry_kind(descriptor) is not None: + files.append(descriptor) + elif len(categories) < MAX_SKILL_ROOTS: + categories.append(child) + + for category in categories: + children, overflow = _bounded_children(category) + if overflow: + rejected.append(ImportItem( + "skill", _relative_label(source_root, category), "", STATUS_REJECTED, + f"skill category exceeds {MAX_DIRECTORY_ENTRIES} visible entries", + )) + for child in children: + child_kind = _entry_kind(child) + if child_kind == "symlink": + rejected.append(ImportItem( + "skill", _relative_label(source_root, child), "", STATUS_REJECTED, + "skill directory is a symlink", + )) + continue + descriptor = child / skills.SKILL_FILENAME + if child_kind == "directory" and _entry_kind(descriptor) is not None: + files.append(descriptor) + if len(files) >= MAX_SKILL_CANDIDATES: + rejected.append(ImportItem( + "skill", _relative_label(source_root, category), "", STATUS_REJECTED, + f"skill candidate limit of {MAX_SKILL_CANDIDATES} reached", + )) + break + if len(files) >= MAX_SKILL_CANDIDATES: + break + ordered = tuple(sorted(set(files[:MAX_SKILL_CANDIDATES]), key=lambda path: path.as_posix())) + return ordered, rejected + + +def _bounded_children(root: Path) -> tuple[tuple[Path, ...], bool]: + """Enumerate a directory through a pinned no-follow fd with bounded memory.""" + try: + descriptor = _open_directory_nofollow(root) + except OSError as error: + raise ImportPlanError("skill directory could not be inspected safely") from error + names: list[str] = [] + try: + with os.scandir(descriptor) as entries: + for count, entry in enumerate(entries, start=1): + if count > MAX_DIRECTORY_ENTRIES: + return (), True + if entry.name and not entry.name.startswith("."): + names.append(entry.name) + except OSError as error: + raise ImportPlanError("skill directory could not be inspected safely") from error + finally: + os.close(descriptor) + return tuple(root / name for name in sorted(names)), False + + +def _skipped_items(kind: str) -> list[ImportItem]: + if kind == "hermes": + rows = ( + ("secrets", ".env, auth.json, mcp-tokens/", "credentials are re-entered through Talos setup"), + ("authority", "config.yaml approvals and command allowlists", "authority never migrates"), + ("executables", "plugins/, hooks/, MCP command configuration", "executable configuration stays inactive"), + ("schedules", "cron/, plans/", "scheduled effects require separate operator review"), + ("sessions", "sessions/, logs/, databases", "private history is archive-only"), + ) + else: + rows = ( + ("secrets", ".env, auth profiles, tokens inside openclaw.json", "credentials are re-entered through Talos setup"), + ("authority", "exec-approvals.json and approval policy", "authority never migrates"), + ("executables", "plugins/, hooks/, MCP command configuration", "executable configuration stays inactive"), + ("schedules", "cron/, HEARTBEAT.md", "scheduled effects require separate operator review"), + ("sessions", "sessions/, logs/, databases", "private history is archive-only"), + ) + return [ImportItem(category, source, "", STATUS_SKIPPED, reason) + for category, source, reason in rows] + + +def _safe_root(raw: str | os.PathLike[str], label: str, *, must_exist: bool) -> Path: + path = Path(raw).expanduser() + if not path.is_absolute(): + raise ImportPlanError(f"{label} path must be absolute") + path = Path(os.path.abspath(path)) + current = Path(path.anchor) + for part in path.parts[1:]: + current /= part + try: + info = current.lstat() + except FileNotFoundError: + if must_exist: + raise ImportPlanError(f"{label} path does not exist") from None + break + if stat.S_ISLNK(info.st_mode): + raise ImportPlanError(f"{label} path contains a symlink") + if must_exist: + try: + info = path.lstat() + except FileNotFoundError: + raise ImportPlanError(f"{label} path does not exist") from None + if not stat.S_ISDIR(info.st_mode): + raise ImportPlanError(f"{label} path must be a directory") + elif _entry_kind(path) not in {None, "directory"}: + raise ImportPlanError(f"{label} path must be a directory or absent") + return path + + +def _entry_kind(path: Path) -> str | None: + try: + mode = path.lstat().st_mode + except (FileNotFoundError, NotADirectoryError): + return None + if stat.S_ISLNK(mode): + return "symlink" + if stat.S_ISDIR(mode): + return "directory" + if stat.S_ISREG(mode): + return "file" + return "special" + + +def _relative_label(root: Path, path: Path) -> str: + try: + relative = path.relative_to(root) + except ValueError: + return "[outside source refused]" + return relative.as_posix() or "." + + +def _display(value: str) -> str: + """Quote attacker-controlled labels so terminals cannot interpret controls.""" + return json.dumps(value, ensure_ascii=True) + + +def _paths_overlap(first: Path, second: Path) -> bool: + return first == second or first in second.parents or second in first.parents + + +def _target_state(root: Path, path: Path) -> str: + """Probe target ancestry one component at a time, never through a symlink.""" + try: + relative = path.relative_to(root) + except ValueError: + return "unsafe" + current = root + for index, part in enumerate(relative.parts): + current /= part + kind = _entry_kind(current) + if kind is None: + return "absent" + if kind == "symlink": + return "unsafe" + if index < len(relative.parts) - 1 and kind != "directory": + return "conflict" + return "conflict" + + +def _read_portable(root: Path, path: Path) -> bytes: + """Open each component without following links; read a bounded regular file.""" + try: + relative = path.relative_to(root) + except ValueError: + raise ImportPlanError("portable path escapes the source root") from None + if not relative.parts or any(part in {"", ".", ".."} for part in relative.parts): + raise ImportPlanError("invalid portable path") + descriptors: list[int] = [] + try: + current = _open_directory_nofollow(root) + descriptors.append(current) + for part in relative.parts[:-1]: + current = os.open(part, os.O_RDONLY | _DIRECTORY | _NOFOLLOW, dir_fd=current) + descriptors.append(current) + fd = os.open(relative.parts[-1], os.O_RDONLY | _NOFOLLOW | _NONBLOCK, dir_fd=current) + descriptors.append(fd) + info = os.fstat(fd) + if not stat.S_ISREG(info.st_mode): + raise ImportPlanError("portable source is not a regular file") + if info.st_nlink != 1: + raise ImportPlanError("portable source must not be hard-linked") + if info.st_size > MAX_PORTABLE_BYTES: + raise ImportPlanError(f"portable source exceeds {MAX_PORTABLE_BYTES} bytes") + chunks: list[bytes] = [] + remaining = MAX_PORTABLE_BYTES + 1 + while remaining: + chunk = os.read(fd, min(64 * 1024, remaining)) + if not chunk: + break + chunks.append(chunk) + remaining -= len(chunk) + data = b"".join(chunks) + if len(data) > MAX_PORTABLE_BYTES: + raise ImportPlanError(f"portable source exceeds {MAX_PORTABLE_BYTES} bytes") + return data + except OSError as error: + raise ImportPlanError(f"portable source refused: {error.strerror or error.__class__.__name__}") from error + finally: + for fd in reversed(descriptors): + try: + os.close(fd) + except OSError: + pass + + +def _open_directory_nofollow(path: Path) -> int: + """Open an absolute directory one component at a time without following links.""" + flags = os.O_RDONLY | _DIRECTORY | _NOFOLLOW | _NONBLOCK + current = os.open(path.anchor, flags) + try: + for part in path.parts[1:]: + following = os.open(part, flags, dir_fd=current) + os.close(current) + current = following + return current + except BaseException: + os.close(current) + raise diff --git a/tests/test_cli_onboarding.py b/tests/test_cli_onboarding.py index b486d93..63f2737 100644 --- a/tests/test_cli_onboarding.py +++ b/tests/test_cli_onboarding.py @@ -125,12 +125,45 @@ def test_bad_arguments_never_dispatch_an_agent_run(args): def test_command_help_and_typo_recovery_do_not_load_configuration(monkeypatch, capsys): monkeypatch.setattr(config, 'load_config', lambda **kw: pytest.fail('loaded config')) - for command in ('config', 'setup', 'ask', 'chat', 'status', 'verify', 'version'): + for command in ('config', 'setup', 'import', 'ask', 'chat', 'status', 'verify', 'version'): assert cli.dispatch(['help', command]) == 0 assert cli.dispatch(['doctro']) == 2 assert 'Did you mean talos doctor?' in capsys.readouterr().out +def test_import_command_previews_without_writing_source_or_target(tmp_path, monkeypatch, capsys): + from talos import paths + + source = tmp_path / 'hermes-home' + source.mkdir() + (source / 'config.yaml').write_text('model: example\n', encoding='utf-8') + (source / 'SOUL.md').write_text('# Existing agent\n', encoding='utf-8') + target = tmp_path / 'talos-home' + before = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob('*'))) + monkeypatch.setattr(paths, 'PROFILE_DIR', target) + + assert cli.dispatch(['import', 'hermes', '--source', str(source), '--dry-run']) == 0 + + after = tuple(sorted(path.relative_to(tmp_path) for path in tmp_path.rglob('*'))) + text = capsys.readouterr().out + assert before == after and not target.exists() + assert 'Dry run only' in text and '[staged] identity' in text + assert str(source) not in text + + +@pytest.mark.parametrize('args', [ + ['import'], + ['import', 'hermes'], + ['import', 'hermes', '--source', '/tmp/example'], + ['import', 'hermes', '--dry-run'], + ['import', 'unknown', '--source', '/tmp/example', '--dry-run'], + ['import', 'hermes', '--source', '/tmp/example', '--source', '/tmp/other', '--dry-run'], +]) +def test_import_command_requires_explicit_source_and_dry_run(args, capsys): + assert cli.dispatch(args) == 2 + assert 'usage: talos import' in capsys.readouterr().out + + def test_config_search_and_describe_explain_without_disclosing_secrets(tmp_path): path = tmp_path / 'config.env' path.write_text('TELEGRAM_BOT_TOKEN=never-show-this\n') diff --git a/tests/test_import_agents.py b/tests/test_import_agents.py new file mode 100644 index 0000000..6ee2a23 --- /dev/null +++ b/tests/test_import_agents.py @@ -0,0 +1,304 @@ +from __future__ import annotations + +import os +from pathlib import Path + +import pytest + +from talos import import_agents + + +def _write(path: Path, content: str | bytes) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + if isinstance(content, bytes): + path.write_bytes(content) + else: + path.write_text(content, encoding="utf-8") + + +def _hermes(root: Path) -> None: + _write(root / "config.yaml", "provider: local\n") + + +def _openclaw(root: Path) -> None: + _write(root / "openclaw.json", "not parsed by the importer") + + +def _skill(root: Path, name: str, *, description: str = "Portable guidance") -> Path: + path = root / name / "SKILL.md" + _write( + path, + f"---\nname: {name}\ndescription: {description}\n---\n\n# Instructions\nStay careful.\n", + ) + return path + + +def _snapshot(root: Path) -> tuple[tuple[str, str, int, bytes], ...]: + rows: list[tuple[str, str, int, bytes]] = [] + for path in sorted(root.rglob("*"), key=lambda item: item.as_posix()): + relative = path.relative_to(root).as_posix() + info = path.lstat() + if path.is_symlink(): + rows.append((relative, "symlink", info.st_mode, os.readlink(path).encode())) + elif path.is_file(): + rows.append((relative, "file", info.st_mode, path.read_bytes())) + else: + rows.append((relative, "directory", info.st_mode, b"")) + return tuple(rows) + + +def _items(plan: import_agents.ImportPlan, category: str) -> list[import_agents.ImportItem]: + return [item for item in plan.items if item.category == category] + + +def test_hermes_plan_is_deterministic_versioned_and_writes_nothing(tmp_path: Path) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + _write(source / "SOUL.md", "# Identity\nHelpful and careful.\n") + _write(source / "AGENTS.md", "# Local instructions\nAsk before effects.\n") + _write(source / "memories" / "USER.md", "# Preferences\nUse concise answers.\n") + _write(source / "memories" / "MEMORY.md", "# Memory\nA portable fact.\n") + _skill(source / "skills", "safe-helper") + before = _snapshot(source) + + first = import_agents.plan_import(source.resolve(), target.resolve()) + second = import_agents.plan_import(source.resolve(), target.resolve()) + + assert first == second + assert first.version == import_agents.PLAN_VERSION == 1 + assert first.source_type == "hermes" + assert first.to_dict() == second.to_dict() + assert first.render() == second.render() + assert first.render().startswith("Talos import preview v1 — hermes\nDry run only:") + assert str(tmp_path) not in first.render() + assert str(tmp_path) not in repr(first) + assert str(tmp_path) not in repr(first.to_dict()) + assert "Helpful and careful" not in first.render() + assert {item.status for item in first.items if item.category not in { + "secrets", "authority", "executables", "schedules", "sessions" + }} == {import_agents.STATUS_STAGED} + assert {item.category for item in first.items if item.status == import_agents.STATUS_SKIPPED} == { + "secrets", "authority", "executables", "schedules", "sessions" + } + assert any(item.target == "data/import-staging/hermes/skills/safe-helper/SKILL.md" + for item in first.items) + assert not target.exists() + assert _snapshot(source) == before + + +def test_plan_never_returns_or_renders_secret_values(tmp_path: Path) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + secret = "sk-" + "AbCd1234EfGh5678IjKl9012" + _write(source / "memories" / "USER.md", f"api_key = {secret}\n") + + plan = import_agents.plan_import(source.resolve(), target.resolve()) + + item = _items(plan, "preferences")[0] + assert item.status == import_agents.STATUS_REJECTED + assert "secret-like content" in item.reason + for output in (repr(plan), plan.render(), repr(plan.to_dict())): + assert secret not in output + + +def test_symlinked_portable_file_is_rejected_without_reading_target(tmp_path: Path) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + outside = tmp_path / "outside.txt" + _hermes(source) + secret = "ghp_" + "AbCdEf1234567890AbCdEf1234567890" + _write(outside, secret) + (source / "SOUL.md").symlink_to(outside) + + plan = import_agents.plan_import(source.resolve(), target.resolve()) + + item = _items(plan, "identity")[0] + assert item.status == import_agents.STATUS_REJECTED + assert "refused" in item.reason + assert secret not in repr(plan.to_dict()) + + +def test_source_root_and_openclaw_workspace_symlinks_are_rejected(tmp_path: Path) -> None: + real = tmp_path / "real" + alias = tmp_path / "alias" + target = tmp_path / "talos" + _hermes(real) + alias.symlink_to(real, target_is_directory=True) + with pytest.raises(import_agents.ImportPlanError, match="contains a symlink"): + import_agents.plan_import(alias.absolute(), target.resolve()) + + openclaw = tmp_path / "openclaw" + workspace = tmp_path / "external-workspace" + _openclaw(openclaw) + workspace.mkdir() + (openclaw / "workspace").symlink_to(workspace, target_is_directory=True) + with pytest.raises(import_agents.ImportPlanError, match="workspace.*no symlinks"): + import_agents.plan_import(openclaw.resolve(), target.resolve()) + + +def test_target_conflict_and_symlink_ancestry_are_fail_closed(tmp_path: Path) -> None: + source = tmp_path / "hermes" + _hermes(source) + _write(source / "SOUL.md", "# Identity\nPortable.\n") + + conflict_target = tmp_path / "conflict-target" + conflict = conflict_target / "data/import-staging/hermes/profile/SOUL.md" + _write(conflict, "existing") + plan = import_agents.plan_import(source.resolve(), conflict_target.resolve()) + item = _items(plan, "identity")[0] + assert item.status == import_agents.STATUS_CONFLICT + assert "no overwrite" in item.reason + + unsafe_target = tmp_path / "unsafe-target" + outside = tmp_path / "outside-target" + unsafe_target.mkdir() + outside.mkdir() + (unsafe_target / "data").symlink_to(outside, target_is_directory=True) + unsafe = import_agents.plan_import(source.resolve(), unsafe_target.resolve()) + item = _items(unsafe, "identity")[0] + assert item.status == import_agents.STATUS_REJECTED + assert "target ancestry contains a symlink" in item.reason + assert list(outside.iterdir()) == [] + + +def test_malformed_and_oversize_portable_inputs_are_rejected(tmp_path: Path) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + _write(source / "AGENTS.md", b"\xff\xfe\xfd") + _write(source / "memories" / "MEMORY.md", b"x" * (import_agents.MAX_PORTABLE_BYTES + 1)) + + plan = import_agents.plan_import(source.resolve(), target.resolve()) + + instructions = _items(plan, "instructions")[0] + memory = _items(plan, "memory")[0] + assert instructions.status == import_agents.STATUS_REJECTED + assert instructions.reason == "not valid UTF-8" + assert memory.status == import_agents.STATUS_REJECTED + assert "exceeds" in memory.reason + assert not target.exists() + + +def test_agent_skills_support_one_category_layer_and_reject_malformed(tmp_path: Path) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + _skill(source / "skills" / "writing", "proofreader") + malformed = source / "skills" / "bad-skill" / "SKILL.md" + _write(malformed, "---\nname: do-not-echo-this\n---\nbody\n") + + plan = import_agents.plan_import(source.resolve(), target.resolve()) + + skills = _items(plan, "skill") + assert any(item.status == import_agents.STATUS_STAGED + and item.source == "skills/writing/proofreader/SKILL.md" for item in skills) + bad = next(item for item in skills if item.source == "skills/bad-skill/SKILL.md") + assert bad.status == import_agents.STATUS_REJECTED + assert bad.reason == "malformed Agent Skill frontmatter" + assert "do-not-echo-this" not in plan.render() + + +def test_skill_directory_enumeration_stops_at_the_hard_limit(tmp_path: Path) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + skill_root = source / "skills" + for index in range(import_agents.MAX_DIRECTORY_ENTRIES + 1): + (skill_root / f"candidate-{index:04d}").mkdir(parents=True) + + plan = import_agents.plan_import(source.resolve(), target.resolve()) + + item = next(item for item in plan.items if "exceeds" in item.reason) + assert item.status == import_agents.STATUS_REJECTED + assert not any(item.status == import_agents.STATUS_STAGED for item in _items(plan, "skill")) + + +def test_skill_directory_inspection_errors_fail_closed(tmp_path: Path, monkeypatch) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + (source / "skills").mkdir() + + def refuse_inspection(_descriptor): + raise OSError("synthetic inspection refusal") + + monkeypatch.setattr(import_agents.os, "scandir", refuse_inspection) + with pytest.raises(import_agents.ImportPlanError, match="could not be inspected safely"): + import_agents.plan_import(source.resolve(), target.resolve()) + + +def test_render_quotes_control_characters_from_foreign_filenames(tmp_path: Path) -> None: + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + hostile = "evil\n\x1b[31m[staged] forged" + _write(source / "skills" / hostile / "SKILL.md", "not valid frontmatter\n") + + plan = import_agents.plan_import(source.resolve(), target.resolve()) + rendered = plan.render() + + assert "\x1b" not in rendered + assert "evil\n" not in rendered + assert "evil\\n\\u001b[31m[staged] forged" in rendered + + +def test_openclaw_uses_workspace_allowlist_and_skips_sensitive_categories(tmp_path: Path) -> None: + source = tmp_path / "openclaw" + target = tmp_path / "talos" + _openclaw(source) + _write(source / "workspace" / "SOUL.md", "# Identity\nCareful.\n") + _write(source / "workspace" / "USER.md", "# User\nConcise.\n") + _write(source / "exec-approvals.json", "this sensitive file must never be parsed") + + plan = import_agents.plan_import(source.resolve(), target.resolve()) + + assert plan.source_type == "openclaw" + assert {item.source for item in plan.items if item.status == import_agents.STATUS_STAGED} == { + "workspace/SOUL.md", "workspace/USER.md" + } + assert {item.category for item in plan.items if item.status == import_agents.STATUS_SKIPPED} == { + "secrets", "authority", "executables", "schedules", "sessions" + } + + +def test_detection_uses_regular_markers_and_reports_ambiguous_source_cleanly(tmp_path: Path) -> None: + source = tmp_path / "source" + source.mkdir() + (source / "config.yaml").symlink_to(tmp_path / "missing") + with pytest.raises(import_agents.ImportPlanError, match="not a recognised"): + import_agents.detect_source(source.resolve()) + + (source / "config.yaml").unlink() + _write(source / "config.yaml", "ignored") + _write(source / "openclaw.json", "ignored") + with pytest.raises(import_agents.ImportPlanError, match="choose the source type explicitly") as error: + import_agents.detect_source(source.resolve()) + assert "--from" not in str(error.value) + + +def test_paths_must_be_explicit_and_regular_files_must_not_be_hardlinked(tmp_path: Path) -> None: + with pytest.raises(import_agents.ImportPlanError, match="absolute"): + import_agents.plan_import("relative", (tmp_path / "target").resolve(), source_type="hermes") + + source = tmp_path / "hermes" + target = tmp_path / "talos" + _hermes(source) + _write(source / "SOUL.md", "# Identity\nPortable.\n") + os.link(source / "SOUL.md", source / "copy.md") + plan = import_agents.plan_import(source.resolve(), target.resolve()) + item = _items(plan, "identity")[0] + assert item.status == import_agents.STATUS_REJECTED + assert "hard-linked" in item.reason + + +def test_source_and_target_must_not_overlap(tmp_path: Path) -> None: + source = tmp_path / "hermes" + _hermes(source) + with pytest.raises(import_agents.ImportPlanError, match="must not overlap"): + import_agents.plan_import(source.resolve(), source.resolve(), source_type="hermes") + with pytest.raises(import_agents.ImportPlanError, match="must not overlap"): + import_agents.plan_import( + source.resolve(), (source / "prospective-talos").resolve(), source_type="hermes" + )