From 9ce9a036178ed0069ec7ea52b6fcf1320f608527 Mon Sep 17 00:00:00 2001 From: Dennis F <13630119+RomeoRaven@users.noreply.github.com> Date: Sat, 22 Aug 2026 22:32:57 -0400 Subject: [PATCH] feat: add Codex snapshot migration slice --- README.md | 12 +- __init__.py | 14 +- boundaries.py | 42 ++++ docs/IMPLEMENTATION_PLAN.md | 2 +- export_bundle.py | 124 ++++++++++++ migration.py | 363 +++++++++++++++++++++++++++++++++++ protoagent.plugin.yaml | 6 +- pyproject.toml | 4 +- requirements-dev.txt | 1 + skills/codex-bridge/SKILL.md | 6 +- stores.py | 314 ++++++++++++++++++++++++++++++ tests/conftest.py | 103 ++++++++++ tests/test_export_bundle.py | 100 ++++++++++ tests/test_migration.py | 237 +++++++++++++++++++++++ tests/test_scaffold.py | 14 +- tests/test_stores.py | 98 ++++++++++ tests/test_tools.py | 77 ++++++++ tools_import.py | 140 ++++++++++++++ 18 files changed, 1635 insertions(+), 22 deletions(-) create mode 100644 boundaries.py create mode 100644 export_bundle.py create mode 100644 migration.py create mode 100644 stores.py create mode 100644 tests/conftest.py create mode 100644 tests/test_export_bundle.py create mode 100644 tests/test_migration.py create mode 100644 tests/test_stores.py create mode 100644 tests/test_tools.py create mode 100644 tools_import.py diff --git a/README.md b/README.md index cec0d84..24653bf 100644 --- a/README.md +++ b/README.md @@ -6,7 +6,7 @@ Codex Bridge is a dedicated [protoAgent](https://github.com/protoLabsAI/protoAgent) plugin for migrating selected operator-owned [Codex](https://developers.openai.com/codex) state into a new protoAgent agent. -Status: foundation and source contract. Migration tools are not implemented yet. Track the first vertical slice in this repository's GitHub issues. +Status: v0.1 snapshot migration candidate. Source inventory, decision-grade review, skill/instruction/memory/MCP translation, and secret-free snapshot export are implemented. Fresh-agent apply remains a later tranche. ## Planned v0.1 scope @@ -57,6 +57,16 @@ The bridge will never read or migrate: - Fresh protoAgent agent creation only after explicit acknowledgement. - No source mutation, in-place target merge, credential transfer, or plugin installation. +## Available tools + +| Tool | Purpose | +|---|---| +| `codex_sources` | Report availability of configured user and repository roots without discovering others | +| `codex_import_scan` | Build a write-nothing human or JSON review and plan SHA-256 | +| `codex_export_snapshot` | Preview or write a secret-free protoAgent snapshot archive | + +Snapshot export is dry-run by default. This candidate does not yet create a protoAgent agent. + ## Planned delivery 1. Source inventory and classification. diff --git a/__init__.py b/__init__.py index c2f27a4..3caa31c 100644 --- a/__init__.py +++ b/__init__.py @@ -1,9 +1,4 @@ -"""Codex Bridge foundation plugin. - -The foundation registers only its bundled planning skill. Source inventory, -translation, snapshot, and acknowledged-apply tools land through repository -issues and the implementation plan. -""" +"""Codex Bridge preview and snapshot plugin.""" from __future__ import annotations @@ -13,6 +8,9 @@ def register(registry) -> None: - """Register the bundled Codex Bridge planning guidance.""" + """Register Codex source-review and snapshot-export tools plus guidance.""" + from .tools_import import build_tools + + registry.register_tools(build_tools(dict(getattr(registry, "config", {}) or {}))) registry.register_skill_dir("skills") - log.info("[codex_bridge] foundation registered; migration tools are not implemented") + log.info("[codex_bridge] registered preview and snapshot-export tools") diff --git a/boundaries.py b/boundaries.py new file mode 100644 index 0000000..17b3b19 --- /dev/null +++ b/boundaries.py @@ -0,0 +1,42 @@ +"""Fenced, complete, byte-capped reads for Codex Bridge.""" + +from __future__ import annotations + +from pathlib import Path + + +class ReadLimitExceeded(ValueError): + """A source file exceeds its configured complete-read cap.""" + + +class FencedRoot: + def __init__(self, name: str, root: str | Path) -> None: + self.name = name + self.root = Path(root).expanduser().resolve() + + def resolve(self, rel: str = "") -> Path: + rel = str(rel or "").strip() + parts = Path(rel).parts + if rel.startswith(("/", "~")) or ".." in parts: + raise ValueError(f"path escapes the {self.name} root: {rel!r}") + target = (self.root / rel).resolve() if rel else self.root + if target != self.root and not target.is_relative_to(self.root): + raise ValueError(f"path escapes the {self.name} root: {rel!r}") + return target + + def read_bytes(self, rel: str, max_bytes: int, *, limit_name: str = "max_read_bytes") -> bytes: + try: + limit = int(max_bytes) + except (TypeError, ValueError) as exc: + raise ValueError(f"{limit_name} must be a positive integer") from exc + if limit <= 0: + raise ValueError(f"{limit_name} must be a positive integer") + target = self.resolve(rel) + with target.open("rb") as handle: + data = handle.read(limit + 1) + if len(data) > limit: + raise ReadLimitExceeded(f"{rel} exceeds {limit_name}={limit}") + return data + + def read_text(self, rel: str, max_bytes: int) -> str: + return self.read_bytes(rel, max_bytes).decode("utf-8", errors="strict") diff --git a/docs/IMPLEMENTATION_PLAN.md b/docs/IMPLEMENTATION_PLAN.md index d6de222..2efa581 100644 --- a/docs/IMPLEMENTATION_PLAN.md +++ b/docs/IMPLEMENTATION_PLAN.md @@ -1,6 +1,6 @@ # Codex Bridge implementation plan -Status: approved foundation; v0.1 migration implementation pending. +Status: v0.1 snapshot migration candidate. Inventory, decision-grade review, translation, and secret-free snapshot export are implemented. Fresh-agent apply and release qualification remain later tranches. ## Goal diff --git a/export_bundle.py b/export_bundle.py new file mode 100644 index 0000000..94bd755 --- /dev/null +++ b/export_bundle.py @@ -0,0 +1,124 @@ +"""Build secret-free protoAgent snapshots from Codex migration plans.""" + +from __future__ import annotations + +import io +import zipfile +from dataclasses import dataclass +from datetime import UTC, datetime +from typing import Any + +import yaml + +from .migration import MigrationPlan + + +SNAPSHOT_MANIFEST = "agent.snapshot.yaml" +SNAPSHOT_VERSION = 1 + + +@dataclass(frozen=True) +class BundleReceipt: + agent_name: str + skills: list[str] + mcp_servers: list[str] + knowledge: dict[str, int] + required_values: list[dict[str, Any]] + + +def _knowledge_text(plan: MigrationPlan, domain: str) -> str: + sections = [] + for document in plan.knowledge.get(domain, []): + sections.append( + f"## {document.heading}\n\n" + f"_Imported from Codex {document.source_class} source `{document.source}`; " + f"precedence {document.precedence}._\n\n" + f"{document.content}" + ) + return "\n\n".join(sections) + + +def _review(receipt: BundleReceipt, plan: MigrationPlan, exported_at: str) -> str: + lines = [ + f"# Snapshot review: {receipt.agent_name}", + "", + f"Translated from selected Codex sources and exported {exported_at}.", + "", + "This is a fresh-agent recipe, not a Codex backup. Credentials, sessions, databases, logs, caches, plugin code, and raw history are excluded.", + "", + "No plugins are pinned. Importing this snapshot installs no third-party plugin code.", + "", + f"Plan SHA-256: `{plan.review_dict()['plan_sha256']}`", + "", + "## Required values", + "", + ] + if receipt.required_values: + lines += ["| Value | Present in source |", "|---|---|"] + lines += [ + f"| `{item['name']}` | {'yes' if item.get('was_set') else 'no'} |" for item in receipt.required_values + ] + else: + lines.append("None detected.") + lines += [ + "", + "Names only are included. Source values stayed behind.", + "", + "## Report-only state", + "", + ] + lines += [f"- `{item}`" for item in plan.report_only] or ["None detected."] + lines += [ + "", + "Knowledge documents may contain private operator material. Review them before sharing this archive.", + ] + return "\n".join(lines) + "\n" + + +def build_bundle( + *, + agent_name: str, + migration: MigrationPlan, + now: datetime | None = None, +) -> tuple[bytes, BundleReceipt]: + if migration.problems: + raise ValueError("migration plan contains unresolved migration problems") + stamp = (now or datetime.now(UTC)).isoformat() + knowledge_counts = {domain: len(documents) for domain, documents in migration.knowledge.items() if documents} + receipt = BundleReceipt( + agent_name=agent_name, + skills=[skill.name for skill in migration.skills], + mcp_servers=[str(server.get("name") or "?") for server in migration.mcp_servers], + knowledge=knowledge_counts, + required_values=list(migration.required_values), + ) + config: dict[str, Any] = {"identity": {"name": agent_name}} + if migration.mcp_servers: + config["mcp"] = {"servers": migration.mcp_servers} + manifest = { + "snapshot_version": SNAPSHOT_VERSION, + "kind": "agent-snapshot", + "exported_at": stamp, + "agent": {"name": agent_name}, + "plugins": [], + "config": config, + "soul": None, + "required_secrets": migration.required_values, + "knowledge": {"domains": knowledge_counts} if knowledge_counts else None, + "excludes": { + "credentials": "all Codex credential values remain in their source stores", + "runtime_state": "sessions, databases, logs, caches, plugins, and raw history are excluded", + "policy": "model, approval, sandbox, command-rule, and provider policy are report-only", + }, + "translated_from": {"agent": "codex"}, + } + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive: + archive.writestr(SNAPSHOT_MANIFEST, yaml.safe_dump(manifest, sort_keys=False, allow_unicode=True)) + for skill in migration.skills: + for rel, content in skill.files.items(): + archive.writestr(f"skills/codex/{skill.name}/{rel}", content) + for domain in sorted(knowledge_counts): + archive.writestr(f"knowledge/{domain}.md", _knowledge_text(migration, domain)) + archive.writestr("REVIEW.md", _review(receipt, migration, stamp)) + return buffer.getvalue(), receipt diff --git a/migration.py b/migration.py new file mode 100644 index 0000000..e30da7c --- /dev/null +++ b/migration.py @@ -0,0 +1,363 @@ +"""Pure Codex-to-protoAgent planning and translation.""" + +from __future__ import annotations + +import copy +import hashlib +import json +import re +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +import yaml + +from .boundaries import FencedRoot +from .stores import InstructionSource, SkillSource, SourceScan, SourceSelection, scan_sources + + +_NAME_RE = re.compile(r"[^a-z0-9]+") +_HARD_EXCLUDED_SUPPORT_PARTS = { + ".git", + "cache", + "caches", + "log", + "logs", + "plugin", + "plugins", + "sessions", + "shell_snapshots", + "tmp", + "temp", +} +_HARD_EXCLUDED_SUPPORT_NAMES = {".env", "auth.json", "installation_id", "models_cache.json"} +_HARD_EXCLUDED_SUPPORT_SUFFIXES = (".jsonl", ".sqlite", ".db", ".wal", ".shm", "-wal", "-shm") + + +@dataclass +class TranslatedSkill: + name: str + description: str + source_class: str + source: str + files: dict[str, bytes] + source_files: dict[str, dict[str, Any]] + + def as_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "description": self.description, + "source_class": self.source_class, + "source": self.source, + "source_files": copy.deepcopy(self.source_files), + "files": { + path: {"sha256": hashlib.sha256(content).hexdigest(), "bytes": len(content)} + for path, content in sorted(self.files.items()) + }, + } + + +@dataclass(frozen=True) +class KnowledgeDocument: + domain: str + heading: str + content: str + source: str + source_class: str + precedence: int = 0 + + def as_dict(self) -> dict[str, Any]: + return { + "domain": self.domain, + "heading": self.heading, + "content": self.content, + "source": self.source, + "source_class": self.source_class, + "precedence": self.precedence, + } + + +@dataclass +class MigrationPlan: + selection: SourceSelection + max_read_bytes: int = 65_536 + max_skill_bytes: int = 262_144 + skills: list[TranslatedSkill] = field(default_factory=list) + knowledge: dict[str, list[KnowledgeDocument]] = field(default_factory=dict) + mcp_servers: list[dict[str, Any]] = field(default_factory=list) + required_values: list[dict[str, Any]] = field(default_factory=list) + report_only: list[str] = field(default_factory=list) + excluded: list[str] = field(default_factory=list) + problems: list[str] = field(default_factory=list) + source_files: dict[str, dict[str, Any]] = field(default_factory=dict) + + def as_dict(self) -> dict[str, Any]: + return { + "source": "codex", + "limits": { + "supporting_file_bytes": self.max_read_bytes, + "primary_skill_bytes": self.max_skill_bytes, + }, + "skills": [item.as_dict() for item in self.skills], + "knowledge": { + domain: [document.as_dict() for document in documents] + for domain, documents in sorted(self.knowledge.items()) + }, + "mcp_servers": copy.deepcopy(self.mcp_servers), + "required_values": copy.deepcopy(self.required_values), + "source_files": copy.deepcopy(self.source_files), + "report_only": list(self.report_only), + "excluded": list(self.excluded), + "problems": list(self.problems), + } + + def review_dict(self) -> dict[str, Any]: + return { + "source": "codex", + "limits": self.as_dict()["limits"], + "skills": [item.as_dict() for item in self.skills], + "knowledge": { + domain: [ + { + "heading": doc.heading, + "source": doc.source, + "source_class": doc.source_class, + "precedence": doc.precedence, + "sha256": hashlib.sha256(doc.content.encode("utf-8")).hexdigest(), + "bytes": len(doc.content.encode("utf-8")), + } + for doc in documents + ] + for domain, documents in sorted(self.knowledge.items()) + }, + "mcp_servers": copy.deepcopy(self.mcp_servers), + "required_values": copy.deepcopy(self.required_values), + "source_files": copy.deepcopy(self.source_files), + "report_only": list(self.report_only), + "excluded": list(self.excluded), + "problems": list(self.problems), + "plan_sha256": plan_fingerprint(self), + } + + def render(self) -> str: + lines = ["Codex migration dry run"] + lines.append( + f"limits: primary SKILL.md={self.max_skill_bytes} bytes; supporting file={self.max_read_bytes} bytes" + ) + lines.extend( + f"skill {item.source_class}: {item.name} ({item.source}; files={len(item.files)})" for item in self.skills + ) + lines.extend( + f"knowledge {domain}: {doc.heading} ({doc.source}; precedence={doc.precedence})" + for domain, documents in sorted(self.knowledge.items()) + for doc in documents + ) + lines.extend(f"MCP server: {server['name']}" for server in self.mcp_servers) + lines.extend( + f"required value: {item['name']} (was_set={str(item['was_set']).lower()})" for item in self.required_values + ) + lines.extend(f"report-only: {item}" for item in self.report_only) + lines.extend(f"excluded: {item}" for item in self.excluded) + lines.extend(f"problem: {item}" for item in self.problems) + lines.append(f"plan sha256: {plan_fingerprint(self)}") + lines.append("DRY RUN: source and protoAgent state were not modified") + return "\n".join(lines) + + +def _slug(value: object) -> str: + slug = _NAME_RE.sub("-", str(value or "").strip().lower()).strip("-") + return slug[:64].rstrip("-") or "imported-skill" + + +def _frontmatter(text: str) -> tuple[dict[str, Any], str]: + if not text.startswith("---"): + return {}, text + try: + _, raw, body = text.split("---", 2) + parsed = yaml.safe_load(raw) + except (ValueError, yaml.YAMLError): + return {}, text + return (parsed if isinstance(parsed, dict) else {}), body.lstrip("\n") + + +def _hard_excluded_support(rel: str) -> bool: + path = Path(rel) + parts = {part.lower() for part in path.parts} + name = path.name.lower() + return ( + bool(parts & _HARD_EXCLUDED_SUPPORT_PARTS) + or name in _HARD_EXCLUDED_SUPPORT_NAMES + or name.endswith(_HARD_EXCLUDED_SUPPORT_SUFFIXES) + ) + + +def _translate_skill( + source: SkillSource, + max_read_bytes: int, + max_skill_bytes: int, + excluded: list[str], +) -> TranslatedSkill: + fence = FencedRoot(f"{source.source_class} skill {source.name}", source.path) + raw_skill = fence.read_bytes("SKILL.md", max_skill_bytes, limit_name="max_skill_bytes") + text = raw_skill.decode("utf-8", errors="strict") + meta, body = _frontmatter(text) + name = _slug(meta.get("name") or source.name) + description = str(meta.get("description") or f"Imported Codex skill {name}.").strip()[:1024] + translated_meta: dict[str, Any] = {"name": name, "description": description} + if meta.get("tools"): + translated_meta["tools"] = meta["tools"] + translated_meta["metadata"] = { + "imported-from": "codex", + "source-class": source.source_class, + "source-path": source.relative_path, + } + rendered = "---\n" + yaml.safe_dump(translated_meta, sort_keys=False, allow_unicode=True).strip() + "\n---\n\n" + files = {"SKILL.md": (rendered + body.strip() + "\n").encode("utf-8")} + source_files = {"SKILL.md": {"sha256": hashlib.sha256(raw_skill).hexdigest(), "bytes": len(raw_skill)}} + for path in sorted(source.path.rglob("*")): + if path.is_symlink() or not path.is_file() or path.name == "SKILL.md": + continue + rel = str(path.relative_to(source.path)) + if _hard_excluded_support(rel): + excluded.append(f"skill {source.name}/{rel}: hard excluded") + continue + raw = fence.read_bytes(rel, max_read_bytes) + files[rel] = raw + source_files[rel] = {"sha256": hashlib.sha256(raw).hexdigest(), "bytes": len(raw)} + return TranslatedSkill( + name=name, + description=description, + source_class=source.source_class, + source=source.relative_path, + files=files, + source_files=source_files, + ) + + +def _knowledge(scan: SourceScan) -> dict[str, list[KnowledgeDocument]]: + result: dict[str, list[KnowledgeDocument]] = {} + for name, content in scan.memory_docs.items(): + domain = "codex-memory" if name == "MEMORY.md" else "codex-memory-summary" + result.setdefault(domain, []).append( + KnowledgeDocument( + domain=domain, + heading="Codex curated memory" if name == "MEMORY.md" else "Codex memory summary", + content=content, + source=f"memories/{name}", + source_class="codex-home", + ) + ) + for instruction in scan.instructions: + result.setdefault("codex-instructions", []).append(_instruction_document(instruction)) + return result + + +def _instruction_document(source: InstructionSource) -> KnowledgeDocument: + return KnowledgeDocument( + domain="codex-instructions", + heading=f"Codex instructions: {source.relative_path}", + content=source.content, + source=source.relative_path, + source_class="repository", + precedence=source.precedence, + ) + + +def _required(name: str, kind: str, value: object) -> dict[str, Any]: + if isinstance(value, (list, dict)): + was_set = bool(value) + else: + was_set = bool(str(value or "").strip()) + return { + "name": name, + "kind": kind, + "was_set": was_set, + "description": f"Codex source value for {name!r} was removed during migration.", + } + + +def _sanitize_mcp( + raw_servers: dict[str, dict[str, Any]], +) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str]]: + servers: list[dict[str, Any]] = [] + required: list[dict[str, Any]] = [] + report_only: list[str] = [] + scalar_fields = ("command", "cwd", "url", "bearer_token_env_var") + list_fields = ("args",) + mapping_fields = ("env", "http_headers", "env_http_headers") + structural_fields = ("enabled", "transport", "startup_timeout_sec", "tool_timeout_sec") + supported = set(scalar_fields + list_fields + mapping_fields + structural_fields) + for name, raw in sorted(raw_servers.items()): + entry: dict[str, Any] = {"name": name} + report_only.extend(f"mcp.{name}.unsupported.{field_name}" for field_name in sorted(set(raw) - supported)) + for field_name in structural_fields: + if field_name in raw: + required.append(_required(f"mcp.{name}.{field_name}", "mcp_setting", raw[field_name])) + for field_name in scalar_fields: + if field_name not in raw: + continue + entry[field_name] = "" + required.append(_required(f"mcp.{name}.{field_name}", f"mcp_{field_name}", raw[field_name])) + for field_name in list_fields: + values = raw.get(field_name) + if not isinstance(values, list): + continue + entry[field_name] = [] + required.append(_required(f"mcp.{name}.{field_name}", "mcp_args", values)) + for field_name in mapping_fields: + values = raw.get(field_name) + if not isinstance(values, dict): + continue + entry[field_name] = {str(key): "" for key in sorted(values)} + for key, value in sorted(values.items()): + required.append( + _required( + f"mcp.{name}.{field_name}.{key}", + "mcp_env" if field_name == "env" else "mcp_header", + value, + ) + ) + servers.append(entry) + return servers, required, report_only + + +def build_migration_plan( + selection: SourceSelection, + *, + max_read_bytes: int = 65_536, + max_skill_bytes: int = 262_144, +) -> MigrationPlan: + scan = scan_sources(selection, max_read_bytes=max_read_bytes) + plan = MigrationPlan( + selection=selection, + max_read_bytes=max_read_bytes, + max_skill_bytes=max_skill_bytes, + report_only=list(scan.report_only), + excluded=list(scan.excluded), + problems=list(scan.problems), + source_files=copy.deepcopy(scan.source_files), + ) + translated: list[TranslatedSkill] = [] + for source in scan.skills: + try: + translated.append(_translate_skill(source, max_read_bytes, max_skill_bytes, plan.excluded)) + except (OSError, ValueError, UnicodeDecodeError) as exc: + plan.problems.append(f"skill {source.name}: REFUSED ({exc})") + by_name: dict[str, list[TranslatedSkill]] = {} + for item in translated: + by_name.setdefault(item.name, []).append(item) + for name, items in sorted(by_name.items()): + if len(items) > 1: + sources = ", ".join(f"{item.source_class}:{item.source}" for item in items) + plan.problems.append(f"skill name collision {name!r}: {sources}") + continue + plan.skills.append(items[0]) + plan.knowledge = _knowledge(scan) + plan.mcp_servers, plan.required_values, mcp_report = _sanitize_mcp(scan.mcp_servers) + plan.report_only = sorted(set(plan.report_only + mcp_report)) + return plan + + +def plan_fingerprint(plan: MigrationPlan) -> str: + payload = json.dumps(plan.as_dict(), sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode("utf-8") + return hashlib.sha256(payload).hexdigest() diff --git a/protoagent.plugin.yaml b/protoagent.plugin.yaml index cd79518..8aef899 100644 --- a/protoagent.plugin.yaml +++ b/protoagent.plugin.yaml @@ -1,9 +1,9 @@ id: codex_bridge name: Codex Bridge -version: 0.0.1 +version: 0.1.0 description: >- - Foundation for preview-first migration of selected operator-owned Codex - skills, instructions, memory, and MCP definitions into protoAgent. + Preview-first migration of selected operator-owned Codex skills, + instructions, memory, and MCP definitions into protoAgent snapshots. enabled: false repository: https://github.com/RomeoRaven/codex-bridge-plugin homepage: https://agent.protolabs.studio diff --git a/pyproject.toml b/pyproject.toml index 707b178..5db5db3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,7 +1,7 @@ [project] name = "codex-bridge-plugin" -version = "0.0.1" -description = "protoAgent plugin foundation for reviewed Codex state migration." +version = "0.1.0" +description = "protoAgent plugin for reviewed Codex state migration and snapshot export." requires-python = ">=3.11" [tool.pytest.ini_options] diff --git a/requirements-dev.txt b/requirements-dev.txt index 70b7b05..14d0eb2 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -1,2 +1,3 @@ pytest>=8 pyyaml>=6 +langchain-core>=0.2 diff --git a/skills/codex-bridge/SKILL.md b/skills/codex-bridge/SKILL.md index 9d820b2..0d36880 100644 --- a/skills/codex-bridge/SKILL.md +++ b/skills/codex-bridge/SKILL.md @@ -5,9 +5,11 @@ description: Use when planning or reviewing Codex state migration into protoAgen # Codex Bridge -This repository currently provides the source contract and plugin foundation. +This repository implements the first write-nothing Codex inventory, translation, review, and snapshot slice. -- Select Codex user and repository roots explicitly. +- Use `codex_sources` to verify only the explicitly selected Codex and repository roots. +- Use `codex_import_scan` to review the full plan and its SHA-256; `output_format=json` returns the machine-readable review. +- Use `codex_export_snapshot` in dry-run mode first; `apply=true` writes only the reviewed archive and does not create a pA agent. - Treat current `.agents/skills` and legacy `.codex/skills` as distinct sources. - Never read credentials, sessions, SQLite state, logs, caches, or downloaded plugin code. - Keep raw memories and rollout summaries excluded by default. diff --git a/stores.py b/stores.py new file mode 100644 index 0000000..c5184d7 --- /dev/null +++ b/stores.py @@ -0,0 +1,314 @@ +"""Explicit source adapters over selected Codex user and repository state.""" + +from __future__ import annotations + +import hashlib +import tomllib +from dataclasses import dataclass, field, replace +from pathlib import Path +from typing import Any + +from .boundaries import FencedRoot + + +_SYSTEM_PARTS = {".system", ".codex", "system"} +_PLUGIN_SKILL_PARTS = {"plugins", "plugin-cache", "cache"} +_REPORT_ONLY_KEYS = { + "approval_policy", + "features", + "model", + "model_reasoning_effort", + "notice", + "personality", + "plugins", + "profiles", + "projects", + "rules", + "sandbox_mode", + "tui", +} +_SKIP_PROJECT_DIRS = {".git", ".hg", ".svn", "node_modules", ".venv", "venv", "__pycache__"} + + +@dataclass(frozen=True) +class SourceSelection: + codex_home: Path + user_skills_root: Path + legacy_skills_root: Path + project_roots: tuple[Path, ...] + include_memory_summary: bool = False + + @classmethod + def resolve( + cls, + *, + codex_home: str | Path, + user_skills_root: str | Path, + legacy_skills_root: str | Path, + project_roots: list[str | Path] | tuple[str | Path, ...], + include_memory_summary: bool = False, + ) -> "SourceSelection": + home = Path(codex_home).expanduser().resolve() + if not home.is_dir(): + raise ValueError(f"Codex home does not exist: {home}") + projects = tuple(Path(path).expanduser().resolve() for path in project_roots) + missing = [str(path) for path in projects if not path.is_dir()] + if missing: + raise ValueError(f"project root does not exist: {missing[0]}") + if len(set(projects)) != len(projects): + raise ValueError("duplicate project roots are not allowed") + return cls( + codex_home=home, + user_skills_root=Path(user_skills_root).expanduser().resolve(), + legacy_skills_root=Path(legacy_skills_root).expanduser().resolve(), + project_roots=projects, + include_memory_summary=bool(include_memory_summary), + ) + + def with_memory_summary(self, enabled: bool) -> "SourceSelection": + return replace(self, include_memory_summary=bool(enabled)) + + +@dataclass(frozen=True) +class SkillSource: + name: str + path: Path + source_class: str + relative_path: str + root: Path + + +@dataclass(frozen=True) +class InstructionSource: + project_root: Path + project_index: int + relative_path: str + precedence: int + content: str + + +@dataclass +class SourceScan: + selection: SourceSelection + skills: list[SkillSource] = field(default_factory=list) + instructions: list[InstructionSource] = field(default_factory=list) + memory_docs: dict[str, str] = field(default_factory=dict) + mcp_servers: dict[str, dict[str, Any]] = field(default_factory=dict) + report_only: list[str] = field(default_factory=list) + excluded: list[str] = field(default_factory=list) + problems: list[str] = field(default_factory=list) + source_files: dict[str, dict[str, Any]] = field(default_factory=dict) + + def render(self) -> str: + lines = ["Codex source inventory"] + lines.extend(f"skill {item.source_class}: {item.name} ({item.relative_path})" for item in self.skills) + lines.extend(f"instruction precedence={item.precedence}: {item.relative_path}" for item in self.instructions) + lines.extend(f"memory: {name}" for name in sorted(self.memory_docs)) + lines.extend(f"mcp: {name}" for name in sorted(self.mcp_servers)) + lines.extend(f"report-only: {name}" for name in sorted(self.report_only)) + lines.extend(f"excluded: {item}" for item in self.excluded) + lines.extend(f"problem: {item}" for item in self.problems) + return "\n".join(lines) + + +def _walk_skill_markdown(root: Path, fence: FencedRoot, problems: list[str]) -> list[Path]: + if not root.is_dir(): + return [] + pending = [root] + found: list[Path] = [] + while pending: + directory = pending.pop() + try: + children = sorted(directory.iterdir()) + except OSError as exc: + problems.append(f"skills/{directory.name}: REFUSED ({exc})") + continue + for child in children: + if child.is_symlink(): + try: + fence.resolve(str(child.relative_to(root))) + except (OSError, ValueError): + problems.append(f"skill {child.name}: REFUSED (outside declared skill root)") + continue + if child.is_dir(): + pending.append(child) + elif child.name == "SKILL.md": + found.append(child) + return sorted(found) + + +def _skills_from_root( + root: Path, + source_class: str, + excluded: list[str], + problems: list[str], +) -> list[SkillSource]: + if not root.is_dir(): + return [] + fence = FencedRoot(f"{source_class} skills", root) + result: list[SkillSource] = [] + for markdown in _walk_skill_markdown(root, fence, problems): + try: + rel_file = markdown.relative_to(root) + safe = fence.resolve(str(rel_file)) + except (OSError, ValueError) as exc: + problems.append(f"skill {markdown.parent.name}: REFUSED ({exc})") + continue + parts = safe.relative_to(root).parts + name = safe.parent.name + if any(part in _SYSTEM_PARTS or part.startswith(".") for part in parts[:-1]): + excluded.append(f"skill {name}: system skill") + continue + if any(part.lower() in _PLUGIN_SKILL_PARTS for part in parts[:-1]): + excluded.append(f"skill {name}: plugin-delivered/cache skill") + continue + result.append( + SkillSource( + name=name, + path=safe.parent, + source_class=source_class, + relative_path=str(safe.parent.relative_to(root)), + root=root, + ) + ) + return result + + +def _project_instructions( + project: Path, project_index: int, max_bytes: int, problems: list[str] +) -> list[InstructionSource]: + fence = FencedRoot("project", project) + found: list[InstructionSource] = [] + pending = [project] + while pending: + directory = pending.pop() + try: + children = sorted(directory.iterdir()) + except OSError as exc: + problems.append(f"project directory {directory.name}: REFUSED ({exc})") + continue + for child in children: + if child.is_symlink(): + rel = str(child.relative_to(project)) + try: + fence.resolve(rel) + except (OSError, ValueError): + if child.name == "AGENTS.md": + problems.append(f"instruction {rel}: REFUSED (outside declared project root)") + else: + problems.append(f"project symlink {rel}: REFUSED (outside declared project root)") + continue + if child.is_dir(): + if child.name not in _SKIP_PROJECT_DIRS: + pending.append(child) + continue + if child.name != "AGENTS.md": + continue + rel = str(child.relative_to(project)) + try: + text = fence.read_text(rel, max_bytes) + except (OSError, ValueError, UnicodeDecodeError) as exc: + problems.append(f"instruction {rel}: REFUSED ({exc})") + continue + found.append( + InstructionSource( + project_root=project, + project_index=project_index, + relative_path=rel, + precedence=len(Path(rel).parts) - 1, + content=text, + ) + ) + return sorted(found, key=lambda item: (item.precedence, item.relative_path)) + + +def _memory( + selection: SourceSelection, max_bytes: int, problems: list[str] +) -> tuple[dict[str, str], dict[str, dict[str, Any]]]: + fence = FencedRoot("Codex home", selection.codex_home) + names = ["MEMORY.md"] + if selection.include_memory_summary: + names.append("memory_summary.md") + result: dict[str, str] = {} + source_files: dict[str, dict[str, Any]] = {} + for name in names: + rel = f"memories/{name}" + if not (selection.codex_home / rel).is_file(): + continue + try: + text = fence.read_text(rel, max_bytes) + result[name] = text + raw = text.encode("utf-8") + source_files[f"codex-home:{rel}"] = { + "sha256": hashlib.sha256(raw).hexdigest(), + "bytes": len(raw), + } + except (OSError, ValueError, UnicodeDecodeError) as exc: + problems.append(f"memory {name}: REFUSED ({exc})") + return result, source_files + + +def _config( + selection: SourceSelection, max_bytes: int, problems: list[str] +) -> tuple[dict[str, dict[str, Any]], list[str], dict[str, dict[str, Any]]]: + path = selection.codex_home / "config.toml" + report = [] + if (selection.codex_home / "rules").is_dir(): + report.append("rules") + if not path.is_file(): + return {}, report, {} + fence = FencedRoot("Codex home", selection.codex_home) + try: + text = fence.read_text("config.toml", max_bytes) + doc = tomllib.loads(text) + except (OSError, ValueError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc: + problems.append(f"config.toml: REFUSED ({exc})") + return {}, report, {} + if not isinstance(doc, dict): + return {}, report, {} + report.extend(str(key) for key in doc if key in _REPORT_ONLY_KEYS) + raw = doc.get("mcp_servers") or {} + servers = ( + {str(name): value for name, value in raw.items() if isinstance(value, dict)} if isinstance(raw, dict) else {} + ) + raw_text = text.encode("utf-8") + source_files = { + "codex-home:config.toml": { + "sha256": hashlib.sha256(raw_text).hexdigest(), + "bytes": len(raw_text), + } + } + return servers, sorted(set(report)), source_files + + +def scan_sources(selection: SourceSelection, *, max_read_bytes: int = 65_536) -> SourceScan: + scan = SourceScan(selection=selection) + scan.skills.extend(_skills_from_root(selection.user_skills_root, "current-user", scan.excluded, scan.problems)) + scan.skills.extend(_skills_from_root(selection.legacy_skills_root, "legacy-user", scan.excluded, scan.problems)) + for index, project in enumerate(selection.project_roots): + repo_skills = project / ".agents" / "skills" + scan.skills.extend(_skills_from_root(repo_skills, "repository", scan.excluded, scan.problems)) + instructions = _project_instructions(project, index, max_read_bytes, scan.problems) + scan.instructions.extend(instructions) + for instruction in instructions: + raw = instruction.content.encode("utf-8") + scan.source_files[f"project[{instruction.project_index}]:{instruction.relative_path}"] = { + "sha256": hashlib.sha256(raw).hexdigest(), + "bytes": len(raw), + } + if repo_skills.is_dir(): + scan.report_only.append(f"project[{index}].skills") + scan.skills.sort(key=lambda item: (item.source_class, item.name, item.relative_path)) + scan.memory_docs, memory_files = _memory(selection, max_read_bytes, scan.problems) + scan.source_files.update(memory_files) + scan.mcp_servers, report, config_files = _config(selection, max_read_bytes, scan.problems) + scan.source_files.update(config_files) + scan.report_only.extend(report) + scan.report_only = sorted(set(scan.report_only)) + scan.excluded.extend( + [ + "auth/session/database/log/cache/plugin/runtime state: hard excluded", + "raw memories and rollout summaries: hard excluded", + ] + ) + return scan diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..442514b --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import importlib.util +import json +import sys +from pathlib import Path + +import pytest + + +ROOT = Path(__file__).resolve().parent.parent +PKG = "codex_bridge" + +if PKG not in sys.modules: + spec = importlib.util.spec_from_file_location(PKG, ROOT / "__init__.py", submodule_search_locations=[str(ROOT)]) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + sys.modules[PKG] = module + spec.loader.exec_module(module) + + +def write_skill(root: Path, name: str, body: str = "Do the safe thing.\n") -> Path: + skill = root / name + skill.mkdir(parents=True) + (skill / "SKILL.md").write_text( + f"---\nname: {name}\ndescription: {name} fixture.\n---\n\n{body}", + encoding="utf-8", + ) + (skill / "reference.md").write_text(f"reference for {name}\n", encoding="utf-8") + return skill + + +@pytest.fixture() +def codex_sources(tmp_path: Path) -> dict[str, Path]: + codex_home = tmp_path / ".codex" + current_skills = tmp_path / ".agents" / "skills" + legacy_skills = codex_home / "skills" + project = tmp_path / "project" + + write_skill(current_skills, "current-skill") + write_skill(legacy_skills, "legacy-skill") + write_skill(legacy_skills / ".system", "system-skill", "SYSTEM_SKILL_SECRET\n") + write_skill(project / ".agents" / "skills", "repo-skill") + + (project / "src" / "feature").mkdir(parents=True) + (project / "AGENTS.md").write_text("# Root instructions\n\nROOT_INSTRUCTION_MARKER\n", encoding="utf-8") + (project / "src" / "AGENTS.md").write_text("# Source instructions\n\nSOURCE_INSTRUCTION_MARKER\n", encoding="utf-8") + + memories = codex_home / "memories" + memories.mkdir(parents=True) + (memories / "MEMORY.md").write_text("# Curated Codex memory\n\nCURATED_MEMORY_MARKER\n \n", encoding="utf-8") + (memories / "memory_summary.md").write_text("# Summary\n\nSUMMARY_MARKER\n", encoding="utf-8") + (memories / "raw_memories.md").write_text("RAW_MEMORY_SECRET\n", encoding="utf-8") + summaries = memories / "rollout_summaries" + summaries.mkdir() + (summaries / "summary.md").write_text("ROLLOUT_SUMMARY_SECRET\n", encoding="utf-8") + + config = """ +model = "do-not-migrate" +approval_policy = "never" +sandbox_mode = "danger-full-access" +personality = "do-not-migrate" + +[projects."/private/project"] +trust_level = "trusted" + +[plugins] +enabled = true + +[mcp_servers.docs] +command = "runner --token COMMAND_SECRET" +transport = "SECRET_TRANSPORT" +startup_timeout_sec = 12345 +args = ["serve", "--token", "ARG_SECRET"] +cwd = "/private/workdir/CWD_SECRET" +env = { DOCS_TOKEN = "ENV_SECRET", REGION = "west" } +custom_secret = "UNSUPPORTED_SECRET" + +[mcp_servers.remote] +url = "https://example.test/URL_SECRET" +bearer_token_env_var = "REMOTE_TOKEN_NAME" +http_headers = { Authorization = "Bearer HEADER_SECRET" } +""".strip() + codex_home.mkdir(exist_ok=True) + (codex_home / "config.toml").write_text(config + "\n", encoding="utf-8") + (codex_home / "auth.json").write_text(json.dumps({"token": "AUTH_SECRET"}), encoding="utf-8") + (codex_home / "state_5.sqlite").write_bytes(b"SQLITE_SECRET") + (codex_home / "state_5.sqlite-wal").write_bytes(b"WAL_SECRET") + (codex_home / "sessions").mkdir() + (codex_home / "sessions" / "rollout.jsonl").write_text("SESSION_SECRET\n", encoding="utf-8") + (codex_home / "log").mkdir() + (codex_home / "log" / "codex.log").write_text("LOG_SECRET\n", encoding="utf-8") + (codex_home / "plugins" / "cache" / "fixture").mkdir(parents=True) + (codex_home / "plugins" / "cache" / "fixture" / "plugin.py").write_text("PLUGIN_CACHE_SECRET\n", encoding="utf-8") + (codex_home / "rules").mkdir() + (codex_home / "rules" / "default.rules").write_text("RULE_BODY_SECRET\n", encoding="utf-8") + + return { + "codex_home": codex_home, + "current_skills": current_skills, + "legacy_skills": legacy_skills, + "project": project, + } diff --git a/tests/test_export_bundle.py b/tests/test_export_bundle.py new file mode 100644 index 0000000..33f1246 --- /dev/null +++ b/tests/test_export_bundle.py @@ -0,0 +1,100 @@ +from __future__ import annotations + +import io +import os +import sys +import zipfile +from pathlib import Path + +import pytest +import yaml + +from codex_bridge.export_bundle import build_bundle +from codex_bridge.migration import build_migration_plan +from codex_bridge.stores import SourceSelection + + +def selection(sources: dict[str, Path]) -> SourceSelection: + return SourceSelection.resolve( + codex_home=sources["codex_home"], + user_skills_root=sources["current_skills"], + legacy_skills_root=sources["legacy_skills"], + project_roots=[sources["project"]], + ) + + +def test_secret_free_snapshot_contains_supported_vertical_slice(codex_sources: dict[str, Path]) -> None: + plan = build_migration_plan(selection(codex_sources)) + + data, receipt = build_bundle(agent_name="codex-import", migration=plan) + + assert receipt.skills == ["current-skill", "legacy-skill", "repo-skill"] + assert receipt.mcp_servers == ["docs", "remote"] + assert receipt.knowledge == {"codex-instructions": 2, "codex-memory": 1} + with zipfile.ZipFile(io.BytesIO(data)) as archive: + names = set(archive.namelist()) + manifest = yaml.safe_load(archive.read("agent.snapshot.yaml")) + all_content = b"".join(archive.read(name) for name in names) + assert "skills/codex/current-skill/SKILL.md" in names + assert "skills/codex/legacy-skill/reference.md" in names + assert "skills/codex/repo-skill/SKILL.md" in names + assert "knowledge/codex-instructions.md" in names + assert "knowledge/codex-memory.md" in names + assert "REVIEW.md" in names + assert manifest["snapshot_version"] == 1 + assert manifest["kind"] == "agent-snapshot" + assert manifest["plugins"] == [] + assert manifest["soul"] is None + assert manifest["translated_from"] == {"agent": "codex"} + assert manifest["config"]["mcp"]["servers"][0]["command"] == "" + assert manifest["required_secrets"] == plan.required_values + for secret in ( + b"COMMAND_SECRET", + b"ARG_SECRET", + b"CWD_SECRET", + b"ENV_SECRET", + b"URL_SECRET", + b"HEADER_SECRET", + b"AUTH_SECRET", + b"SQLITE_SECRET", + b"SESSION_SECRET", + b"LOG_SECRET", + b"PLUGIN_CACHE_SECRET", + b"RAW_MEMORY_SECRET", + b"ROLLOUT_SUMMARY_SECRET", + b"RULE_BODY_SECRET", + ): + assert secret not in all_content + + +def test_snapshot_refuses_problem_plan(codex_sources: dict[str, Path]) -> None: + collision = codex_sources["project"] / ".agents" / "skills" / "current-skill" + collision.mkdir() + (collision / "SKILL.md").write_text("---\nname: current-skill\n---\n", encoding="utf-8") + plan = build_migration_plan(selection(codex_sources)) + + with pytest.raises(ValueError, match="unresolved migration problems"): + build_bundle(agent_name="codex-import", migration=plan) + + +def test_snapshot_passes_current_protoagent_host_free_inspector(codex_sources: dict[str, Path]) -> None: + source = os.environ.get("PROTOAGENT_SOURCE") + if not source: + pytest.skip("set PROTOAGENT_SOURCE to run host contract validation") + source_path = Path(source).resolve() + sys.path.insert(0, str(source_path)) + try: + from graph.snapshot_import import inspect_snapshot + + data, _ = build_bundle( + agent_name="codex-import", + migration=build_migration_plan(selection(codex_sources)), + ) + inspected = inspect_snapshot(data).as_dict() + finally: + sys.path.remove(str(source_path)) + + assert inspected["agent_name"] == "codex-import" + assert inspected["plugins"] == [] + assert inspected["runs_code"] is False + assert sorted(inspected["knowledge"]) == ["codex-instructions", "codex-memory"] diff --git a/tests/test_migration.py b/tests/test_migration.py new file mode 100644 index 0000000..4b79cf3 --- /dev/null +++ b/tests/test_migration.py @@ -0,0 +1,237 @@ +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +from codex_bridge.migration import build_migration_plan, plan_fingerprint +from codex_bridge.stores import SourceSelection + + +def selection(sources: dict[str, Path], *, summary: bool = False) -> SourceSelection: + return SourceSelection.resolve( + codex_home=sources["codex_home"], + user_skills_root=sources["current_skills"], + legacy_skills_root=sources["legacy_skills"], + project_roots=[sources["project"]], + include_memory_summary=summary, + ) + + +def tree_hash(roots: list[Path]) -> str: + digest = hashlib.sha256() + for root in roots: + for path in sorted(p for p in root.rglob("*") if p.is_file() and not p.is_symlink()): + digest.update(str(path.relative_to(root)).encode()) + digest.update(b"\0") + digest.update(path.read_bytes()) + digest.update(b"\0") + return digest.hexdigest() + + +def test_plan_translates_supported_sources_without_writing_or_leaking(codex_sources: dict[str, Path]) -> None: + roots = [ + codex_sources["codex_home"], + codex_sources["current_skills"], + codex_sources["project"], + ] + before = tree_hash(roots) + + plan = build_migration_plan(selection(codex_sources)) + rendered = plan.render() + serialized = json.dumps(plan.as_dict(), sort_keys=True) + + assert tree_hash(roots) == before + assert [(skill.name, skill.source_class) for skill in plan.skills] == [ + ("current-skill", "current-user"), + ("legacy-skill", "legacy-user"), + ("repo-skill", "repository"), + ] + assert all(skill.source_files["SKILL.md"]["sha256"] for skill in plan.skills) + assert plan.skills[0].files["reference.md"] == b"reference for current-skill\n" + assert [doc.source for doc in plan.knowledge["codex-instructions"]] == [ + "AGENTS.md", + "src/AGENTS.md", + ] + assert [doc.precedence for doc in plan.knowledge["codex-instructions"]] == [0, 1] + assert plan.knowledge["codex-memory"][0].content == "# Curated Codex memory\n\nCURATED_MEMORY_MARKER\n \n" + assert "codex-memory-summary" not in plan.knowledge + assert sorted(server["name"] for server in plan.mcp_servers) == ["docs", "remote"] + assert plan.mcp_servers[0]["command"] == "" + assert plan.mcp_servers[0]["args"] == [] + assert plan.mcp_servers[0]["cwd"] == "" + assert plan.mcp_servers[0]["env"] == {"DOCS_TOKEN": "", "REGION": ""} + assert plan.mcp_servers[1]["url"] == "" + assert plan.mcp_servers[1]["bearer_token_env_var"] == "" + assert plan.mcp_servers[1]["http_headers"] == {"Authorization": ""} + required_names = sorted(item["name"] for item in plan.required_values) + assert required_names == [ + "mcp.docs.args", + "mcp.docs.command", + "mcp.docs.cwd", + "mcp.docs.env.DOCS_TOKEN", + "mcp.docs.env.REGION", + "mcp.docs.startup_timeout_sec", + "mcp.docs.transport", + "mcp.remote.bearer_token_env_var", + "mcp.remote.http_headers.Authorization", + "mcp.remote.url", + ] + assert set(plan.report_only) >= { + "approval_policy", + "model", + "personality", + "plugins", + "projects", + "rules", + "sandbox_mode", + } + assert "mcp.docs.unsupported.custom_secret" in plan.report_only + assert len(plan_fingerprint(plan)) == 64 + for secret in ( + "COMMAND_SECRET", + "ARG_SECRET", + "CWD_SECRET", + "ENV_SECRET", + "URL_SECRET", + "HEADER_SECRET", + "AUTH_SECRET", + "SQLITE_SECRET", + "SESSION_SECRET", + "LOG_SECRET", + "PLUGIN_CACHE_SECRET", + "RAW_MEMORY_SECRET", + "ROLLOUT_SUMMARY_SECRET", + "RULE_BODY_SECRET", + "UNSUPPORTED_SECRET", + "SECRET_TRANSPORT", + "12345", + "/private/project", + ): + assert secret not in rendered + assert secret not in serialized + + +def test_memory_summary_is_separate_opt_in_domain(codex_sources: dict[str, Path]) -> None: + plan = build_migration_plan(selection(codex_sources, summary=True)) + + assert plan.knowledge["codex-memory-summary"][0].source == "memories/memory_summary.md" + assert plan.knowledge["codex-memory-summary"][0].content == "# Summary\n\nSUMMARY_MARKER\n" + + +def test_colliding_skill_names_are_omitted_and_reported(codex_sources: dict[str, Path]) -> None: + collision = codex_sources["project"] / ".agents" / "skills" / "current-skill" + collision.mkdir() + (collision / "SKILL.md").write_text( + "---\nname: current-skill\ndescription: collision\n---\n\nCollision.\n", + encoding="utf-8", + ) + + plan = build_migration_plan(selection(codex_sources)) + + assert "current-skill" not in [skill.name for skill in plan.skills] + assert any("skill name collision 'current-skill'" in problem for problem in plan.problems) + + +def test_skill_caps_refuse_whole_artifacts(codex_sources: dict[str, Path]) -> None: + skill = codex_sources["current_skills"] / "current-skill" + (skill / "SKILL.md").write_bytes(b"x" * 129) + + plan = build_migration_plan(selection(codex_sources), max_skill_bytes=128) + + assert "current-skill" not in [item.name for item in plan.skills] + assert any("SKILL.md exceeds max_skill_bytes=128" in problem for problem in plan.problems) + + +def test_fingerprint_changes_when_skill_or_instruction_source_changes(codex_sources: dict[str, Path]) -> None: + selected = selection(codex_sources) + baseline = plan_fingerprint(build_migration_plan(selected)) + reference = codex_sources["current_skills"] / "current-skill" / "reference.md" + reference.write_text("changed reference\n", encoding="utf-8") + changed_skill = plan_fingerprint(build_migration_plan(selected)) + instruction = codex_sources["project"] / "src" / "AGENTS.md" + instruction.write_text("# Changed\n", encoding="utf-8") + changed_instruction = plan_fingerprint(build_migration_plan(selected)) + + config = codex_sources["codex_home"] / "config.toml" + config.write_text( + config.read_text(encoding="utf-8").replace("COMMAND_SECRET", "DIFFERENT_COMMAND_SECRET"), + encoding="utf-8", + ) + changed_config = plan_fingerprint(build_migration_plan(selected)) + + assert len({baseline, changed_skill, changed_instruction, changed_config}) == 4 + + +def test_hard_excluded_paths_are_never_opened(codex_sources: dict[str, Path], monkeypatch) -> None: + opened: list[Path] = [] + original_open = Path.open + + def recording_open(path: Path, *args, **kwargs): + opened.append(path) + return original_open(path, *args, **kwargs) + + monkeypatch.setattr(Path, "open", recording_open) + + build_migration_plan(selection(codex_sources)) + + opened_text = "\n".join(str(path) for path in opened) + for excluded in ( + "auth.json", + "state_5.sqlite", + "state_5.sqlite-wal", + "rollout.jsonl", + "codex.log", + "plugin.py", + "raw_memories.md", + "rollout_summaries", + "default.rules", + ): + assert excluded not in opened_text + + +def test_hard_excluded_supporting_files_are_not_opened_or_exported(codex_sources: dict[str, Path], monkeypatch) -> None: + skill = codex_sources["current_skills"] / "current-skill" + (skill / "auth.json").write_text("SUPPORT_AUTH_SECRET\n", encoding="utf-8") + (skill / "history.jsonl").write_text("SUPPORT_HISTORY_SECRET\n", encoding="utf-8") + (skill / "state.sqlite").write_bytes(b"SUPPORT_SQLITE_SECRET") + cache = skill / "cache" / "plugin" + cache.mkdir(parents=True) + (cache / "plugin.py").write_text("SUPPORT_PLUGIN_SECRET\n", encoding="utf-8") + opened: list[Path] = [] + original_open = Path.open + + def recording_open(path: Path, *args, **kwargs): + opened.append(path) + return original_open(path, *args, **kwargs) + + monkeypatch.setattr(Path, "open", recording_open) + plan = build_migration_plan(selection(codex_sources)) + migrated = next(item for item in plan.skills if item.name == "current-skill") + output = json.dumps(plan.as_dict(), sort_keys=True) + + assert sorted(migrated.files) == ["SKILL.md", "reference.md"] + assert any("current-skill/auth.json: hard excluded" in item for item in plan.excluded) + for marker in ("SUPPORT_AUTH_SECRET", "SUPPORT_HISTORY_SECRET", "SUPPORT_SQLITE_SECRET", "SUPPORT_PLUGIN_SECRET"): + assert marker not in output + opened_text = "\n".join(str(path) for path in opened) + assert "auth.json" not in opened_text + assert "history.jsonl" not in opened_text + assert "state.sqlite" not in opened_text + assert "cache/plugin/plugin.py" not in opened_text + + +def test_plugin_delivered_skills_are_excluded(codex_sources: dict[str, Path]) -> None: + delivered = codex_sources["current_skills"] / "plugins" / "cache" / "delivered" + delivered.mkdir(parents=True) + (delivered / "SKILL.md").write_text( + "---\nname: delivered\ndescription: plugin delivered\n---\n\nPLUGIN_DELIVERED_SECRET\n", + encoding="utf-8", + ) + + plan = build_migration_plan(selection(codex_sources)) + output = plan.render() + json.dumps(plan.as_dict()) + + assert "delivered" not in [item.name for item in plan.skills] + assert "skill delivered: plugin-delivered/cache skill" in output + assert "PLUGIN_DELIVERED_SECRET" not in output diff --git a/tests/test_scaffold.py b/tests/test_scaffold.py index aef5d89..b2f95eb 100644 --- a/tests/test_scaffold.py +++ b/tests/test_scaffold.py @@ -26,8 +26,8 @@ def test_manifest_is_disabled_scoped_and_versioned() -> None: pyproject = (ROOT / "pyproject.toml").read_text(encoding="utf-8") assert manifest["id"] == "codex_bridge" - assert manifest["version"] == "0.0.1" - assert 'version = "0.0.1"' in pyproject + assert manifest["version"] == "0.1.0" + assert 'version = "0.1.0"' in pyproject assert manifest["enabled"] is False assert manifest["min_protoagent_version"] == "0.125.0" assert manifest["capabilities"] == {"network": [], "filesystem": "scoped"} @@ -37,7 +37,7 @@ def test_manifest_is_disabled_scoped_and_versioned() -> None: assert manifest["config"]["max_skill_bytes"] == 262144 -def test_foundation_registers_guidance_but_no_migration_tools() -> None: +def test_plugin_registers_guidance_and_snapshot_tools() -> None: spec = importlib.util.spec_from_file_location("codex_bridge", ROOT / "__init__.py") assert spec and spec.loader module = importlib.util.module_from_spec(spec) @@ -47,14 +47,18 @@ def test_foundation_registers_guidance_but_no_migration_tools() -> None: module.register(registry) assert registry.skill_dirs == ["skills"] - assert registry.tools == [] + assert [item.name for item in registry.tools] == [ + "codex_sources", + "codex_import_scan", + "codex_export_snapshot", + ] def test_public_repository_contract_is_present() -> None: readme = (ROOT / "README.md").read_text(encoding="utf-8") plan = (ROOT / "docs" / "IMPLEMENTATION_PLAN.md").read_text(encoding="utf-8") - assert "Migration tools are not implemented yet" in readme + assert "snapshot migration candidate" in readme assert "~/.agents/skills" in readme assert "~/.codex/skills" in readme assert "No active primary/development protoAgent runtime" in plan diff --git a/tests/test_stores.py b/tests/test_stores.py new file mode 100644 index 0000000..5e93047 --- /dev/null +++ b/tests/test_stores.py @@ -0,0 +1,98 @@ +from __future__ import annotations + +from pathlib import Path + +import pytest + +from codex_bridge.stores import SourceSelection, scan_sources + + +def selection(sources: dict[str, Path]) -> SourceSelection: + return SourceSelection.resolve( + codex_home=sources["codex_home"], + user_skills_root=sources["current_skills"], + legacy_skills_root=sources["legacy_skills"], + project_roots=[sources["project"]], + ) + + +def test_scan_classifies_explicit_sources_without_excluded_state(codex_sources: dict[str, Path]) -> None: + scan = scan_sources(selection(codex_sources)) + rendered = scan.render() + + assert [(item.name, item.source_class) for item in scan.skills] == [ + ("current-skill", "current-user"), + ("legacy-skill", "legacy-user"), + ("repo-skill", "repository"), + ] + assert "system-skill" in rendered and "system skill" in rendered + assert [item.relative_path for item in scan.instructions] == ["AGENTS.md", "src/AGENTS.md"] + assert scan.memory_docs == {"MEMORY.md": "# Curated Codex memory\n\nCURATED_MEMORY_MARKER\n \n"} + assert sorted(scan.mcp_servers) == ["docs", "remote"] + assert set(scan.report_only) >= { + "approval_policy", + "model", + "personality", + "plugins", + "projects", + "rules", + "sandbox_mode", + } + for secret in ( + "AUTH_SECRET", + "SQLITE_SECRET", + "WAL_SECRET", + "SESSION_SECRET", + "LOG_SECRET", + "PLUGIN_CACHE_SECRET", + "RAW_MEMORY_SECRET", + "ROLLOUT_SUMMARY_SECRET", + "RULE_BODY_SECRET", + ): + assert secret not in rendered + + +def test_memory_summary_is_opt_in(codex_sources: dict[str, Path]) -> None: + base = selection(codex_sources) + + without = scan_sources(base) + with_summary = scan_sources(base.with_memory_summary(True)) + + assert "memory_summary.md" not in without.memory_docs + assert with_summary.memory_docs["memory_summary.md"] == "# Summary\n\nSUMMARY_MARKER\n" + + +def test_project_root_escape_and_symlinked_skill_escape_are_refused( + codex_sources: dict[str, Path], tmp_path: Path +) -> None: + outside = tmp_path / "outside" + outside.mkdir() + (outside / "SKILL.md").write_text("---\nname: escaped\n---\n\nOUTSIDE_SECRET\n", encoding="utf-8") + (codex_sources["current_skills"] / "escaped").symlink_to(outside, target_is_directory=True) + + scan = scan_sources(selection(codex_sources)) + rendered = scan.render() + + assert "escaped" in rendered and "REFUSED" in rendered + assert "OUTSIDE_SECRET" not in rendered + with pytest.raises(ValueError): + SourceSelection.resolve( + codex_home=codex_sources["codex_home"], + user_skills_root=codex_sources["current_skills"], + legacy_skills_root=codex_sources["legacy_skills"], + project_roots=[codex_sources["project"] / "missing"], + ) + + +def test_repository_agents_symlink_escape_is_refused_and_named(codex_sources: dict[str, Path], tmp_path: Path) -> None: + outside = tmp_path / "outside-agents.md" + outside.write_text("OUTSIDE_INSTRUCTION_SECRET\n", encoding="utf-8") + link = codex_sources["project"] / "linked" / "AGENTS.md" + link.parent.mkdir() + link.symlink_to(outside) + + scan = scan_sources(selection(codex_sources)) + rendered = scan.render() + + assert "instruction linked/AGENTS.md: REFUSED (outside declared project root)" in rendered + assert "OUTSIDE_INSTRUCTION_SECRET" not in rendered diff --git a/tests/test_tools.py b/tests/test_tools.py new file mode 100644 index 0000000..46df046 --- /dev/null +++ b/tests/test_tools.py @@ -0,0 +1,77 @@ +from __future__ import annotations + +import json +from pathlib import Path + +from codex_bridge.tools_import import build_tools + + +def roots(sources: dict[str, Path]) -> dict: + return { + "codex_home": str(sources["codex_home"]), + "user_skills_root": str(sources["current_skills"]), + "legacy_skills_root": str(sources["legacy_skills"]), + "project_roots": [str(sources["project"])], + } + + +def test_tools_list_scan_json_and_export_dry_run(codex_sources: dict[str, Path], tmp_path: Path) -> None: + tools = {item.name: item for item in build_tools({})} + kwargs = roots(codex_sources) + + sources = tools["codex_sources"].invoke(kwargs) + scan = tools["codex_import_scan"].invoke(kwargs) + review = tools["codex_import_scan"].invoke({**kwargs, "output_format": "json"}) + destination = tmp_path / "codex.zip" + export = tools["codex_export_snapshot"].invoke({**kwargs, "agent_name": "codex-import", "out": str(destination)}) + + assert "codex_home: available" in sources + assert "project[0]: available" in sources + assert "current-skill" in scan and "plan sha256:" in scan + assert "required value: mcp.docs.command" in scan + assert json.loads(review)["plan_sha256"] + assert "snapshot bytes:" in export and "Dry run" in export + assert not destination.exists() + + +def test_export_writes_only_when_explicitly_requested(codex_sources: dict[str, Path], tmp_path: Path) -> None: + tools = {item.name: item for item in build_tools({})} + destination = tmp_path / "codex.zip" + + result = tools["codex_export_snapshot"].invoke( + { + **roots(codex_sources), + "agent_name": "codex-import", + "out": str(destination), + "apply": True, + } + ) + + assert destination.is_file() + assert "Wrote" in result + + +def test_plugin_registers_three_tools_and_guidance() -> None: + import codex_bridge as module + + class Registry: + def __init__(self): + self.skill_dirs = [] + self.tools = [] + self.config = {} + + def register_skill_dir(self, path): + self.skill_dirs.append(path) + + def register_tools(self, tools): + self.tools.extend(tools) + + registry = Registry() + module.register(registry) + + assert registry.skill_dirs == ["skills"] + assert [item.name for item in registry.tools] == [ + "codex_sources", + "codex_import_scan", + "codex_export_snapshot", + ] diff --git a/tools_import.py b/tools_import.py new file mode 100644 index 0000000..3afe5d7 --- /dev/null +++ b/tools_import.py @@ -0,0 +1,140 @@ +"""Preview and snapshot-export tools for Codex Bridge.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +from langchain_core.tools import tool + +from .export_bundle import build_bundle +from .migration import build_migration_plan +from .stores import SourceSelection + + +def build_tools(cfg: dict) -> list: + defaults = dict(cfg or {}) + + def _selection( + codex_home: str, + user_skills_root: str, + legacy_skills_root: str, + project_roots: list[str], + include_memory_summary: bool, + ) -> SourceSelection: + return SourceSelection.resolve( + codex_home=codex_home or defaults.get("codex_home") or "~/.codex", + user_skills_root=user_skills_root or defaults.get("user_skills_root") or "~/.agents/skills", + legacy_skills_root=legacy_skills_root or defaults.get("legacy_skills_root") or "~/.codex/skills", + project_roots=project_roots or defaults.get("project_roots") or [], + include_memory_summary=bool(include_memory_summary or defaults.get("include_memory_summary", False)), + ) + + def _plan(selection: SourceSelection, max_read_bytes: int, max_skill_bytes: int): + return build_migration_plan( + selection, + max_read_bytes=int(max_read_bytes or defaults.get("max_read_bytes") or 65_536), + max_skill_bytes=int(max_skill_bytes or defaults.get("max_skill_bytes") or 262_144), + ) + + @tool + def codex_sources( + codex_home: str = "", + user_skills_root: str = "", + legacy_skills_root: str = "", + project_roots: list[str] = [], + ) -> str: + """Report availability of explicit Codex user and repository source roots.""" + rows = [ + ("codex_home", codex_home or defaults.get("codex_home") or "~/.codex"), + ( + "user_skills_root", + user_skills_root or defaults.get("user_skills_root") or "~/.agents/skills", + ), + ( + "legacy_skills_root", + legacy_skills_root or defaults.get("legacy_skills_root") or "~/.codex/skills", + ), + ] + projects = project_roots or defaults.get("project_roots") or [] + rows.extend((f"project[{index}]", value) for index, value in enumerate(projects)) + return "Codex source roots:\n" + "\n".join( + f"- {name}: {'available' if Path(value).expanduser().is_dir() else 'missing'}" for name, value in rows + ) + + @tool + def codex_import_scan( + codex_home: str = "", + user_skills_root: str = "", + legacy_skills_root: str = "", + project_roots: list[str] = [], + include_memory_summary: bool = False, + max_read_bytes: int = 0, + max_skill_bytes: int = 0, + output_format: str = "text", + ) -> str: + """Build a decision-grade write-nothing Codex migration plan.""" + try: + selection = _selection( + codex_home, + user_skills_root, + legacy_skills_root, + project_roots, + include_memory_summary, + ) + plan = _plan(selection, max_read_bytes, max_skill_bytes) + if str(output_format).lower() == "json": + return json.dumps(plan.review_dict(), sort_keys=True) + return plan.render() + except (OSError, ValueError) as exc: + return f"error: {exc}" + + @tool + def codex_export_snapshot( + codex_home: str = "", + user_skills_root: str = "", + legacy_skills_root: str = "", + project_roots: list[str] = [], + include_memory_summary: bool = False, + agent_name: str = "codex-import", + out: str = "", + apply: bool = False, + max_read_bytes: int = 0, + max_skill_bytes: int = 0, + ) -> str: + """Preview or write a secret-free protoAgent snapshot from explicit Codex sources.""" + try: + selection = _selection( + codex_home, + user_skills_root, + legacy_skills_root, + project_roots, + include_memory_summary, + ) + plan = _plan(selection, max_read_bytes, max_skill_bytes) + data, receipt = build_bundle(agent_name=agent_name, migration=plan) + except (OSError, ValueError) as exc: + return f"error: {exc}" + report = ( + f"{plan.render()}\n" + f"snapshot bytes: {len(data)}\n" + f"snapshot skills: {len(receipt.skills)}; MCP: {len(receipt.mcp_servers)}; " + f"knowledge domains: {len(receipt.knowledge)}" + ) + if not apply: + return report + "\nDry run: nothing written. Re-run with apply=True after review." + destination = Path(out).expanduser() if str(out or "").strip() else Path.cwd() / f"{agent_name}.zip" + if destination.exists() and destination.is_dir(): + destination = destination / f"{agent_name}.zip" + destination.parent.mkdir(parents=True, exist_ok=True) + temporary = destination.with_name(f".{destination.name}.tmp-{os.getpid()}") + try: + temporary.write_bytes(data) + os.replace(temporary, destination) + finally: + if temporary.exists(): + temporary.unlink() + return report + f"\nWrote {destination}" + + return [codex_sources, codex_import_scan, codex_export_snapshot]