diff --git a/src/cli_agent_orchestrator/schemas/agent_profile.schema.json b/src/cli_agent_orchestrator/schemas/agent_profile.schema.json index c41fe4275..9c2e76623 100644 --- a/src/cli_agent_orchestrator/schemas/agent_profile.schema.json +++ b/src/cli_agent_orchestrator/schemas/agent_profile.schema.json @@ -108,6 +108,11 @@ }, "useLegacyMcpJson": { "type": "boolean" + }, + "x-cao-source-stem": { + "type": "string", + "pattern": "^[A-Za-z0-9_-]{1,64}$", + "description": "CAO-generated context-copy provenance. Records the original profile source stem." } }, "additionalProperties": false diff --git a/src/cli_agent_orchestrator/services/install_service.py b/src/cli_agent_orchestrator/services/install_service.py index 1f2f6fa0a..f181267c6 100644 --- a/src/cli_agent_orchestrator/services/install_service.py +++ b/src/cli_agent_orchestrator/services/install_service.py @@ -2,13 +2,17 @@ import logging import os +import platform import re +import secrets +import stat from pathlib import Path from typing import Dict, List, Literal, Optional, Tuple from urllib.parse import urlparse import frontmatter import requests # type: ignore[import-untyped] +import yaml from pydantic import BaseModel from cli_agent_orchestrator.constants import ( @@ -32,6 +36,7 @@ from cli_agent_orchestrator.utils.env import resolve_env_vars, set_env_var from cli_agent_orchestrator.utils.mcp_resolution import resolve_mcp_server_config from cli_agent_orchestrator.utils.opencode_config import ( + OpenCodeAgentIdCollisionError, ensure_skills_symlink, remove_agent_tools, to_opencode_agent_id, @@ -65,6 +70,13 @@ class InstallResult(BaseModel): # CodeQL also recognises this regex as a path-injection sanitiser. _PROFILE_NAME_RE = re.compile(r"^[A-Za-z0-9_-]{1,64}$") +# Context-copy provenance marker — stamped into AGENT_CONTEXT_DIR/.md +# frontmatter to record the original install source stem (the stem/name passed to +# `cao install`). Used by the opencode collision guard to distinguish a profile's +# own installed copy from a different profile that resolves to the same agent id. +_CONTEXT_SOURCE_STEM_KEY = "x-cao-source-stem" +_CONTEXT_SOURCE_STEM_RE = re.compile(rf"^\s*{re.escape(_CONTEXT_SOURCE_STEM_KEY)}\s*:") + # Per-MCP-server tool-call timeout (milliseconds) injected into cao-mcp-server # entries in kiro agent profiles. kiro-cli's default MCP tool-call timeout # (~120s, inherited from the Q Developer CLI) is far too short for the handoff @@ -218,11 +230,363 @@ def parse_env_assignment(env_assignment: str) -> Tuple[str, str]: return key, value -def _write_context_file(agent_name: str, raw_content: str) -> Path: - """Write the unresolved profile source to the shared context directory.""" +def _line_body_and_ending(line: str) -> Tuple[str, str]: + if line.endswith("\r\n"): + return line[:-2], "\r\n" + if line.endswith("\n"): + return line[:-1], "\n" + if line.endswith("\r"): + return line[:-1], "\r" + return line, "" + + +_FRONTMATTER_DELIMITER_RE = re.compile(r"^-{3,}$") + + +def _is_frontmatter_delimiter(line_body: str, *, allow_bom: bool = False) -> bool: + if allow_bom: + line_body = line_body.removeprefix("\ufeff") + # python-frontmatter's YAMLHandler accepts 3+ dashes as a delimiter + # (`^-{3,}\s*$`); matching that here keeps this writer's notion of "where + # the frontmatter block is" in sync with the parser CAO uses everywhere + # else, so real frontmatter with a `----` delimiter is not demoted into + # the body. + return bool(_FRONTMATTER_DELIMITER_RE.match(line_body.strip(" \t"))) + + +def _first_newline(raw_content: str) -> str: + match = re.search(r"\r\n|\n|\r", raw_content) + return match.group(0) if match else "\n" + + +def _parses_as_yaml_mapping(text: str) -> bool: + """Return True if ``text`` is what python-frontmatter would treat as real + frontmatter metadata: a YAML mapping, or empty (``frontmatter.parse`` only + merges ``fm_data`` into ``metadata`` when it is a ``dict``; anything else \u2014 + a bare scalar, a list, invalid YAML \u2014 is silently NOT metadata there). + """ + try: + loaded = yaml.safe_load(text) + except yaml.YAMLError: + return False + return loaded is None or isinstance(loaded, dict) + + +def _find_frontmatter_block(lines: List[str]) -> Optional[Tuple[int, int]]: + """Return opening/closing line indexes for the leading frontmatter block. + + A candidate span only counts as frontmatter if the text between the + delimiters actually parses as a YAML mapping (see + :func:`_parses_as_yaml_mapping`) \u2014 matching what ``frontmatter.loads`` + treats as real metadata, rather than a purely lexical dash match. Without + this, a frontmatter-less document whose body opens with a markdown + thematic break (a line of 3+ dashes) gets mistaken for a frontmatter + opener, the marker gets inserted into the middle of prose, and the + document becomes invalid YAML. + """ + opening_idx: Optional[int] = None + for idx, line in enumerate(lines): + body, _ = _line_body_and_ending(line) + if body.removeprefix("\ufeff").strip(" \t") == "": + continue + if _is_frontmatter_delimiter(body, allow_bom=True): + opening_idx = idx + break + + if opening_idx is None: + return None + + for idx in range(opening_idx + 1, len(lines)): + body, _ = _line_body_and_ending(lines[idx]) + if _is_frontmatter_delimiter(body): + block_text = "".join(lines[opening_idx + 1 : idx]) + if _parses_as_yaml_mapping(block_text): + return opening_idx, idx + return None + return None + + +def _frontmatter_block_indent(lines: List[str], opening_idx: int, closing_idx: int) -> str: + """Return the leading whitespace of the block's first real content line. + + Frontmatter keys are not required to sit at column 0 \u2014 YAML only needs + consistent indentation. Inserting the marker at column 0 into a block + indented some other way breaks that consistency and corrupts the YAML; + matching the block's own indentation keeps it valid. + """ + for idx in range(opening_idx + 1, closing_idx): + body, _ = _line_body_and_ending(lines[idx]) + stripped = body.lstrip(" \t") + if stripped == "" or stripped.startswith("#"): + continue + return body[: len(body) - len(stripped)] + return "" + + +def _yaml_single_quoted(value: str) -> str: + """Render a one-line YAML string scalar.""" + if "\n" in value or "\r" in value: + raise ValueError("Context source stem must fit on one YAML line") + return "'" + value.replace("'", "''") + "'" + + +def _context_marker_line(source_name: str, newline: str) -> str: + return f"{_CONTEXT_SOURCE_STEM_KEY}: {_yaml_single_quoted(source_name)}{newline}" + + +def _context_content_with_provenance(raw_content: str, source_name: str) -> str: + """Return context markdown annotated without reserializing frontmatter. + + If a leading frontmatter block exists, every textually-matching marker + line is removed and a single clean one is inserted in the first matched + line's place (or at the top of the block if none matched). Documents + without a leading block get a minimal frontmatter block prepended, + leaving the original content byte-for-byte intact after that inserted + block. + + The line-regex insertion above only recognises an unquoted, column-0 + ``x-cao-source-stem:`` key. A source profile can carry a marker spelled a + way the regex cannot see (a quoted key, a folded/multi-line value, a + flow-mapping frontmatter document) while PyYAML's parser — the reader + every consumer of this content actually uses — sees it as the *same* key + and would resolve it (last-wins on duplicates) to a value CAO never + wrote. Trusting the regex's view there would let profile content dictate + its own provenance, defeating the guard this marker exists for. So the + assembled content is read back through :func:`_context_source_stem` — + the exact function the collision guard calls — and the install is + refused unless that readback agrees with ``source_name``. This also + catches content the textual insertion accidentally corrupted into + invalid YAML (e.g. a folded scalar's continuation line left orphaned) + before it is ever written to disk. + """ + lines = raw_content.splitlines(keepends=True) + block = _find_frontmatter_block(lines) + if block is None: + newline = _first_newline(raw_content) + marker = _context_marker_line(source_name, newline) + content = f"---{newline}{marker}---{newline}{raw_content}" + else: + opening_idx, closing_idx = block + _, opening_newline = _line_body_and_ending(lines[opening_idx]) + newline = opening_newline or _first_newline(raw_content) + indent = _frontmatter_block_indent(lines, opening_idx, closing_idx) + marker = indent + _context_marker_line(source_name, newline) + + existing_indices = [ + idx + for idx in range(opening_idx + 1, closing_idx) + if _CONTEXT_SOURCE_STEM_RE.match(_line_body_and_ending(lines[idx])[0]) + ] + insert_at = existing_indices[0] if existing_indices else opening_idx + 1 + for idx in reversed(existing_indices): + del lines[idx] + lines.insert(insert_at, marker) + content = "".join(lines) + + try: + verified_stem = _context_source_stem(content) + verify_exc: Optional[Exception] = None + except Exception as exc: + verified_stem = None + verify_exc = exc + if verified_stem != source_name: + if verify_exc is not None: + cause = ( + "the assembled context copy did not parse as valid YAML " + f"frontmatter ({verify_exc})" + ) + elif verified_stem is None: + cause = f"the assembled context copy has no readable '{_CONTEXT_SOURCE_STEM_KEY}' value" + else: + cause = ( + "the assembled context copy reads back " + f"'{_CONTEXT_SOURCE_STEM_KEY}: {verified_stem}' instead of " + f"'{source_name}' — the source profile's own frontmatter " + f"likely defines a conflicting '{_CONTEXT_SOURCE_STEM_KEY}' key" + ) + raise ValueError( + "Refusing to write context copy: could not stamp a trustworthy " + f"'{_CONTEXT_SOURCE_STEM_KEY}' provenance marker for install " + f"source '{source_name}' because {cause}. Fix the source " + "profile's frontmatter (remove or rename the conflicting key, or " + "repair its YAML syntax), then reinstall." + ) + return content + + +def _context_source_stem(raw_content: str) -> Optional[str]: + """Read CAO source-stem provenance from generated context frontmatter.""" + post = frontmatter.loads(raw_content) + value = post.metadata.get(_CONTEXT_SOURCE_STEM_KEY) + if isinstance(value, str): + return value + + lines = raw_content.splitlines(keepends=True) + block = _find_frontmatter_block(lines) + if block is None: + return None + + opening_idx, closing_idx = block + for idx in range(opening_idx + 1, closing_idx): + body, _ = _line_body_and_ending(lines[idx]) + if not _CONTEXT_SOURCE_STEM_RE.match(body): + continue + marker_post = frontmatter.loads(f"---\n{body}\n---\n") + marker_value = marker_post.metadata.get(_CONTEXT_SOURCE_STEM_KEY) + return marker_value if isinstance(marker_value, str) else None + return None + + +def _installed_context_copy_path(stem: str) -> Path: + """Return the installed context path for a discovered installed candidate.""" + from cli_agent_orchestrator.services.settings_service import get_agent_dirs + + installed_dir = Path(get_agent_dirs().get("cao_installed", str(AGENT_CONTEXT_DIR))) + flat = installed_dir / f"{stem}.md" + if flat.exists(): + return flat + nested = installed_dir / stem / "agent.md" + if nested.exists(): + return nested + return flat + + +def _installed_context_copy_remedy(path: Path) -> str: + """Tell operators how to recover from an unproven installed context copy.""" + return ( + f"If '{path}' is your own profile's context copy from an earlier CAO " + "version, delete it and reinstall." + ) + + +def _installed_profile_display( + stem: str, provenance_stem: Optional[str], candidate_path: Path +) -> str: + """Render an installed discovery candidate for collision errors.""" + suffix = f" at '{candidate_path}'" + if provenance_stem: + return f"'{provenance_stem}.md' (installed copy '{stem}.md'{suffix})" + return f"'{stem}.md' (installed copy without CAO source provenance{suffix})" + + +def _raise_unloadable_installed_collision( + target_id: str, source_name: str, profile_name: str, candidate_path: Path +) -> None: + """Block an installed target-slot candidate whose ownership is unknowable.""" + raise OpenCodeAgentIdCollisionError( + f"OpenCode agent id '{target_id}' is already occupied by installed " + f"context copy '{candidate_path}', but CAO cannot read or validate that " + "file, so it cannot prove whether it belongs to the profile being " + f"installed ('{source_name}.md', name '{profile_name}'). The install " + "was refused to avoid silently overwriting existing OpenCode artifacts. " + f"{_installed_context_copy_remedy(candidate_path)}" + ) + + +_TEMP_FILE_NAME_ATTEMPTS = 100 + + +def _non_regular_target_error(context_file: Path) -> ValueError: + return ValueError( + f"Context file '{context_file}' is already occupied by a non-regular " + "filesystem entry. The install was refused to avoid writing through " + "a symlink or overwriting a directory, device, socket, or FIFO. " + "Remove that path or replace it with a regular file, then reinstall." + ) + + +def _create_context_temp_file(context_file: Path) -> Tuple[int, Path]: + """Create a same-directory temp file for the context copy and return its + open fd and path. + + Requests mode 0o666 so the kernel applies the process umask atomically at + file-creation time, instead of this function reading the umask itself via + the ``os.umask(0)``/restore idiom — which briefly zeroes the umask + *process-wide* and can widen the mode of files concurrently created by + other threads (e.g. the API server's ``asyncio.to_thread`` workers). + """ + last_exc: Optional[OSError] = None + for _ in range(_TEMP_FILE_NAME_ATTEMPTS): + candidate = context_file.parent / f".{context_file.name}.{secrets.token_hex(8)}.tmp" + try: + fd = os.open(candidate, os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o666) + except FileExistsError as exc: + last_exc = exc + continue + return fd, candidate + raise OSError( + f"Could not create a unique temporary file next to '{context_file}'" + ) from last_exc + + +def _write_context_file(agent_name: str, raw_content: str, source_name: str) -> Path: + """Write the unresolved profile source to the shared context directory. + + ``agent_name`` is the *resolved* profile name (frontmatter ``name:``) and + determines the filename — the context copy lives at + ``AGENT_CONTEXT_DIR/.md``, NOT under the original install + stem. ``source_name`` is the install *source handle* (the stem/name passed + to ``cao install``), so it can be stamped into the copy's frontmatter under + ``_CONTEXT_SOURCE_STEM_KEY``. The opencode collision guard later uses that + marker to prove "this installed-dir + artifact is a prior copy of the profile being reinstalled" versus "this is + a different profile that resolves to the same agent id" (see + :func:`_guard_opencode_agent_id_collision`). The marker is inserted + textually, preserving source formatting aside from that one marker line. + + The target context path must be absent or a regular file. Symlinks, + directories, FIFOs, sockets, and devices are refused before writing (one + ``lstat`` serves both that check and the existing-mode read below, so + neither follows a symlink planted in the window between them). The + content is written to a same-directory temporary file and atomically + replaced into place so CAO never opens the target path itself and a + failed write does not leave a truncated context copy. The temp file's + mode is restored before the replace to the existing target's mode on a + reinstall (via ``os.fchmod``); a brand-new copy keeps the umask-derived + mode the kernel already applied when the temp file was created — + otherwise ``os.replace`` would carry an unrelated mode onto the target, + silently tightening or widening permissions on every install. + """ AGENT_CONTEXT_DIR.mkdir(parents=True, exist_ok=True) context_file = AGENT_CONTEXT_DIR / f"{agent_name}.md" - context_file.write_text(raw_content, encoding="utf-8") + try: + st = os.lstat(context_file) + except FileNotFoundError: + st = None + if st is not None and not stat.S_ISREG(st.st_mode): + raise _non_regular_target_error(context_file) + existing_mode = stat.S_IMODE(st.st_mode) if st is not None else None + + content = _context_content_with_provenance(raw_content, source_name) + temp_path: Optional[Path] = None + try: + fd, temp_path = _create_context_temp_file(context_file) + with os.fdopen(fd, "w", encoding="utf-8", newline="") as tmp: + tmp.write(content) + tmp.flush() + os.fsync(tmp.fileno()) + if existing_mode is not None and platform.system() != "Windows": + os.fchmod(tmp.fileno(), existing_mode) + os.replace(temp_path, context_file) + except Exception as exc: + if temp_path is not None: + try: + temp_path.unlink() + except FileNotFoundError: + pass + try: + recheck = os.lstat(context_file) + except FileNotFoundError: + recheck = None + if recheck is not None and not stat.S_ISREG(recheck.st_mode): + raise _non_regular_target_error(context_file) from exc + # Name the real target the operator asked to install, not an + # internal, randomly-suffixed temp filename that may no longer even + # exist (e.g. a read-only context dir before the temp file was ever + # created, or a mid-write failure, or a `.tmp` cleaner racing + # `os.replace`). `strerror` (unlike `str(exc)`) never embeds a path. + detail = getattr(exc, "strerror", None) or str(exc) + raise OSError(f"Failed to write context file '{context_file}': {detail}") from exc return context_file @@ -239,6 +603,155 @@ def _build_provider_config( ) +def _guard_opencode_agent_id_collision(source_name: str, profile_name: str) -> None: + """Fail loud if another installable profile shares this profile's agent id. + + The installed OpenCode id derives from a profile's *resolved name* + (frontmatter ``name:``) via :func:`to_opencode_agent_id`, and that + derivation is many-to-one, so TWO distinct profile files can land on the + same ``.md`` / ``agent.`` — whichever installs second silently + overwrites the first. Two independent ways this happens: + + * the non-injective ``'/'`` -> ``'__'`` rewrite (name ``"a/b"`` and a + literal ``"a__b"`` both yield id ``a__b``); and + * two different files carrying the *identical* frontmatter ``name:`` (same + resolved string -> same id). + + Install runs one profile at a time, so this guard reconstructs the id-space + the way installs actually populate it: + + * Discovery (:func:`list_agent_profiles`) keys profiles by file *stem* + (``source_name`` — never contains ``/``), which is the handle you pass to + ``cao install``. + * The installed id, however, derives from the profile's *resolved name* + (frontmatter ``name:``, which MAY contain ``/``), not the stem. + + We resolve every OTHER installable profile's name and compare its id to the + one being installed. Candidates are excluded by *stem* identity + (``source_name``), so every remaining entry is a genuinely different file on + disk: any id match is a real collision — even when the two resolved names + are byte-for-byte identical. That same-resolved-name case must still be + caught, so the exclusion keys on stem (not resolved name) and the guard + raises :class:`OpenCodeAgentIdCollisionError` (a ``ValueError``) as soon as a + different file's id matches. The raised message names both profiles by stem + and resolved name so an operator can find and rename one. + + **Own-copy exception.** ``_write_context_file`` writes each + opencode install's shared context copy to + ``AGENT_CONTEXT_DIR/.md`` — named by the *resolved* name, not + the install stem. When the install stem differs from ``name:`` (e.g. + ``cao install ./my-agent.md`` with ``name: developer``), discovery surfaces + that copy as a separate ``source == "installed"`` candidate whose id + necessarily equals the target id, so a naive guard would flag a profile + against its OWN prior copy and permanently break reinstall/upgrade. We must + NOT fix this by blanket-excluding ``source == "installed"``: that reopens + the silent-overwrite bug (install A → ``cao profile remove A`` drops only + the local-store copy, leaving A's installed artifact → install a DIFFERENT + profile B with a colliding id → B clobbers A with no error). Instead each + installed copy carries a provenance marker (``_CONTEXT_SOURCE_STEM_KEY``) + recording the original install stem, and an installed candidate is skipped + ONLY when that marker proves it is this very profile's prior copy. A missing + marker (a copy written before this marker existed) cannot prove its original + source stem, and legitimate upgrades change the body, so payload equality is + not an identity signal. Markerless installed copies occupying the target id + therefore block with a recovery message instead of being treated as self. + + Only collisions implicating the profile being installed block the install; + a pre-existing clash between two OTHER profiles is left alone. Discovery / + per-profile load failures are non-fatal except for installed candidates + occupying the target id slot: those block because CAO cannot establish + ownership. This guard is still a pre-write check; it does not add file + locking between the check and the write. + """ + try: + from cli_agent_orchestrator.utils.agent_profiles import list_agent_profiles + + candidates = list_agent_profiles() + except Exception as exc: # pragma: no cover - defensive, discovery is best-effort + logger.debug("Skipping OpenCode agent-id collision check: %s", exc) + return + + target_id = to_opencode_agent_id(profile_name) + for candidate in candidates: + stem = candidate.get("name") + candidate_source = candidate.get("source") + # Skip the profile being installed (by its stem/source handle). + # Excluding by STEM (not resolved name) + # is what keeps reinstalling the same profile idempotent while still + # catching a *different* file that resolves to the same name. + if not stem or stem == source_name: + continue + if not candidate.get("loadable", True): + if candidate_source == "installed" and to_opencode_agent_id(stem) == target_id: + _raise_unloadable_installed_collision( + target_id, + source_name, + profile_name, + _installed_context_copy_path(stem), + ) + continue + try: + raw = _read_agent_profile_source(stem) + resolved_name = parse_agent_profile_text(raw, stem).name + except Exception as exc: + if candidate_source == "installed" and to_opencode_agent_id(stem) == target_id: + _raise_unloadable_installed_collision( + target_id, + source_name, + profile_name, + _installed_context_copy_path(stem), + ) + logger.debug("Skipping unreadable profile '%s' in collision check: %s", stem, exc) + continue + + # Own-copy exception: installed-dir candidates need provenance + # checks before we can determine if they're a collision. Check provenance + # BEFORE the id check so we skip self-copies early. + provenance_stem: Optional[str] = None + candidate_path: Optional[Path] = None + if candidate_source == "installed": + candidate_path = _installed_context_copy_path(stem) + try: + provenance_stem = _context_source_stem(raw) + # Marker present and matches: this is our own prior copy. + if provenance_stem == source_name: + continue + except Exception as exc: + if to_opencode_agent_id(stem) == target_id: + _raise_unloadable_installed_collision( + target_id, + source_name, + profile_name, + candidate_path, + ) + logger.debug("Could not read installed-profile provenance for '%s': %s", stem, exc) + + # Now check if this is an actual id collision. + if to_opencode_agent_id(resolved_name) != target_id: + continue + + # A genuinely different file (stem != source_name) whose resolved name + # maps to the same agent id. Raising here (keyed on stem, not resolved + # name) catches the same-resolved-name-different-file case that a plain + # name-string dedup would swallow. + existing_profile = f"'{stem}.md'" + recovery = "" + if candidate_source == "installed" and candidate_path is not None: + existing_profile = _installed_profile_display(stem, provenance_stem, candidate_path) + recovery = f" {_installed_context_copy_remedy(candidate_path)}" + + raise OpenCodeAgentIdCollisionError( + f"OpenCode agent id '{target_id}' is produced by both the profile " + f"being installed ('{source_name}.md', name '{profile_name}') and " + f"the existing profile {existing_profile} (name '{resolved_name}'). Two " + "distinct profiles cannot share an OpenCode agent id: they install " + f"to the same '{target_id}.md' file and 'agent.{target_id}' config " + "section, so the second would silently overwrite the first. Rename " + "one of these profiles (their frontmatter 'name:', after '/' -> " + f"'__' rewriting, must differ).{recovery}" + ) + + def install_agent( source: str, provider: Optional[str] = None, @@ -334,7 +847,6 @@ def install_agent( } unresolved_vars = sorted(set(re.findall(r"\$\{(\w+)\}", resolved_content))) - context_file = _write_context_file(profile.name, raw_content) mcp_server_names = list(profile.mcpServers.keys()) if profile.mcpServers else None allowed_tools = resolve_allowed_tools(profile.allowedTools, profile.role, mcp_server_names) @@ -342,6 +854,18 @@ def install_agent( agent_file: Optional[Path] = None safe_filename = profile.name.replace("/", "__") + # OpenCode collision guard must run BEFORE any destructive write. The + # guard prevents opencode_cli/agents/.md from being overwritten when + # a second profile resolves to the same id, but that is only correct if + # AGENT_CONTEXT_DIR/.md (the shared context file) is also protected. + # Running the guard here — before the context write — ensures a rejected + # install leaves ALL files (provider-specific AND shared) untouched. For + # non-opencode providers, the shared context file is written early (no + # guard needed); for opencode, it is written AFTER the guard passes. + if provider == ProviderType.OPENCODE_CLI.value: + _guard_opencode_agent_id_collision(agent_name, profile.name) + context_file = _write_context_file(profile.name, raw_content, agent_name) + if provider == ProviderType.KIRO_CLI.value: if profile.engine == KiroEngine.KAS: raise ValueError( @@ -420,6 +944,7 @@ def install_agent( mode="all", permission=cao_tools_to_opencode_permission(allowed_tools), ) + agent_id = to_opencode_agent_id(profile.name) agent_file = OPENCODE_AGENTS_DIR / f"{agent_id}.md" agent_file.write_text( diff --git a/src/cli_agent_orchestrator/utils/opencode_config.py b/src/cli_agent_orchestrator/utils/opencode_config.py index 5814050ad..9bfdb0f82 100644 --- a/src/cli_agent_orchestrator/utils/opencode_config.py +++ b/src/cli_agent_orchestrator/utils/opencode_config.py @@ -37,10 +37,31 @@ def to_opencode_agent_id(profile_name: str) -> str: - the value passed to ``opencode --agent `` at runtime Idempotent: inputs that contain no ``/`` are returned unchanged. + + The ``/`` → ``__`` collapse is not injective: distinct profile names can + map to the same id (e.g. ``"a/b"`` and a literal ``"a__b"`` both become + ``"a__b"``). Because the id is used as a dict key on disk (the ``.md`` + filename and the ``agent.`` section of ``opencode.json``), such a + collision silently cross-wires two profiles. The opencode install path + guards against this via ``_guard_opencode_agent_id_collision`` in + ``services/install_service.py``, which fails loud rather than overwriting. + Note names that differ only by spaces or punctuation (``"foo bar"`` vs + ``"foo-bar"``) do NOT collide here — only ``/`` is rewritten. """ return profile_name.replace("/", "__") +class OpenCodeAgentIdCollisionError(ValueError): + """Two distinct profile names collapse to the same OpenCode agent id. + + Raised by the opencode install guard + (``_guard_opencode_agent_id_collision``). Subclasses ``ValueError`` so + existing ``except ValueError`` / broad handlers (e.g. ``install_agent``'s + ``except Exception``) surface it as a clean CLI error rather than a + traceback. + """ + + def ensure_skills_symlink() -> None: """Create ``OPENCODE_CONFIG_DIR/skills`` as a symlink pointing at ``SKILLS_DIR``. diff --git a/test/cli/commands/test_install_opencode.py b/test/cli/commands/test_install_opencode.py index 28a2c7ab0..f8cc64af3 100644 --- a/test/cli/commands/test_install_opencode.py +++ b/test/cli/commands/test_install_opencode.py @@ -12,6 +12,7 @@ from click.testing import CliRunner from cli_agent_orchestrator.cli.commands.install import install +from cli_agent_orchestrator.cli.commands.profile import profile as profile_cmd # --------------------------------------------------------------------------- # Shared fixtures @@ -71,6 +72,62 @@ def install_workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Dict[s } +@pytest.fixture() +def install_workspace_with_installed_dir( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> Dict[str, Any]: + """Redirect install paths while keeping the real default agent-dir discovery shape.""" + local_store = tmp_path / "agent-store" + context_dir = tmp_path / "agent-context" + opencode_agents = tmp_path / "opencode_cli" / "agents" + opencode_config = tmp_path / "opencode_cli" / "opencode.json" + + local_store.mkdir(parents=True) + context_dir.mkdir(parents=True) + + monkeypatch.setattr( + "cli_agent_orchestrator.services.profile_store.LOCAL_AGENT_STORE_DIR", local_store + ) + monkeypatch.setattr( + "cli_agent_orchestrator.utils.agent_profiles.LOCAL_AGENT_STORE_DIR", local_store + ) + monkeypatch.setattr( + "cli_agent_orchestrator.cli.commands.profile.LOCAL_AGENT_STORE_DIR", local_store + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.install_service.AGENT_CONTEXT_DIR", context_dir + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.install_service.OPENCODE_AGENTS_DIR", opencode_agents + ) + monkeypatch.setattr( + "cli_agent_orchestrator.utils.opencode_config.OPENCODE_CONFIG_FILE", opencode_config + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.settings_service.SETTINGS_FILE", + tmp_path / "settings.json", + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.settings_service._DEFAULTS", + { + "kiro_cli": str(tmp_path / "kiro" / "agents"), + "claude_code": str(local_store), + "codex": str(local_store), + "cao_installed": str(context_dir), + }, + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.install_service.ensure_skills_symlink", lambda: None + ) + + return { + "local_store": local_store, + "context_dir": context_dir, + "agents_dir": opencode_agents, + "config_file": opencode_config, + } + + def _write_profile( profile_path: Path, *, @@ -507,6 +564,274 @@ def test_reinstall_without_mcp_removes_agent_tools( ), "stale agent..tools entry must be removed on reinstall without MCP" +# --------------------------------------------------------------------------- +# Agent-id collision guard: '/' -> '__' derivation is not injective +# --------------------------------------------------------------------------- + + +class TestAgentIdCollisionGuard: + """Installing a profile whose id collides with another must fail loud. + + The id derivation replaces '/' with '__', so a profile named ``a/b`` and a + literal profile named ``a__b`` both map to the ``a__b`` id — the second + install would silently overwrite the first's ``a__b.md`` file and + ``agent.a__b`` config. The guard turns that into a clean CLI error. + """ + + def test_real_collision_fails_and_names_both_profiles( + self, runner: CliRunner, install_workspace: Dict[str, Any] + ): + store = install_workspace["local_store"] + # A sibling profile that literally occupies the "a__b" id. + _write_profile(store / "a__b.md", name="a__b") + # The profile we install has frontmatter name "a/b" -> id "a__b". + # File stem must be a legal source name; the '/' lives in frontmatter. + _write_profile(store / "slash-named.md", name="a/b") + # profile.name "a/b" → context path context_dir/a/b.md; pre-create the + # intermediate dir so the context write (which runs before the provider + # branch) doesn't fail before the collision guard is reached. + (install_workspace["context_dir"] / "a").mkdir(parents=True, exist_ok=True) + + result = runner.invoke(install, ["slash-named", "--provider", "opencode_cli"]) + + assert result.exit_code == 0 # install_agent returns a failure result, not a crash + assert "Error:" in result.output + assert "a/b" in result.output and "a__b" in result.output + # The colliding profile must NOT have been written under the shared id. + # Only the pre-existing sibling (if installed) could own a__b.md; the + # slash-named install must be refused before writing. + assert not (install_workspace["agents_dir"] / "a__b.md").exists() + + def test_same_resolved_name_different_stem_fails_and_preserves_first( + self, runner: CliRunner, install_workspace: Dict[str, Any] + ): + """Two distinct files with the IDENTICAL frontmatter name collide. + + Both ``profile-one.md`` and ``profile-two.md`` carry ``name: shared-alias``, + so both resolve to the id ``shared-alias`` — the same ``shared-alias.md`` + file and ``agent.shared-alias`` config section. The second install must + fail (naming both files/name) rather than silently overwrite the first. + This is the same-resolved-name-different-stem gap: the id derivation is + many-to-one on the *name* even without any '/' rewrite. + """ + store = install_workspace["local_store"] + # Install the first profile while it is the only one on disk. + _write_profile(store / "profile-one.md", name="shared-alias", body="First agent body.") + r1 = runner.invoke(install, ["profile-one", "--provider", "opencode_cli"]) + assert r1.exit_code == 0 and "Error:" not in r1.output + agent_file = install_workspace["agents_dir"] / "shared-alias.md" + assert agent_file.exists() + first_contents = agent_file.read_text() + assert "First agent body." in first_contents + + # Also capture the shared context file from the first install. + context_file = install_workspace["context_dir"] / "shared-alias.md" + assert context_file.exists() + first_context = context_file.read_text() + + # A second, DIFFERENT file later appears with the same resolved name. + _write_profile(store / "profile-two.md", name="shared-alias", body="Second agent body.") + + # Second install (different file, same resolved name) must be refused. + r2 = runner.invoke(install, ["profile-two", "--provider", "opencode_cli"]) + assert r2.exit_code == 0 # returns a failure result, not a crash + assert "Error:" in r2.output + # The error must name both offending profiles and the shared name. + assert "profile-one" in r2.output + assert "profile-two" in r2.output + assert "shared-alias" in r2.output + # The first install's file must be intact — NOT overwritten by the second. + assert agent_file.read_text() == first_contents + assert "Second agent body." not in agent_file.read_text() + # Regression check: the shared context file must ALSO be + # preserved. Before the fix, the guard ran AFTER _write_context_file(), + # so the rejected second install would corrupt AGENT_CONTEXT_DIR/.md + # even though opencode_cli/agents/.md was protected. + assert context_file.read_text() == first_context + assert "Second agent body." not in context_file.read_text() + + def test_reinstall_same_profile_stays_idempotent_despite_guard( + self, runner: CliRunner, install_workspace: Dict[str, Any] + ): + """Reinstalling the SAME profile (same stem) must not trip the guard. + + The guard excludes candidates by stem, so a profile never collides with + itself even though discovery lists it with its own resolved name. + """ + _write_profile(install_workspace["local_store"] / "test-agent.md", name="test-agent") + + r1 = runner.invoke(install, ["test-agent", "--provider", "opencode_cli"]) + r2 = runner.invoke(install, ["test-agent", "--provider", "opencode_cli"]) + + assert r1.exit_code == 0 and "Error:" not in r1.output + assert r2.exit_code == 0 and "Error:" not in r2.output + assert (install_workspace["agents_dir"] / "test-agent.md").exists() + + def test_non_colliding_spaces_vs_dash_both_install( + self, runner: CliRunner, install_workspace: Dict[str, Any] + ): + """ "foo bar" and "foo-bar" do NOT collide (only '/' is rewritten).""" + store = install_workspace["local_store"] + _write_profile(store / "foo-space.md", name="foo bar") + _write_profile(store / "foo-dash.md", name="foo-bar") + + r1 = runner.invoke(install, ["foo-space", "--provider", "opencode_cli"]) + r2 = runner.invoke(install, ["foo-dash", "--provider", "opencode_cli"]) + + assert r1.exit_code == 0 and "Error:" not in r1.output + assert r2.exit_code == 0 and "Error:" not in r2.output + # Distinct ids => distinct files, both present. + assert (install_workspace["agents_dir"] / "foo bar.md").exists() + assert (install_workspace["agents_dir"] / "foo-bar.md").exists() + + def test_normal_single_profile_install_unaffected( + self, runner: CliRunner, install_workspace: Dict[str, Any] + ): + """The guard is a no-op for an ordinary, non-colliding profile.""" + _write_profile(install_workspace["local_store"] / "solo-agent.md", name="solo-agent") + + result = runner.invoke(install, ["solo-agent", "--provider", "opencode_cli"]) + + assert result.exit_code == 0 + assert "Error:" not in result.output + assert (install_workspace["agents_dir"] / "solo-agent.md").exists() + + +# --------------------------------------------------------------------------- +# Provenance guard: installed copies with provenance markers +# --------------------------------------------------------------------------- + + +class TestAgentIdCollisionGuardInstalledProvenance: + """Collision checks must distinguish an installed self-copy from another profile.""" + + def test_stem_not_name_reinstall_succeeds_with_installed_copy_discovered( + self, runner: CliRunner, install_workspace_with_installed_dir: Dict[str, Any] + ): + store = install_workspace_with_installed_dir["local_store"] + _write_profile(store / "my-agent.md", name="resolved-agent", body="Resolved v1.") + + r1 = runner.invoke(install, ["my-agent", "--provider", "opencode_cli"]) + r2 = runner.invoke(install, ["my-agent", "--provider", "opencode_cli"]) + + assert r1.exit_code == 0 and "Error:" not in r1.output + assert r2.exit_code == 0 and "Error:" not in r2.output + + context_file = install_workspace_with_installed_dir["context_dir"] / "resolved-agent.md" + post = frontmatter.loads(context_file.read_text()) + assert post.metadata["x-cao-source-stem"] == "my-agent" + assert post.metadata["name"] == "resolved-agent" + assert post.content.strip() == "Resolved v1." + + def test_multiple_stem_not_name_reinstall_cycles_remain_idempotent( + self, runner: CliRunner, install_workspace_with_installed_dir: Dict[str, Any] + ): + store = install_workspace_with_installed_dir["local_store"] + _write_profile(store / "my-agent.md", name="resolved-agent", body="Resolved v1.") + + first_agent_bytes = None + first_context_bytes = None + for _ in range(4): + result = runner.invoke(install, ["my-agent", "--provider", "opencode_cli"]) + assert result.exit_code == 0 and "Error:" not in result.output + + agent_bytes = ( + install_workspace_with_installed_dir["agents_dir"] / "resolved-agent.md" + ).read_bytes() + context_bytes = ( + install_workspace_with_installed_dir["context_dir"] / "resolved-agent.md" + ).read_bytes() + if first_agent_bytes is None: + first_agent_bytes = agent_bytes + first_context_bytes = context_bytes + assert agent_bytes == first_agent_bytes + assert context_bytes == first_context_bytes + + def test_different_local_profiles_with_same_agent_id_still_raise( + self, runner: CliRunner, install_workspace_with_installed_dir: Dict[str, Any] + ): + store = install_workspace_with_installed_dir["local_store"] + _write_profile(store / "profile-a.md", name="shared-alias", body="First profile.") + _write_profile(store / "profile-b.md", name="shared-alias", body="Second profile.") + + result = runner.invoke(install, ["profile-b", "--provider", "opencode_cli"]) + + assert result.exit_code == 0 + assert "Error:" in result.output + assert "profile-a" in result.output + assert "profile-b" in result.output + assert not (install_workspace_with_installed_dir["agents_dir"] / "shared-alias.md").exists() + + def test_removed_local_profile_leaves_installed_copy_that_still_blocks_collision( + self, runner: CliRunner, install_workspace_with_installed_dir: Dict[str, Any] + ): + store = install_workspace_with_installed_dir["local_store"] + _write_profile(store / "profile-a.md", name="shared-alias", body="First profile.") + + r1 = runner.invoke(install, ["profile-a", "--provider", "opencode_cli"]) + assert r1.exit_code == 0 and "Error:" not in r1.output + + agent_file = install_workspace_with_installed_dir["agents_dir"] / "shared-alias.md" + context_file = install_workspace_with_installed_dir["context_dir"] / "shared-alias.md" + first_agent = agent_file.read_text() + first_context = context_file.read_text() + + removed = runner.invoke(profile_cmd, ["remove", "profile-a", "-y"]) + assert removed.exit_code == 0, removed.output + assert not (store / "profile-a.md").exists() + assert context_file.exists() + + _write_profile(store / "profile-b.md", name="shared-alias", body="Second profile.") + r2 = runner.invoke(install, ["profile-b", "--provider", "opencode_cli"]) + + assert r2.exit_code == 0 + assert "Error:" in r2.output + assert "profile-a" in r2.output + assert "profile-b" in r2.output + assert agent_file.read_text() == first_agent + assert context_file.read_text() == first_context + assert "Second profile." not in agent_file.read_text() + assert "Second profile." not in context_file.read_text() + + def test_legacy_installed_copy_without_marker_blocks_same_profile_reinstall( + self, runner: CliRunner, install_workspace_with_installed_dir: Dict[str, Any] + ): + store = install_workspace_with_installed_dir["local_store"] + _write_profile(store / "legacy-source.md", name="legacy-agent", body="Legacy profile.") + legacy_copy = install_workspace_with_installed_dir["context_dir"] / "legacy-agent.md" + _write_profile(legacy_copy, name="legacy-agent", body="Legacy profile.") + first_context = legacy_copy.read_text() + + result = runner.invoke(install, ["legacy-source", "--provider", "opencode_cli"]) + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + assert str(legacy_copy) in result.output + assert ( + f"If '{legacy_copy}' is your own profile's context copy from an earlier " + "CAO version, delete it and reinstall." + ) in result.output + assert legacy_copy.read_text() == first_context + + def test_legacy_installed_copy_without_marker_still_blocks_id_alias_collision( + self, runner: CliRunner, install_workspace_with_installed_dir: Dict[str, Any] + ): + store = install_workspace_with_installed_dir["local_store"] + legacy_copy = install_workspace_with_installed_dir["context_dir"] / "a__b.md" + _write_profile(legacy_copy, name="a__b", body="Legacy profile.") + first_context = legacy_copy.read_text() + + _write_profile(store / "slash-named.md", name="a/b", body="Second profile.") + result = runner.invoke(install, ["slash-named", "--provider", "opencode_cli"]) + + assert result.exit_code == 0 + assert "Error:" in result.output + assert "slash-named" in result.output + assert "a__b" in result.output + assert legacy_copy.read_text() == first_context + assert "Second profile." not in legacy_copy.read_text() + + # --------------------------------------------------------------------------- # Optional live smoke test: opencode agent list shows the installed agent # --------------------------------------------------------------------------- diff --git a/test/cli/commands/test_install_opencode_provenance.py b/test/cli/commands/test_install_opencode_provenance.py new file mode 100644 index 000000000..24e3f2c61 --- /dev/null +++ b/test/cli/commands/test_install_opencode_provenance.py @@ -0,0 +1,938 @@ +"""Provenance-based opencode collision guard tests (PR #493). + +Unlike ``test_install_opencode.py`` — which monkeypatches ``get_agent_dirs`` to +``{}`` and so never scans the CAO installed dir — these tests keep the DEFAULT +``cao_installed`` mapping pointed at the real context dir. That is the only way +to reproduce the regression: on reinstall, ``list_agent_profiles()`` discovers +the profile's OWN prior context copy (written by ``_write_context_file`` under +the RESOLVED name) as a ``source == "installed"`` candidate, and a naive guard +flags a profile against itself. + +Two problems must hold simultaneously: + +* **A (regression fixed):** reinstalling a profile whose install stem differs + from its frontmatter ``name:`` must succeed (its own installed copy must not + be mistaken for a colliding profile). +* **B (must not reopen):** two GENUINELY different profiles that resolve to the + same opencode agent id must still raise — including the trap case where the + first profile's local-store copy has been removed but its installed artifact + survives. +""" + +import os +import stat +from pathlib import Path +from typing import Any, Dict + +import frontmatter +import pytest +from click.testing import CliRunner + +from cli_agent_orchestrator.cli.commands.install import install +from cli_agent_orchestrator.services.install_service import ( + _CONTEXT_SOURCE_STEM_KEY, + _context_content_with_provenance, + _context_source_stem, +) + + +@pytest.fixture() +def runner() -> CliRunner: + return CliRunner() + + +@pytest.fixture() +def workspace(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Dict[str, Any]: + """Redirect install paths to tmp while keeping the default cao_installed mapping. + + Crucially ``get_agent_dirs`` returns ``{"cao_installed": }`` — + the same directory ``_write_context_file`` writes to — so discovery scans + the installed copies exactly as it does in production. ``AGENT_CONTEXT_DIR`` + is pointed at that same dir so the provenance read and the context write + agree. + """ + local_store = tmp_path / "agent-store" + context_dir = tmp_path / "agent-context" + opencode_agents = tmp_path / "opencode_cli" / "agents" + opencode_config = tmp_path / "opencode_cli" / "opencode.json" + kiro_agents = tmp_path / "kiro" / "agents" + + local_store.mkdir(parents=True) + context_dir.mkdir(parents=True) + + monkeypatch.setattr( + "cli_agent_orchestrator.services.profile_store.LOCAL_AGENT_STORE_DIR", local_store + ) + monkeypatch.setattr( + "cli_agent_orchestrator.utils.agent_profiles.LOCAL_AGENT_STORE_DIR", local_store + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.install_service.AGENT_CONTEXT_DIR", context_dir + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.install_service.OPENCODE_AGENTS_DIR", opencode_agents + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.install_service.KIRO_AGENTS_DIR", kiro_agents + ) + monkeypatch.setattr( + "cli_agent_orchestrator.utils.opencode_config.OPENCODE_CONFIG_FILE", opencode_config + ) + # DEFAULT mapping preserved (NOT {}): cao_installed points at the context dir. + monkeypatch.setattr( + "cli_agent_orchestrator.services.settings_service.get_agent_dirs", + lambda: {"cao_installed": str(context_dir)}, + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.settings_service.get_extra_agent_dirs", lambda: [] + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.settings_service.get_disabled_agent_dirs", lambda: [] + ) + monkeypatch.setattr( + "cli_agent_orchestrator.services.install_service.ensure_skills_symlink", lambda: None + ) + + return { + "local_store": local_store, + "context_dir": context_dir, + "agents_dir": opencode_agents, + "config_file": opencode_config, + "kiro_agents_dir": kiro_agents, + } + + +def _write_profile(path: Path, *, name: str, body: str = "You are a helpful agent.") -> None: + path.write_text(f"---\nname: {name}\ndescription: Test agent\n---\n{body}\n", encoding="utf-8") + + +def _install(runner: CliRunner, stem: str): + return runner.invoke(install, [stem, "--provider", "opencode_cli"]) + + +def _assert_installed_copy_remedy(output: str, context_copy: Path) -> None: + assert str(context_copy) in output + assert ( + f"If '{context_copy}' is your own profile's context copy from an earlier " + "CAO version, delete it and reinstall." + ) in output + + +def _assert_non_regular_context_error(output: str, context_copy: Path) -> None: + assert str(context_copy) in output + assert "non-regular filesystem entry" in output + assert "Remove that path or replace it with a regular file, then reinstall." in output + assert "Errno" not in output + + +def _line_body_and_ending(line: str) -> tuple[str, str]: + if line.endswith("\r\n"): + return line[:-2], "\r\n" + if line.endswith("\n"): + return line[:-1], "\n" + if line.endswith("\r"): + return line[:-1], "\r" + return line, "" + + +def _remove_marker_line(text: str) -> str: + lines = text.splitlines(keepends=True) + for idx, line in enumerate(lines): + body, _ = _line_body_and_ending(line) + if body.lstrip(" \t").startswith(f"{_CONTEXT_SOURCE_STEM_KEY}:"): + return "".join(lines[:idx] + lines[idx + 1 :]) + raise AssertionError("marker line not found") + + +def _quoted_marker_line(source_name: str, newline: str = "\n") -> str: + return f"{_CONTEXT_SOURCE_STEM_KEY}: '{source_name.replace(chr(39), chr(39) * 2)}'{newline}" + + +def _assert_inserted_marker_only(raw: str, stamped: str, source_name: str) -> None: + assert _context_source_stem(stamped) == source_name + assert stamped.count(f"{_CONTEXT_SOURCE_STEM_KEY}:") == 1 + assert _remove_marker_line(stamped).encode("utf-8") == raw.encode("utf-8") + + +# --------------------------------------------------------------------------- +# Problem A: reinstall/upgrade of a stem != name profile must succeed. +# (Fails on b67ba96 — the guard flags the profile's own installed copy.) +# --------------------------------------------------------------------------- + + +class TestProblemAReinstallStemNotEqualName: + def test_reinstall_stem_ne_name_succeeds( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + # Install stem 'my-agent' whose frontmatter name is 'my-resolved'. The + # context copy lands at context/my-resolved.md (named by RESOLVED name). + _write_profile(workspace["local_store"] / "my-agent.md", name="my-resolved") + + r1 = _install(runner, "my-agent") + assert r1.exit_code == 0 and "Error:" not in r1.output, r1.output + + # The installed copy exists under the resolved name and carries the marker. + context_copy = workspace["context_dir"] / "my-resolved.md" + assert context_copy.exists() + assert frontmatter.loads(context_copy.read_text()).metadata[_CONTEXT_SOURCE_STEM_KEY] == ( + "my-agent" + ) + + # Reinstall the SAME profile. On b67ba96 this raises the collision error + # (own copy mistaken for a sibling); on the fix it must succeed. + r2 = _install(runner, "my-agent") + assert r2.exit_code == 0, r2.output + assert "Error:" not in r2.output, r2.output + assert (workspace["agents_dir"] / "my-resolved.md").exists() + + def test_idempotent_across_multiple_cycles( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + _write_profile(workspace["local_store"] / "my-agent.md", name="my-resolved") + + first = None + for _ in range(4): + r = _install(runner, "my-agent") + assert r.exit_code == 0 and "Error:" not in r.output, r.output + contents = (workspace["agents_dir"] / "my-resolved.md").read_bytes() + if first is None: + first = contents + else: + assert contents == first, "reinstall must be byte-identical" + + def test_kiro_then_opencode_same_profile_succeeds( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + _write_profile(workspace["local_store"] / "shared-source.md", name="shared-resolved") + + kiro = runner.invoke(install, ["shared-source", "--provider", "kiro_cli"]) + opencode = runner.invoke(install, ["shared-source", "--provider", "opencode_cli"]) + + assert kiro.exit_code == 0 and "Error:" not in kiro.output, kiro.output + assert opencode.exit_code == 0 and "Error:" not in opencode.output, opencode.output + context_copy = workspace["context_dir"] / "shared-resolved.md" + assert _context_source_stem(context_copy.read_text(encoding="utf-8")) == "shared-source" + assert (workspace["kiro_agents_dir"] / "shared-resolved.json").exists() + assert (workspace["agents_dir"] / "shared-resolved.md").exists() + + +class TestContextWriteTargetIsRegularFile: + def test_dangling_symlink_target_blocks_without_creating_link_target( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + context_dir = workspace["context_dir"] + _write_profile(store / "source.md", name="shared", body="New body.") + context_copy = context_dir / "shared.md" + link_target = context_dir / "dangling-target.md" + context_copy.symlink_to(link_target) + + result = _install(runner, "source") + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + _assert_non_regular_context_error(result.output, context_copy) + assert context_copy.is_symlink() + assert not link_target.exists() + assert not (workspace["agents_dir"] / "shared.md").exists() + + def test_live_symlink_inside_context_dir_blocks_without_modifying_target( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + context_dir = workspace["context_dir"] + _write_profile(store / "source.md", name="shared", body="New body.") + context_copy = context_dir / "shared.md" + link_target = context_dir / "real-target.md" + original_target = "---\nname: shared\ndescription: Existing target\n---\nDo not modify.\n" + link_target.write_text(original_target, encoding="utf-8") + context_copy.symlink_to(link_target) + + result = runner.invoke(install, ["source", "--provider", "kiro_cli"]) + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + _assert_non_regular_context_error(result.output, context_copy) + assert context_copy.is_symlink() + assert link_target.read_text(encoding="utf-8") == original_target + assert not (workspace["kiro_agents_dir"] / "shared.json").exists() + + def test_directory_target_blocks_with_actionable_message( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + context_dir = workspace["context_dir"] + _write_profile(store / "source.md", name="shared", body="New body.") + context_copy = context_dir / "shared.md" + context_copy.mkdir() + + result = _install(runner, "source") + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + _assert_non_regular_context_error(result.output, context_copy) + assert context_copy.is_dir() + assert not (workspace["agents_dir"] / "shared.md").exists() + + +# --------------------------------------------------------------------------- +# Problem B: genuinely different profiles that collide must still raise. +# --------------------------------------------------------------------------- + + +class TestProblemBLocalStoreCollisionStillRaises: + def test_two_local_store_profiles_same_id_raise( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + """Two distinct local-store files with the same resolved name collide. + + This mirrors the existing guard test but with the DEFAULT cao_installed + mapping active, proving the provenance change did not weaken the + local-store collision path. + """ + store = workspace["local_store"] + _write_profile(store / "one.md", name="dup-name", body="First body.") + + r1 = _install(runner, "one") + assert r1.exit_code == 0 and "Error:" not in r1.output, r1.output + + # A second, distinct file with the same resolved name appears later. + _write_profile(store / "two.md", name="dup-name", body="Second body.") + r2 = _install(runner, "two") + assert r2.exit_code == 0 # failure result, not a crash + assert "Error:" in r2.output + assert "one" in r2.output and "two" in r2.output and "dup-name" in r2.output + # First install untouched. + assert "First body." in (workspace["agents_dir"] / "dup-name.md").read_text() + + +# --------------------------------------------------------------------------- +# THE TRAP: install A -> remove A's local-store copy -> install distinct B +# with a colliding id. Guard must RAISE and A's installed artifacts survive. +# A blanket `source == "installed"` exclusion would silently overwrite here. +# --------------------------------------------------------------------------- + + +class TestTrapCaseLeftoverInstalledArtifact: + def test_leftover_installed_artifact_blocks_distinct_profile( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + agents_dir = workspace["agents_dir"] + context_dir = workspace["context_dir"] + + # 1. Install profile A: stem 'alpha', resolved name 'shared'. + _write_profile(store / "alpha.md", name="shared", body="Alpha body.") + rA = _install(runner, "alpha") + assert rA.exit_code == 0 and "Error:" not in rA.output, rA.output + + a_agent = agents_dir / "shared.md" + a_context = context_dir / "shared.md" + assert a_agent.exists() and a_context.exists() + a_agent_bytes = a_agent.read_bytes() + a_context_bytes = a_context.read_bytes() + + # 2. Emulate `cao profile remove alpha`: it deletes ONLY the local-store + # copy (cli/commands/profile.py), leaving the installed artifacts. + (store / "alpha.md").unlink() + + # 3. Install a DIFFERENT profile B: stem 'beta', resolved name 'shared' + # (same id as A). The only surviving trace of A is its installed + # context copy, whose provenance marker records stem 'alpha' != 'beta'. + _write_profile(store / "beta.md", name="shared", body="Beta body.") + rB = _install(runner, "beta") + + # Must RAISE — not silently overwrite. + assert rB.exit_code == 0 # failure result, not a crash + assert "Error:" in rB.output, rB.output + assert "shared" in rB.output + + # A's installed artifacts must be byte-for-byte intact. + assert a_agent.read_bytes() == a_agent_bytes + assert a_context.read_bytes() == a_context_bytes + assert "Beta body." not in a_agent.read_text() + assert "Beta body." not in a_context.read_text() + + +# --------------------------------------------------------------------------- +# Criterion 5: pre-existing installs (no provenance marker) cannot prove +# ownership, so they must block instead of being silently overwritten. +# --------------------------------------------------------------------------- + + +class TestMissingMarkerPreExistingInstalls: + def test_markerless_installed_copy_blocks_distinct_profile_with_same_id( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + context_dir = workspace["context_dir"] + + legacy_copy = context_dir / "shared.md" + _write_profile(legacy_copy, name="shared", body="Original installed body.") + first_context = legacy_copy.read_text() + assert _CONTEXT_SOURCE_STEM_KEY not in first_context + + _write_profile(store / "new-source.md", name="shared", body="New profile body.") + result = _install(runner, "new-source") + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + _assert_installed_copy_remedy(result.output, legacy_copy) + assert legacy_copy.read_text() == first_context + assert "New profile body." not in legacy_copy.read_text() + assert not (workspace["agents_dir"] / "shared.md").exists() + + def test_markerless_installed_copy_blocks_same_profile_reinstall( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + context_dir = workspace["context_dir"] + + _write_profile(store / "my-agent.md", name="my-resolved") + legacy_copy = context_dir / "my-resolved.md" + _write_profile(legacy_copy, name="my-resolved", body="Legacy context body.") + first_context = legacy_copy.read_text() + assert _CONTEXT_SOURCE_STEM_KEY not in first_context + + r = _install(runner, "my-agent") + + assert r.exit_code == 0 + assert "Error:" in r.output, r.output + _assert_installed_copy_remedy(r.output, legacy_copy) + assert legacy_copy.read_text() == first_context + assert not (workspace["agents_dir"] / "my-resolved.md").exists() + + def test_truncated_installed_copy_without_marker_blocks_install( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + context_dir = workspace["context_dir"] + legacy_copy = context_dir / "shared.md" + legacy_copy.write_text("---\nname: shared\ndescription: Partial legacy\n", encoding="utf-8") + first_context = legacy_copy.read_text() + + _write_profile(store / "new-source.md", name="shared", body="New profile body.") + result = _install(runner, "new-source") + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + _assert_installed_copy_remedy(result.output, legacy_copy) + assert legacy_copy.read_text() == first_context + assert not (workspace["agents_dir"] / "shared.md").exists() + + def test_corrupt_installed_copy_blocks_install( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + context_dir = workspace["context_dir"] + legacy_copy = context_dir / "shared.md" + legacy_copy.write_text( + "---\nname: [unterminated\ndescription: Corrupt legacy\n---\nBody\n", + encoding="utf-8", + ) + first_context = legacy_copy.read_text() + + _write_profile(store / "new-source.md", name="shared", body="New profile body.") + result = _install(runner, "new-source") + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + _assert_installed_copy_remedy(result.output, legacy_copy) + assert legacy_copy.read_text() == first_context + assert not (workspace["agents_dir"] / "shared.md").exists() + + def test_legacy_slash_trap_still_raises_without_marker( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + """The '/'->'__' collision still raises even with a marker-less sibling. + + The colliding sibling is a local-store file (source 'local', never + 'installed'), so the provenance fallback does not apply to it and the + normal collision path fires. + """ + store = workspace["local_store"] + # Literal 'a__b' sibling installs first, alone. + _write_profile(store / "a__b.md", name="a__b") + r1 = _install(runner, "a__b") + assert r1.exit_code == 0 and "Error:" not in r1.output, r1.output + + # A slash-named profile ('a/b' -> id 'a__b') appears later and collides. + _write_profile(store / "slash-named.md", name="a/b") + # This pre-creates the slash-name context parent because a separate + # _write_context_file defect still affects successful slash-name writes. + (workspace["context_dir"] / "a").mkdir(parents=True, exist_ok=True) + r2 = _install(runner, "slash-named") + assert r2.exit_code == 0 # failure result, not a crash + assert "Error:" in r2.output, r2.output + assert "a__b" in r2.output + + +class TestNonOpenCodeContextWrites: + def test_kiro_context_copy_differs_from_source_only_by_marker_line( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + source_text = ( + "---\n" + "# leading comment that must survive\n" + 'provider: "kiro_cli"\n' + "tags:\n" + " - beta\n" + " - alpha\n" + 'description: "Quoted description: keep style"\n' + 'name: "kiro-byte-copy"\n' + "# trailing frontmatter comment must survive\n" + "---\n" + "Body keeps ${UNSET_VAR}, unicode café, and markdown # headings untouched.\n" + ) + source = workspace["local_store"] / "kiro-byte-copy.md" + source.write_text(source_text, encoding="utf-8") + + result = runner.invoke(install, ["kiro-byte-copy", "--provider", "kiro_cli"]) + + assert result.exit_code == 0 + assert "Error:" not in result.output, result.output + context_text = (workspace["context_dir"] / "kiro-byte-copy.md").read_text(encoding="utf-8") + _assert_inserted_marker_only(source_text, context_text, "kiro-byte-copy") + assert (workspace["kiro_agents_dir"] / "kiro-byte-copy.json").exists() + + +class TestTextualProvenanceInsert: + def test_frontmatter_shape_cases_preserve_all_non_marker_bytes(self) -> None: + cases = [ + ( + "empty frontmatter block", + "---\n---\nBody\n", + "empty-source", + ), + ( + "CRLF line endings", + "---\r\nname: crlf-agent\r\ndescription: CRLF\r\n---\r\nBody\r\n", + "crlf-source", + ), + ( + "body delimiter is not frontmatter delimiter", + "---\nname: body-delimiter\ndescription: Body delimiter\n---\nBody\n---\nStill body\n", + "body-source", + ), + ( + "leading blank lines before opening delimiter", + "\n\n---\nname: leading-blank\ndescription: Leading blank\n---\nBody\n", + "blank-source", + ), + ( + "BOM before opening delimiter", + "\ufeff---\nname: bom-agent\ndescription: BOM\n---\nBody\n", + "bom-source", + ), + ( + "nested structures comments quoting sequences and unicode", + "---\n" + "# leading comment\n" + 'description: "Quoted: keep # literal"\n' + "name: unicode-agent\n" + "capabilities:\n" + ' - "quoted item"\n' + " - café\n" + "settings:\n" + " nested:\n" + " - keep: order\n" + "# trailing comment\n" + "---\n" + "Unicode body café\n", + "nested-source", + ), + ( + "YAML-ambiguous source stem is quoted", + "---\nname: ambiguous\ndescription: Ambiguous source\n---\nBody\n", + "yes", + ), + ( + "source stem with YAML punctuation is quoted", + "---\nname: punctuation\ndescription: Punctuation source\n---\nBody\n", + "odd: # value ' stem", + ), + ] + + for label, raw, source_name in cases: + stamped = _context_content_with_provenance(raw, source_name) + try: + _assert_inserted_marker_only(raw, stamped, source_name) + except AssertionError as exc: + raise AssertionError(label) from exc + + def test_no_frontmatter_gets_minimal_marker_block(self) -> None: + raw = "Body without frontmatter.\n---\nThis delimiter belongs to the body.\n" + + stamped = _context_content_with_provenance(raw, "plain-source") + + assert _context_source_stem(stamped) == "plain-source" + assert stamped == f"---\n{_quoted_marker_line('plain-source')}---\n{raw}" + + def test_existing_marker_line_is_replaced_in_place(self) -> None: + raw = ( + "---\n" + "name: replace-agent\n" + f"{_CONTEXT_SOURCE_STEM_KEY}: old-source\n" + "description: Replacement keeps neighboring lines\n" + "---\n" + "Body\n" + ) + + stamped = _context_content_with_provenance(raw, "new-source") + + assert _context_source_stem(stamped) == "new-source" + assert stamped == raw.replace( + f"{_CONTEXT_SOURCE_STEM_KEY}: old-source\n", + _quoted_marker_line("new-source"), + 1, + ) + assert stamped.count(f"{_CONTEXT_SOURCE_STEM_KEY}:") == 1 + + def test_duplicate_plain_marker_lines_are_all_removed_before_reinsertion(self) -> None: + """Two competing UNQUOTED marker lines are a fixable case, not just a + raise-worthy one: both are deleted and one clean line is inserted, so + the reader's last-wins duplicate-key resolution can never disagree + with what CAO intended to stamp.""" + raw = ( + "---\n" + f"{_CONTEXT_SOURCE_STEM_KEY}: 'profile-b'\n" + "name: shared\n" + f"{_CONTEXT_SOURCE_STEM_KEY}: 'z'\n" + "description: D\n" + "---\n" + "Body\n" + ) + + stamped = _context_content_with_provenance(raw, "profile-a") + + assert _context_source_stem(stamped) == "profile-a" + assert stamped.count(f"{_CONTEXT_SOURCE_STEM_KEY}:") == 1 + post = frontmatter.loads(stamped) + assert post.metadata["name"] == "shared" + assert post.metadata["description"] == "D" + + +class TestProvenanceMarkerSpoofRefused: + """R1: a marker spelling the textual regex cannot see must not silently + win over the one CAO wrote. Round 3's frontmatter.loads/dumps round-trip + gave this property for free (a dict assignment collapses every spelling + of the key into one entry); the textual inserter must re-earn it by + verifying its own output through the same YAML path the guard reads. + Every shape here reads back a value CAO never wrote if the write is + trusted blindly, so the install must be refused instead of silently + stamping (and therefore trusting) the wrong stem. + """ + + @pytest.mark.parametrize( + "label,extra_frontmatter_line", + [ + ("double-quoted-key", "\"x-cao-source-stem\": 'bbb'\n"), + ("single-quoted-key", "'x-cao-source-stem': 'bbb'\n"), + ], + ) + def test_quoted_key_spoof_refuses_install( + self, label: str, extra_frontmatter_line: str + ) -> None: + raw = f"---\nname: shared\ndescription: A\n{extra_frontmatter_line}---\nA BODY\n" + + with pytest.raises(ValueError, match=_CONTEXT_SOURCE_STEM_KEY): + _context_content_with_provenance(raw, "aaa") + + def test_flow_mapping_frontmatter_refuses_install(self) -> None: + raw = "---\n{x-cao-source-stem: profile-b, name: shared}\n---\nBody\n" + + with pytest.raises(ValueError, match=_CONTEXT_SOURCE_STEM_KEY): + _context_content_with_provenance(raw, "profile-a") + + def test_folded_scalar_marker_value_refuses_install(self) -> None: + """R3: a multi-line marker value would otherwise leave an orphaned + continuation line (invalid YAML) in the written copy. Refusing here + means the corrupt copy is never written, so reinstall is never + permanently blocked by CAO's own output.""" + raw = "---\nx-cao-source-stem: >\n old\nname: foo\n---\nbody\n" + + with pytest.raises(ValueError, match=_CONTEXT_SOURCE_STEM_KEY): + _context_content_with_provenance(raw, "mystem") + + def test_end_to_end_install_of_spoofed_profile_fails_cleanly( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + """CLI-level: a profile carrying a quoted-key marker must fail to + install at all, rather than install successfully with a provenance + marker that lies about its own stem (the exact precondition the R1 + end-to-end silent-overwrite scenario in the review depends on).""" + store = workspace["local_store"] + context_dir = workspace["context_dir"] + (store / "aaa.md").write_text( + "---\nname: shared\ndescription: A\n\"x-cao-source-stem\": 'bbb'\n---\nA BODY\n", + encoding="utf-8", + ) + + result = _install(runner, "aaa") + + assert result.exit_code == 0 # failure result, not a crash + assert "Error:" in result.output, result.output + assert not (context_dir / "shared.md").exists() + assert not (workspace["agents_dir"] / "shared.md").exists() + + def test_trap_case_still_raises_when_a_never_installed_due_to_spoof( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + """End-to-end regression proof for the review's exact repro: since A + (carrying the spoofed marker) never installs at all, B's later + install of the same resolved name has nothing to collide with and + must succeed cleanly — the opposite of a silent overwrite, and not a + crash either.""" + store = workspace["local_store"] + agents_dir = workspace["agents_dir"] + + (store / "aaa.md").write_text( + "---\nname: shared\ndescription: A\n\"x-cao-source-stem\": 'bbb'\n---\nA BODY\n", + encoding="utf-8", + ) + rA = _install(runner, "aaa") + assert rA.exit_code == 0 + assert "Error:" in rA.output, rA.output + assert not (agents_dir / "shared.md").exists() + + (store / "aaa.md").unlink() + _write_profile(store / "bbb.md", name="shared", body="B BODY") + rB = _install(runner, "bbb") + + assert rB.exit_code == 0 and "Error:" not in rB.output, rB.output + assert "B BODY" in (agents_dir / "shared.md").read_text() + + +class TestFourDashFrontmatterDelimiter: + """R2: python-frontmatter accepts 3+ dashes as a delimiter; the writer + must recognise the same shapes or it demotes real frontmatter into the + body.""" + + def test_four_dash_delimiters_are_recognized_as_frontmatter(self) -> None: + raw = "----\nname: dashy-name\ndescription: D\n----\nBODY\n" + + stamped = _context_content_with_provenance(raw, "dashy") + + assert _context_source_stem(stamped) == "dashy" + post = frontmatter.loads(stamped) + assert post.metadata["name"] == "dashy-name" + assert post.metadata["description"] == "D" + assert post.content.strip() == "BODY" + + def test_four_dash_delimiters_preserve_bytes_outside_marker(self) -> None: + raw = "----\nname: dashy-name\ndescription: D\n----\nBODY\n" + + stamped = _context_content_with_provenance(raw, "dashy") + + _assert_inserted_marker_only(raw, stamped, "dashy") + + def test_end_to_end_install_recognizes_four_dash_frontmatter( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + (store / "dashy.md").write_text( + "----\nname: dashy-name\ndescription: D\n----\nBODY\n", encoding="utf-8" + ) + + result = _install(runner, "dashy") + + assert result.exit_code == 0 and "Error:" not in result.output, result.output + context_text = (workspace["context_dir"] / "dashy-name.md").read_text(encoding="utf-8") + post = frontmatter.loads(context_text) + assert post.metadata["name"] == "dashy-name" + assert post.metadata["description"] == "D" + + +class TestFrontmatterlessBodyOpeningWithDashRule: + """Regression: the R2 delimiter widening (``^-{3,}$``) made + ``_find_frontmatter_block`` open a "frontmatter block" at ANY leading + line of 3+ dashes, including a markdown thematic-break rule in a + frontmatter-less document. The marker then got inserted into the middle + of prose, the assembled content was invalid YAML, and the readback gate + (correctly) refused the install — but the refusal took down a profile + shape that fixup 4 (and pre-PR CAO) installed successfully. A leading + dash line only counts as frontmatter if what follows it actually parses + as a YAML mapping; a bare markdown rule does not, and must fall back to + the ordinary "no leading block -> prepend a clean one" path. + """ + + def test_three_dash_rule_body_installs_and_copy_parses(self) -> None: + raw = "---\n\n# My Agent\n\nDoes stuff.\n\n---\n\n## Details\n" + + stamped = _context_content_with_provenance(raw, "profile-a") + + assert _context_source_stem(stamped) == "profile-a" + post = frontmatter.loads(stamped) # must not raise + assert post.metadata[_CONTEXT_SOURCE_STEM_KEY] == "profile-a" + # Prepend form: original bytes intact after the inserted block. + assert stamped == f"---\n{_quoted_marker_line('profile-a')}---\n{raw}" + + def test_four_dash_rule_body_installs_and_copy_parses(self) -> None: + raw = "----\n\n# My Agent\n\nDoes stuff.\n\n----\n\n## Details\n" + + stamped = _context_content_with_provenance(raw, "profile-a") + + assert _context_source_stem(stamped) == "profile-a" + frontmatter.loads(stamped) # must not raise + assert stamped == f"---\n{_quoted_marker_line('profile-a')}---\n{raw}" + + def test_end_to_end_install_of_dash_rule_body_succeeds( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + store_agent = store / "ruled.md" + store_agent.write_text( + "----\n\n# My Agent\n\nDoes stuff.\n\n----\n\n## Details\n", encoding="utf-8" + ) + + result = _install(runner, "ruled") + + assert result.exit_code == 0 and "Error:" not in result.output, result.output + context_copy = workspace["context_dir"] / "ruled.md" + assert context_copy.exists() + # The written copy must itself round-trip through the parser CAO + # uses everywhere else, not merely "the CLI exited 0". + post = frontmatter.loads(context_copy.read_text(encoding="utf-8")) + assert post.metadata[_CONTEXT_SOURCE_STEM_KEY] == "ruled" + + def test_indented_frontmatter_keys_install_and_copy_parses(self) -> None: + """Real frontmatter whose keys are indented (still a valid YAML + mapping) must have the marker inserted at the SAME indentation, not + column 0 — a column-0 insertion breaks the block's indentation + consistency and corrupts the YAML.""" + raw = "---\n name: indented-name\n description: D\n---\nBody\n" + + stamped = _context_content_with_provenance(raw, "profile-a") + + assert _context_source_stem(stamped) == "profile-a" + post = frontmatter.loads(stamped) + assert post.metadata["name"] == "indented-name" + assert post.metadata["description"] == "D" + assert post.content.strip() == "Body" + _assert_inserted_marker_only(raw, stamped, "profile-a") + + def test_end_to_end_install_of_indented_frontmatter_succeeds( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + store = workspace["local_store"] + (store / "indented.md").write_text( + "---\n name: indented-name\n description: D\n---\nBody\n", encoding="utf-8" + ) + + result = _install(runner, "indented") + + assert result.exit_code == 0 and "Error:" not in result.output, result.output + context_copy = workspace["context_dir"] / "indented-name.md" + post = frontmatter.loads(context_copy.read_text(encoding="utf-8")) + assert post.metadata["name"] == "indented-name" + assert post.metadata[_CONTEXT_SOURCE_STEM_KEY] == "indented" + + def test_refusal_message_names_real_cause_not_a_nonexistent_key(self) -> None: + """A profile that genuinely has no ``x-cao-source-stem`` key must + never be told to go remove one if it is (still, safely) refused.""" + raw = "---\nfoo: |\n ----\nx-cao-source-stem: 'old'\n---\nBody\n" + + with pytest.raises(ValueError) as excinfo: + _context_content_with_provenance(raw, "profile-a") + + message = str(excinfo.value) + # This shape DOES carry a real conflicting key, so blaming it is + # accurate here — the point of the assertion is that the message + # names the readback's actual disagreement, not a boilerplate guess. + assert "reads back" in message + assert "'profile-a'" in message + + def test_spoof_matrix_stays_closed(self) -> None: + """The mapping-validity check must not reopen any previously-closed + spoofing shape: quoted key, duplicate plain marker, folded value, + flow mapping, and the 4-dash delimiter must all behave exactly as + before this fix.""" + control = _context_content_with_provenance("---\nname: a\n---\nBody\n", "profile-a") + assert _context_source_stem(control) == "profile-a" + + with pytest.raises(ValueError, match=_CONTEXT_SOURCE_STEM_KEY): + _context_content_with_provenance( + '---\n"x-cao-source-stem": "evil"\nname: a\n---\nBody\n', "profile-a" + ) + with pytest.raises(ValueError, match=_CONTEXT_SOURCE_STEM_KEY): + _context_content_with_provenance( + "---\n'x-cao-source-stem': 'evil'\nname: a\n---\nBody\n", "profile-a" + ) + with pytest.raises(ValueError, match=_CONTEXT_SOURCE_STEM_KEY): + _context_content_with_provenance( + "---\nx-cao-source-stem: >\n evil\nname: a\n---\nBody\n", "profile-a" + ) + with pytest.raises(ValueError, match=_CONTEXT_SOURCE_STEM_KEY): + _context_content_with_provenance( + "---\n{x-cao-source-stem: evil, name: a}\n---\nBody\n", "profile-a" + ) + + duplicate = _context_content_with_provenance( + "---\nx-cao-source-stem: evil\nname: a\nx-cao-source-stem: evil2\n---\nBody\n", + "profile-a", + ) + assert _context_source_stem(duplicate) == "profile-a" + + four_dash = _context_content_with_provenance("----\nname: a\n----\nBody\n", "profile-a") + assert _context_source_stem(four_dash) == "profile-a" + + +class TestContextFileModePreservation: + """R4: os.replace() carries the temp file's mode (always 0600 from + tempfile.NamedTemporaryFile) onto the target, so without restoring the + mode first, every reinstall silently tightens an existing copy's + permissions and every brand-new copy is 0600 instead of umask-derived.""" + + def test_reinstall_preserves_existing_target_mode( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + _write_profile(workspace["local_store"] / "mode-agent.md", name="mode-agent") + r1 = _install(runner, "mode-agent") + assert r1.exit_code == 0 and "Error:" not in r1.output, r1.output + + context_copy = workspace["context_dir"] / "mode-agent.md" + os.chmod(context_copy, 0o644) + + r2 = _install(runner, "mode-agent") + assert r2.exit_code == 0 and "Error:" not in r2.output, r2.output + + mode = stat.S_IMODE(context_copy.stat().st_mode) + assert mode == 0o644, oct(mode) + + def test_brand_new_copy_uses_umask_default_not_hardcoded_0600( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + _write_profile(workspace["local_store"] / "new-agent.md", name="new-agent") + + old_umask = os.umask(0o022) + try: + result = _install(runner, "new-agent") + finally: + os.umask(old_umask) + + assert result.exit_code == 0 and "Error:" not in result.output, result.output + context_copy = workspace["context_dir"] / "new-agent.md" + mode = stat.S_IMODE(context_copy.stat().st_mode) + assert mode == 0o644, oct(mode) + + +class TestReadOnlyContextDirErrorNamesRealTarget: + """R5: a read-only context dir must report the target path the user + actually cares about, not tempfile's internal randomly-named ``.tmp`` + file (which no longer exists by the time the error is shown).""" + + def test_permission_denied_names_target_not_temp_file( + self, runner: CliRunner, workspace: Dict[str, Any] + ) -> None: + _write_profile(workspace["local_store"] / "ro-agent.md", name="ro-agent") + context_dir = workspace["context_dir"] + os.chmod(context_dir, 0o500) + try: + result = _install(runner, "ro-agent") + finally: + os.chmod(context_dir, 0o700) + + assert result.exit_code == 0 + assert "Error:" in result.output, result.output + assert str(context_dir / "ro-agent.md") in result.output + assert ".tmp" not in result.output diff --git a/test/utils/test_opencode_config.py b/test/utils/test_opencode_config.py index b01cdd3d2..a485ba2ef 100644 --- a/test/utils/test_opencode_config.py +++ b/test/utils/test_opencode_config.py @@ -8,9 +8,11 @@ import cli_agent_orchestrator.utils.opencode_config as cfg_module from cli_agent_orchestrator.utils.opencode_config import ( + OpenCodeAgentIdCollisionError, ensure_skills_symlink, read_config, remove_agent_tools, + to_opencode_agent_id, translate_mcp_server_config, upsert_agent_tools, upsert_mcp_server, @@ -26,6 +28,33 @@ def tmp_config(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): return config_file +class TestToOpencodeAgentId: + """to_opencode_agent_id rewrites '/' -> '__' and nothing else.""" + + def test_slash_replaced_with_double_underscore(self): + assert to_opencode_agent_id("a/b") == "a__b" + + def test_no_slash_unchanged(self): + assert to_opencode_agent_id("developer") == "developer" + + def test_spaces_and_punctuation_untouched(self): + # Documents that spaces/dashes are NOT rewritten (only '/'), so they + # cannot cause a collision in this scheme. + assert to_opencode_agent_id("foo bar") == "foo bar" + assert to_opencode_agent_id("foo-bar") == "foo-bar" + + +class TestOpenCodeAgentIdCollisionError: + """The collision error type contract the install guard relies on.""" + + def test_collision_is_valueerror_subclass(self): + # _guard_opencode_agent_id_collision raises this; install_agent()'s + # broad `except Exception` turns it into a clean CLI "Error:" message. + # (The end-to-end collision behaviour is covered in + # test/cli/commands/test_install_opencode.py against the real CLI.) + assert issubclass(OpenCodeAgentIdCollisionError, ValueError) + + class TestEnsureSkillsSymlink: """ensure_skills_symlink() creates/validates the skills → SKILLS_DIR symlink."""