From 49752daa7c3c60f0940eb92051b7378a6a75654f Mon Sep 17 00:00:00 2001 From: Evandro Date: Fri, 25 Sep 2026 14:21:59 -0300 Subject: [PATCH] feat(adapters): package OMP and confine autonomous runners --- adapters/omp/build.py | 168 +++++++++++++ qa/test_adapters.py | 416 ++++++++++++++++++++++++++++++++ qa/test_omp_adapter.py | 45 ++++ scripts/harness_runner.py | 496 ++++++++++++++++++++++++++++++++++++++ scripts/install.py | 78 +++++- 5 files changed, 1191 insertions(+), 12 deletions(-) create mode 100644 adapters/omp/build.py create mode 100755 qa/test_adapters.py create mode 100644 qa/test_omp_adapter.py create mode 100755 scripts/harness_runner.py diff --git a/adapters/omp/build.py b/adapters/omp/build.py new file mode 100644 index 0000000..06a656f --- /dev/null +++ b/adapters/omp/build.py @@ -0,0 +1,168 @@ +"""Build the native OMP package from the canonical dev-workflows checkout.""" +from __future__ import annotations + +import hashlib +import json +import os +import re +import shutil +import tempfile +from pathlib import Path + + +OWNER_MARKER = ".dev-workflows-omp-package" +def _owner_value(src: Path) -> str: + return f"dev-workflows\nsource_root={src}\n" + + +def _content_digest(package: Path) -> str: + h = hashlib.sha256() + for path in sorted(package.rglob("*")): + if path.is_symlink(): + raise FileExistsError(f"symlink em pacote gerado: {path}") + if not path.is_file() or path.name == OWNER_MARKER: + continue + h.update(path.relative_to(package).as_posix().encode("utf-8")) + h.update(b"\0") + h.update(str(path.stat().st_mode & 0o777).encode("ascii")) + with path.open("rb") as src: + for chunk in iter(lambda: src.read(1024 * 1024), b""): + h.update(chunk) + return h.hexdigest() + +AGENT_NAMES = ( + "explorer", "implementer", "planner", "reviewer", + "security-auditor", "ui-critic", "verifier", +) +MODEL_ID = re.compile(r"^[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+(?::(?:none|minimal|low|medium|high|xhigh|max))?$") +TOOL_NAMES = { + "Read": "read", "Edit": "edit", "Write": "write", "Bash": "bash", + "Grep": "grep", "Glob": "glob", "WebFetch": "web_search", +} + + +def _agent(src: Path, name: str, models: dict[str, str], package_root: Path) -> str: + text = (src / "agents" / f"{name}.md").read_text(encoding="utf-8") + match = re.match(r"\A---\s*\n(.*?)\n---\s*\n(.*)\Z", text, re.S) + if not match: + raise ValueError(f"frontmatter inválido em agents/{name}.md") + raw, prompt = match.groups() + meta: dict[str, str] = {} + for line in raw.splitlines(): + key, sep, value = line.partition(":") + if sep: + meta[key.strip()] = value.strip() + if meta.get("name") != name or not meta.get("description"): + raise ValueError(f"metadados incompletos em agents/{name}.md") + + source_tools = [item.strip() for item in meta.get("tools", "").split(",") if item.strip()] + # An explicit OMP allowlist prevents read-only agents from inheriting write tools. + tools = list(dict.fromkeys(TOOL_NAMES[item] for item in source_tools if item in TOOL_NAMES)) + if not tools: + raise ValueError(f"nenhuma ferramenta OMP reconhecida em agents/{name}.md") + lines = ["---", f"name: {name}", f"description: {meta['description']}", "tools: " + ", ".join(tools)] + skills = re.search(r"^skills:\s*\[(.*?)\]\s*$", raw, re.M) + if skills: + values = [value.strip().strip("'\"") for value in skills.group(1).split(",") if value.strip()] + if values: + lines.append("autoloadSkills:") + lines.extend(f" - {value}" for value in values) + model = models.get(name) + if model is not None: + if not isinstance(model, str) or not MODEL_ID.fullmatch(model): + raise ValueError(f"ID de modelo inválido para {name}: {model!r}") + lines.append(f"model: {model}") + body = prompt.replace("${CLAUDE_PLUGIN_ROOT}", str(package_root)) + return "\n".join(lines + ["---", "", body]) + + +def build_package(src_root, dest, models=None): + """Atomically build an owned OMP plugin package at *dest*. + + ``models`` maps canonical agent names (roles) to verified concrete + ``provider/model[:thinking]`` selectors. Missing roles intentionally omit + ``model`` so OMP inherits the parent model. + """ + src = Path(src_root).resolve() + target = Path(dest).expanduser().absolute() + roles = {} if models is None else dict(models) + unknown = set(roles) - set(AGENT_NAMES) + if unknown: + raise ValueError("papéis desconhecidos: " + ", ".join(sorted(unknown))) + if not (src / "agents").is_dir() or not (src / "skills").is_dir(): + raise ValueError(f"checkout canônico inválido: {src}") + if target.is_symlink(): + raise FileExistsError(f"destino é symlink; recusando substituir: {target}") + if target.exists(): + marker = target / OWNER_MARKER + if not target.is_dir() or not marker.is_file(): + raise FileExistsError(f"destino existente não pertence a dev-workflows: {target}") + if marker.read_text(encoding="utf-8") != _owner_value(src) + f"sha256={_content_digest(target)}\n": + raise FileExistsError(f"pacote existente tem alterações manuais ou origem diferente: {target}") + + target.parent.mkdir(parents=True, exist_ok=True) + stage = Path(tempfile.mkdtemp(prefix=f".{target.name}.stage-", dir=target.parent)) + backup = None + try: + skills_src = src / "skills" + skills_dest = stage / "skills" + shutil.copytree(skills_src, skills_dest, symlinks=False, + ignore=shutil.ignore_patterns(".git", "__pycache__", "*.pyc")) + skill_count = sum(1 for path in skills_dest.glob("*/SKILL.md") if path.is_file()) + for path in skills_dest.rglob("*"): + if not path.is_file(): + continue + try: + content = path.read_text(encoding="utf-8") + except UnicodeError: + continue + if "${CLAUDE_PLUGIN_ROOT}" in content: + path.write_text(content.replace("${CLAUDE_PLUGIN_ROOT}", str(target)), encoding="utf-8") + if skill_count != 21: + raise ValueError(f"esperadas 21 skills canônicas; encontradas {skill_count}") + + agents_dest = stage / "agents" + agents_dest.mkdir() + for name in AGENT_NAMES: + if not (src / "agents" / f"{name}.md").is_file(): + raise ValueError(f"agente canônico ausente: {name}") + (agents_dest / f"{name}.md").write_text(_agent(src, name, roles, target), encoding="utf-8") + + # The quality-gate executable is a package asset; Claude hooks are deliberately + # excluded. The loop/controller remains responsible for enforcing its verdict. + gate = src / "scripts" / "quality-gate" + if gate.is_file(): + (stage / "scripts").mkdir() + shutil.copy2(gate, stage / "scripts" / "quality-gate") + manifest = { + "name": "dev-workflows", + "version": "1.0.0", + "description": "Skills e agentes nativos para oh-my-pi; gate imposto pelo controlador externo.", + "omp": {"extensions": []}, + } + (stage / "package.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + (stage / OWNER_MARKER).write_text(_owner_value(src) + f"sha256={_content_digest(stage)}\n", encoding="utf-8") + + if target.exists(): + backup = Path(tempfile.mkdtemp(prefix=f".{target.name}.backup-", dir=target.parent)) + backup.rmdir() + os.replace(target, backup) + try: + os.replace(stage, target) + except BaseException: + if backup is not None: + os.replace(backup, target) + backup = None + raise + if backup is not None: + shutil.rmtree(backup) + backup = None + return target + finally: + if stage.exists(): + shutil.rmtree(stage) + if backup is not None and backup.exists(): + if not target.exists(): + os.replace(backup, target) + else: + shutil.rmtree(backup) diff --git a/qa/test_adapters.py b/qa/test_adapters.py new file mode 100755 index 0000000..37f700a --- /dev/null +++ b/qa/test_adapters.py @@ -0,0 +1,416 @@ +#!/usr/bin/env python3 +"""Adapter and headless-runner behavior tests; fake CLIs never contact providers.""" +import importlib.util +import json +import os +import pathlib +import sys +import shutil +import unittest +import tempfile +from unittest import mock + +ROOT = pathlib.Path(__file__).resolve().parents[1] +REAL_WHICH = shutil.which + + +def load_module(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +installer = load_module("dev_workflows_install", ROOT / "scripts" / "install.py") +harness = load_module("dev_workflows_harness", ROOT / "scripts" / "harness_runner.py") + + +class AdapterTest(unittest.TestCase): + def test_omp_install_preserves_unowned_destination_and_skips_link(self): + with tempfile.TemporaryDirectory() as temp: + base = pathlib.Path(temp) / ".omp" + dest = base / "plugins" / "dev-workflows" + dest.mkdir(parents=True) + (dest / "user-data").write_text("preserve", encoding="utf-8") + runner = mock.Mock() + with self.assertRaises(SystemExit): + installer.install_omp(base=str(base), runner=runner) + runner.assert_not_called() + self.assertEqual((dest / "user-data").read_text(encoding="utf-8"), "preserve") + + def test_omp_install_keeps_package_for_recovery_if_link_fails(self): + with tempfile.TemporaryDirectory() as temp: + base = pathlib.Path(temp) / ".omp" + runner = mock.Mock(return_value=mock.Mock(returncode=7)) + with self.assertRaisesRegex(SystemExit, "falhou com código 7"): + installer.install_omp(base=str(base), runner=runner) + self.assertTrue((base / "plugins" / "dev-workflows" / ".dev-workflows-omp-package").is_file()) + + + +class HarnessRunnerTest(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.root = pathlib.Path(self.temp.name) + self.cwd = self.root / "state" / "project" / "run" / "worktrees" / "iteration-001" + self.cwd.mkdir(parents=True) + gitdir = self.root / "project" / ".git" / "worktrees" / "iteration-001" + gitdir.mkdir(parents=True) + (self.cwd / ".git").write_text(f"gitdir: {gitdir}\n", encoding="utf-8") + self.bin = self.root / "bin" + self.bin.mkdir() + self.trace = self.root / "calls.jsonl" + self.cache = self.root / "catalog" + self.cache.mkdir() + self._fake_cli() + + def tearDown(self): + self.temp.cleanup() + + def _fake_cli(self): + executable = self.bin / "fake-cli" + executable.write_text( + "#!" + sys.executable + "\n" + "import json, os, pathlib, sys\n" + "name = pathlib.Path(sys.argv[0]).name\n" + "args = sys.argv[1:]\n" + "with open(os.environ['DW_TRACE'], 'a', encoding='utf-8') as out:\n" + " out.write(json.dumps({'name': name, 'args': args, 'cwd': os.getcwd(), 'marker': os.getenv('DW_MARKER')}) + '\\n')\n" + "if name == 'claude' and args[:3] == ['auth', 'status', '--json']:\n" + " print(os.environ.get('DW_CLAUDE_AUTH', '{\\\"loggedIn\\\": true}'))\n" + "elif name == 'opencode' and args[:1] == ['models']:\n" + " print(os.environ.get('DW_MODELS', ''))\n" + "elif name == 'omp' and args[:2] == ['models', args[1] if len(args) > 1 else '']:\n" + " print(os.environ.get('DW_MODELS_JSON', '[]'))\n" + "else:\n" + " print('fake response')\n" + " print('sk-secret-must-not-escape', file=sys.stderr)\n", + encoding="utf-8", + ) + executable.chmod(0o755) + for name in ("claude", "opencode", "omp"): + (self.bin / name).symlink_to(executable) + + def _env(self, **extra): + value = os.environ.copy() + value.update({ + "PATH": str(self.bin) + os.pathsep + value.get("PATH", ""), + "DEV_WORKFLOWS_MODEL_CACHE": str(self.cache), + "DW_TRACE": str(self.trace), + "DW_MARKER": "test-env", + }) + value.update(extra) + return value + + def _snapshot(self, tool, model_key, *, entitled=True, variant=None, provider=None, model_id=None): + row = { + "tool": tool, + "model_key": model_key, + "availability": { + "local_listed": True, + "authenticated": True, + "entitled": entitled, + "source": "manual_audited", + "verified_at": "2026-09-25T12:00:00Z", + }, + } + if variant is not None: + row.update({"variant": variant, "provider": provider, "model_id": model_id}) + (self.cache / f"catalog-{tool}.json").write_text( + json.dumps({"schema_version": 1, "tool": tool, "models": [row]}), encoding="utf-8" + ) + + def _calls(self): + if not self.trace.exists(): + return [] + return [json.loads(line) for line in self.trace.read_text(encoding="utf-8").splitlines()] + + def _fake_bwrap(self): + trace = self.root / "bwrap.jsonl" + executable = self.bin / "bwrap" + executable.write_text( + "#!" + sys.executable + "\n" + "import json, os, pathlib, sys\n" + "args = sys.argv[1:]\n" + "with open(os.environ['DW_BWRAP_TRACE'], 'a', encoding='utf-8') as out:\n" + " out.write(json.dumps(args) + '\\n')\n" + "command = args[args.index('--') + 1:]\n" + "tool = pathlib.Path(command[0]).name\n" + "if tool == 'omp' and 'models' in command:\n" + " print(os.environ.get('DW_MODELS_JSON', '[]'))\n" + "elif tool == 'opencode' and 'models' in command:\n" + " print(os.environ.get('DW_MODELS', ''))\n" + "elif tool == 'claude' and command[1:3] == ['auth', 'status']:\n" + " print(os.environ.get('DW_CLAUDE_AUTH', '{\\\"loggedIn\\\": true}'))\n" + "else:\n" + " print('fake wrapped response')\n", + encoding="utf-8", + ) + executable.chmod(0o755) + return trace + + def _invoke_with_fake_bwrap(self, *args, **kwargs): + def find_executable(name, path=None): + if name == "bwrap": + return str(self.bin / "bwrap") + return REAL_WHICH(name, path=path) + + with mock.patch.object(harness.shutil, "which", side_effect=find_executable): + return harness.invoke(*args, **kwargs) + + def test_claude_alias_and_dontask_preserve_default_permission_gate(self): + result = harness.invoke("claude", "reviewer", "Review safely", str(self.cwd), "opus", env=self._env()) + self.assertEqual(result["returncode"], 0) + calls = self._calls() + self.assertEqual(calls[0]["args"], ["auth", "status", "--json"]) + call = calls[1] + args = call["args"] + self.assertIn("--permission-mode", args) + self.assertEqual(args[args.index("--permission-mode") + 1], "dontAsk") + self.assertIn("--agent", args) + self.assertIn("reviewer", args) + self.assertEqual(args[args.index("--model") + 1], "opus") + self.assertNotIn("--dangerously-skip-permissions", args) + self.assertEqual(call["cwd"], str(self.cwd)) + self.assertEqual(call["marker"], "test-env") + self.assertIsNone(result["usage"]) + self.assertNotIn("sk-secret", result["output"]) + + def test_claude_inherit_keeps_role_alias_without_model_override(self): + result = harness.invoke("claude", "explorer", "Map the project", str(self.cwd), "inherit", env=self._env()) + self.assertEqual(result["returncode"], 0) + calls = self._calls() + self.assertEqual(calls[0]["args"], ["auth", "status", "--json"]) + args = calls[1]["args"] + self.assertIn("--agent", args) + self.assertNotIn("--model", args) + self.assertNotIn("--dangerously-skip-permissions", args) + + def test_sonnet_alias_is_passed_without_rewriting_the_plugin(self): + harness.invoke("claude", "ui-critic", "Critique only", str(self.cwd), "sonnet", env=self._env()) + calls = self._calls() + args = calls[1]["args"] + self.assertEqual(args[args.index("--model") + 1], "sonnet") + self.assertEqual(args[args.index("--agent") + 1], "ui-critic") + + def test_claude_alias_is_blocked_when_local_auth_status_is_negative(self): + with self.assertRaisesRegex(harness.HarnessError, "auth status"): + harness.invoke( + "claude", "reviewer", "Review", str(self.cwd), "opus", + env=self._env(DW_CLAUDE_AUTH='{"loggedIn": false}'), + ) + calls = self._calls() + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]["args"], ["auth", "status", "--json"]) + + def test_opencode_uses_exact_variant_and_role_agent_without_auto_approval(self): + model_key = "openai/gpt-6-luna#high" + self._snapshot("opencode", model_key) + result = harness.invoke( + "opencode", "planner", "Plan only", str(self.cwd), model_key, + env=self._env(DW_MODELS="gpt-6-luna"), + ) + self.assertEqual(result["returncode"], 0) + calls = self._calls() + self.assertEqual(calls[0]["args"], ["models", "openai"]) + args = calls[1]["args"] + self.assertEqual(args[:3], ["run", "--agent", "planner"]) + self.assertEqual(args[args.index("--model") + 1], model_key) + self.assertNotIn("--auto", args) + self.assertEqual(calls[1]["cwd"], str(self.cwd)) + self.assertIn("stderr omitido por segurança", result["output"]) + self.assertNotIn("sk-secret", result["output"]) + + def test_omp_runs_with_exact_tools_and_always_ask(self): + model_key = "openai/gpt-6-luna#high" + self._snapshot("omp", model_key, variant="high", provider="openai", model_id="gpt-6-luna") + result = harness.invoke( + "omp", "explorer", "Inspect only", str(self.cwd), model_key, + env=self._env(DW_MODELS_JSON=json.dumps([{"provider": "openai", "id": "gpt-6-luna", + "thinking": ["low", "high"]}])), + ) + self.assertEqual(result["returncode"], 0) + calls = self._calls() + self.assertEqual(calls[0]["args"], ["models", "openai", "--json"]) + args = calls[1]["args"] + self.assertEqual(args[0], "-p") + self.assertEqual(args[args.index("--model") + 1], "openai/gpt-6-luna") + self.assertEqual(args[args.index("--thinking") + 1], "high") + self.assertEqual(args[args.index("--approval-mode") + 1], "always-ask") + self.assertNotIn("--auto-approve", args) + self.assertEqual(args[args.index("--tools") + 1], "read,grep,glob") + self.assertEqual(args[args.index("--cwd") + 1], str(self.cwd)) + + def test_omp_unknown_thinking_variant_is_rejected_before_listing_or_dispatch(self): + model_key = "openai/model#unsupported" + self._snapshot("omp", model_key, variant="unsupported", provider="openai", model_id="model") + with self.assertRaisesRegex(harness.HarnessError, "variante"): + harness.invoke("omp", "explorer", "Inspect", str(self.cwd), model_key, env=self._env()) + self.assertEqual(self._calls(), []) + + def test_omp_rejects_thinking_not_supported_by_current_model_listing(self): + model_key = "openai/gpt-6-luna#high" + self._snapshot("omp", model_key, variant="high", provider="openai", model_id="gpt-6-luna") + listed = {"models": [{"provider": "openai", "id": "gpt-6-luna", "thinking": ["low"]}]} + with self.assertRaisesRegex(harness.HarnessError, "indisponível"): + harness.invoke("omp", "explorer", "Inspect", str(self.cwd), model_key, + env=self._env(DW_MODELS_JSON=json.dumps(listed))) + self.assertEqual([call["args"] for call in self._calls()], [["models", "openai", "--json"]]) + + def test_omp_slash_model_id_keeps_provider_and_qualified_selector(self): + model_key = "openrouter/anthropic/claude-sonnet-4.6" + self._snapshot("omp", model_key) + listed = {"models": [{"provider": "openrouter", "id": "anthropic/claude-sonnet-4.6", + "selector": model_key, "thinking": None}]} + result = harness.invoke("omp", "explorer", "Inspect", str(self.cwd), model_key, + env=self._env(DW_MODELS_JSON=json.dumps(listed))) + self.assertEqual(result["returncode"], 0) + calls = self._calls() + self.assertEqual(calls[0]["args"], ["models", "openrouter", "--json"]) + self.assertEqual(calls[1]["args"][calls[1]["args"].index("--model") + 1], model_key) + + + def test_autonomous_omp_uses_bwrap_and_bounded_tools_with_auto_approval(self): + model_key = "openai/gpt-6-luna#high" + self._snapshot("omp", model_key, variant="high", provider="openai", model_id="gpt-6-luna") + trace = self._fake_bwrap() + result = self._invoke_with_fake_bwrap( + "omp", "implementer", "Inspect only", str(self.cwd), model_key, + env=self._env( + DEV_WORKFLOWS_AUTONOMOUS="1", + DW_BWRAP_TRACE=str(trace), + DW_MODELS_JSON=json.dumps([{"provider": "openai", "id": "gpt-6-luna", + "thinking": ["low", "high"]}]), + ), + ) + self.assertEqual(result["returncode"], 0) + self.assertEqual(self._calls(), []) + calls = [json.loads(line) for line in trace.read_text(encoding="utf-8").splitlines()] + self.assertEqual(len(calls), 2) + for args in calls: + self.assertEqual(args[:6], ["--unshare-all", "--share-net", "--die-with-parent", "--ro-bind", "/", "/"]) + binds = [args[index + 1:index + 3] for index, value in enumerate(args) if value == "--bind"] + self.assertIn([str(self.cwd), str(self.cwd)], binds) + readonly = [args[index + 1:index + 3] for index, value in enumerate(args) if value == "--ro-bind"] + self.assertIn([str(self.cwd / ".git"), str(self.cwd / ".git")], readonly) + preflight = calls[0][calls[0].index("--") + 1:] + self.assertEqual(pathlib.Path(preflight[0]).name, "omp") + self.assertEqual(preflight[1:4], ["--no-extensions", "models", "openai"]) + command = calls[1][calls[1].index("--") + 1:] + self.assertEqual(pathlib.Path(command[0]).name, "omp") + self.assertEqual(command[1], "-p") + for flag in ("--no-extensions", "--no-lsp", "--no-session", "--auto-approve"): + self.assertIn(flag, command) + self.assertEqual(command[command.index("--model") + 1], "openai/gpt-6-luna") + self.assertEqual(command[command.index("--thinking") + 1], "high") + self.assertEqual(command[command.index("--tools") + 1], "read,grep,glob,edit,write") + self.assertNotIn("bash", command) + self.assertNotIn("web_search", command) + + def test_autonomous_opencode_uses_pure_mode_and_synthetic_deny_agent(self): + model_key = "openai/gpt-6-luna#high" + self._snapshot("opencode", model_key) + trace = self._fake_bwrap() + meta, prompt = harness._agent("implementer") + generated = harness._opencode_loop_agent(meta, prompt) + self.assertIn(' "*": deny', generated) + self.assertIn(" bash: deny", generated) + self.assertIn(" edit: allow", generated) + self.assertIn(" task: deny", generated) + result = self._invoke_with_fake_bwrap( + "opencode", "implementer", "Implement safely", str(self.cwd), model_key, + env=self._env(DEV_WORKFLOWS_AUTONOMOUS="1", DW_BWRAP_TRACE=str(trace), DW_MODELS="gpt-6-luna"), + ) + self.assertEqual(result["returncode"], 0) + self.assertEqual(self._calls(), []) + calls = [json.loads(line) for line in trace.read_text(encoding="utf-8").splitlines()] + self.assertEqual(len(calls), 2) + preflight = calls[0][calls[0].index("--") + 1:] + self.assertEqual(pathlib.Path(preflight[0]).name, "opencode") + self.assertEqual(preflight[1:3], ["--pure", "models"]) + command = calls[1][calls[1].index("--") + 1:] + self.assertEqual(pathlib.Path(command[0]).name, "opencode") + self.assertEqual(command[1:4], ["--pure", "run", "--agent"]) + agent_name = command[4] + self.assertRegex(agent_name, r"\Adev-workflows-loop-implementer-[0-9a-f]{16}\Z") + self.assertIn("--file", calls[1]) + + def test_autonomous_claude_uses_bare_mode_and_declared_safe_tools(self): + trace = self._fake_bwrap() + result = self._invoke_with_fake_bwrap( + "claude", "reviewer", "Review only", str(self.cwd), "opus", + env=self._env(DEV_WORKFLOWS_AUTONOMOUS="1", DW_BWRAP_TRACE=str(trace)), + ) + self.assertEqual(result["returncode"], 0) + self.assertEqual(self._calls(), []) + calls = [json.loads(line) for line in trace.read_text(encoding="utf-8").splitlines()] + self.assertEqual(len(calls), 2) + auth = calls[0][calls[0].index("--") + 1:] + self.assertEqual(pathlib.Path(auth[0]).name, "claude") + self.assertEqual(auth[1:], ["auth", "status", "--json"]) + command = calls[1][calls[1].index("--") + 1:] + self.assertIn("--bare", command) + self.assertIn("--no-session-persistence", command) + self.assertEqual(command[command.index("--permission-mode") + 1], "dontAsk") + agent_config = json.loads(command[command.index("--agents") + 1]) + self.assertEqual(agent_config["reviewer"]["tools"], ["Read", "Grep", "Glob"]) + self.assertEqual(command[command.index("--tools") + 1], "Read,Grep,Glob") + self.assertNotIn("Bash", command) + + def test_autonomous_mode_fails_closed_without_bwrap(self): + self._snapshot("omp", "openai/gpt-6-luna") + def find_without_bwrap(name, path=None): + return None if name == "bwrap" else REAL_WHICH(name, path=path) + with mock.patch.object(harness.shutil, "which", side_effect=find_without_bwrap): + with self.assertRaisesRegex(harness.HarnessError, "bwrap"): + harness.invoke( + "omp", "explorer", "Inspect", str(self.cwd), "openai/gpt-6-luna", + env=self._env(PATH=str(self.bin), DEV_WORKFLOWS_AUTONOMOUS="1"), + ) + self.assertEqual(self._calls(), []) + + def test_autonomous_rejects_non_worktree_cwd_before_dispatch(self): + other = self.root / "ordinary-directory" + other.mkdir() + with self.assertRaisesRegex(harness.HarnessError, "worktree"): + harness.invoke( + "omp", "explorer", "Inspect", str(other), "openai/gpt-6-luna", + env=self._env(DEV_WORKFLOWS_AUTONOMOUS="1"), + ) + self.assertEqual(self._calls(), []) + + def test_autonomous_does_not_resolve_cli_from_writable_worktree(self): + self._snapshot("opencode", "openai/gpt-6-luna") + shadow = self.cwd / "opencode" + shadow.write_text("#!/bin/sh\nexit 0\n", encoding="utf-8") + shadow.chmod(0o755) + with self.assertRaisesRegex(harness.HarnessError, "PATH seguro"): + harness.invoke( + "opencode", "planner", "Plan", str(self.cwd), "openai/gpt-6-luna", + env=self._env(PATH=str(self.cwd), DEV_WORKFLOWS_AUTONOMOUS="1"), + ) + self.assertEqual(self._calls(), []) + + def test_unentitled_model_is_blocked_before_cli_is_started(self): + self._snapshot("opencode", "openai/gpt-6-luna", entitled=False) + with self.assertRaisesRegex(harness.HarnessError, "autorização local"): + harness.invoke( + "opencode", "planner", "Plan", str(self.cwd), "openai/gpt-6-luna", + env=self._env(DW_MODELS="gpt-6-luna"), + ) + self.assertEqual(self._calls(), []) + + def test_model_absent_from_local_cli_is_blocked_before_agent_dispatch(self): + self._snapshot("opencode", "openai/gpt-6-luna") + with self.assertRaisesRegex(harness.HarnessError, "indisponível na CLI local"): + harness.invoke( + "opencode", "planner", "Plan", str(self.cwd), "openai/gpt-6-luna", + env=self._env(DW_MODELS="openai/different-model"), + ) + calls = self._calls() + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0]["args"], ["models", "openai"]) + + +if __name__ == "__main__": + unittest.main(verbosity=1) diff --git a/qa/test_omp_adapter.py b/qa/test_omp_adapter.py new file mode 100644 index 0000000..746abde --- /dev/null +++ b/qa/test_omp_adapter.py @@ -0,0 +1,45 @@ +"""Ownership and safety behavior for native OMP package builds.""" +import importlib.util +import tempfile +import unittest +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location("omp_build", ROOT / "adapters" / "omp" / "build.py") +BUILD = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(BUILD) + + +class OmpAdapterTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.package = Path(self.tmp.name) / "dev-workflows" + BUILD.build_package(ROOT, self.package) + + def test_refuses_to_replace_unowned_destination(self): + with tempfile.TemporaryDirectory() as tmp: + dest = Path(tmp) / "third-party" + dest.mkdir() + sentinel = dest / "keep" + sentinel.write_text("untouched") + with self.assertRaises(FileExistsError): + BUILD.build_package(ROOT, dest) + self.assertEqual(sentinel.read_text(), "untouched") + + def test_preserves_manual_edits_in_managed_package(self): + prompt = self.package / "agents" / "planner.md" + prompt.write_text(prompt.read_text() + "\nMudança do usuário.\n") + with self.assertRaisesRegex(FileExistsError, "alterações manuais"): + BUILD.build_package(ROOT, self.package) + self.assertIn("Mudança do usuário.", prompt.read_text()) + + def test_rejects_invalid_model_selectors(self): + with tempfile.TemporaryDirectory() as tmp: + with self.assertRaises(ValueError): + BUILD.build_package(ROOT, Path(tmp) / "package", {"planner": "gpt-6-luna"}) + + +if __name__ == "__main__": + unittest.main() diff --git a/scripts/harness_runner.py b/scripts/harness_runner.py new file mode 100755 index 0000000..c76e3ed --- /dev/null +++ b/scripts/harness_runner.py @@ -0,0 +1,496 @@ +#!/usr/bin/env python3 +"""Execução headless e conservadora para Claude Code, OpenCode e OMP.""" +from __future__ import annotations + +import importlib.util +import json +import os +import re +import shutil +import secrets +import subprocess +import tempfile +from pathlib import Path +from typing import Any, Mapping + +ROOT = Path(__file__).resolve().parent.parent +AGENTS = ROOT / "agents" +OMP_TOOLS = { + "Read": "read", "Edit": "edit", "Write": "write", "Bash": "bash", + "Grep": "grep", "Glob": "glob", "WebFetch": "web_search", + "WebSearch": "web_search", "NotebookEdit": "notebook", +} + +OMP_THINKING_LEVELS = {"off", "minimal", "low", "medium", "high", "xhigh", "max", "auto"} + + + +class HarnessError(RuntimeError): + """Harness request cannot be safely or exactly dispatched.""" + + +def _load_catalog_module(): + path = ROOT / "scripts" / "model_catalog.py" + spec = importlib.util.spec_from_file_location("dev_workflows_model_catalog", path) + if spec is None or spec.loader is None: + raise HarnessError("catálogo local indisponível") + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def _agent(role: str) -> tuple[dict[str, str], str]: + if not re.fullmatch(r"[a-z][a-z0-9-]*", role): + raise HarnessError(f"papel inválido: {role!r}") + path = AGENTS / f"{role}.md" + try: + text = path.read_text(encoding="utf-8") + except OSError as exc: + raise HarnessError(f"papel desconhecido: {role}") from exc + match = re.match(r"\A---\n(.*?)\n---\n(.*)\Z", text, re.S) + if not match: + raise HarnessError(f"frontmatter inválido para papel {role}") + meta = dict(re.findall(r"^(\w+):\s*(.*)$", match.group(1), re.M)) + if meta.get("name") != role: + raise HarnessError(f"identidade inválida para papel {role}") + return meta, match.group(2).replace("${CLAUDE_PLUGIN_ROOT}", str(ROOT)) + + +def _model_for_cli(tool: str, model_key: str, meta: dict[str, str]) -> tuple[str | None, str]: + """Return CLI selector and exact catalog key; never strip an unsupported variant.""" + if not isinstance(model_key, str) or not model_key: + raise HarnessError("model_key obrigatório") + if tool == "claude": + if model_key == "inherit": + inherited = meta.get("model", "") + if inherited == "inherit": + raise HarnessError("Claude `inherit` não identifica um modelo verificável para este papel") + selector = inherited + catalog_key = inherited + else: + selector = model_key + catalog_key = model_key + if selector not in {"haiku", "sonnet", "opus"} and not re.fullmatch(r"claude-[A-Za-z0-9.-]+", selector): + raise HarnessError("Claude CLI aceita aliases haiku/sonnet/opus ou IDs Claude; model_key não é exato para esta ferramenta") + return (None if model_key == "inherit" else selector), catalog_key + if model_key == "inherit": + raise HarnessError("inherit não identifica provider/model para esta ferramenta") + if not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.:+/-]+(?:#[A-Za-z0-9_.+-]+)?", model_key): + raise HarnessError(f"model_key inválido ou não exato para {tool}: {model_key!r}") + return model_key, model_key + + +def _require_catalog_entry(tool: str, model_key: str, cache_dir: str | None) -> dict[str, Any]: + catalog = _load_catalog_module() + snapshot = catalog.load_snapshot(tool, cache_dir) + if not isinstance(snapshot, dict) or snapshot.get("tool") != tool: + raise HarnessError(f"snapshot local de {tool} indisponível; despacho bloqueado") + matches = [row for row in snapshot.get("models", []) + if isinstance(row, dict) and row.get("tool") == tool and row.get("model_key") == model_key] + if len(matches) != 1: + raise HarnessError(f"model_key {model_key!r} não é um ID exato e único no catálogo local de {tool}") + availability = matches[0].get("availability") + if not isinstance(availability, dict) or any(availability.get(field) is not True + for field in ("local_listed", "authenticated", "entitled")): + raise HarnessError(f"disponibilidade/autorização local não verificada para {model_key!r}; despacho bloqueado") + return matches[0] + +def _role_tools(meta: dict[str, str]) -> list[str]: + declared = {item.strip() for item in meta.get("tools", "").split(",") if item.strip()} + return [name for name in ("Read", "Grep", "Glob", "Edit", "Write") if name in declared] + + +def _opencode_loop_agent(meta: dict[str, str], role_prompt: str) -> str: + tools = set(_role_tools(meta)) + allowed = { + "read": "Read" in tools, + "edit": bool(tools & {"Edit", "Write"}), + "glob": "Glob" in tools, + "grep": "Grep" in tools, + "list": False, + "bash": False, + "task": False, + "external_directory": False, + "todowrite": False, + "webfetch": False, + "websearch": False, + "lsp": False, + "skill": False, + "question": False, + "doom_loop": False, + } + lines = ["---", f"description: {json.dumps(meta['description'], ensure_ascii=False)}", + "mode: primary", "permission:", ' "*": deny'] + lines.extend(f" {name}: {'allow' if enabled else 'deny'}" for name, enabled in allowed.items()) + return "\n".join(lines + ["---", "", role_prompt]) + + +def _memfd(contents: bytes) -> int: + if not hasattr(os, "memfd_create"): + raise HarnessError("sandbox requer suporte a memfd para config efêmera") + fd = os.memfd_create("dev-workflows-harness") + os.write(fd, contents) + os.lseek(fd, 0, os.SEEK_SET) + return fd + + +def _within(path: Path, root: Path) -> bool: + return path == root or root in path.parents + + +def _validate_autonomous_worktree(cwd: str) -> None: + worktree = Path(cwd).resolve() + if worktree.parent.name != "worktrees" or not ( + worktree.name == "baseline" or re.fullmatch(r"iteration-[0-9]+", worktree.name) + ): + raise HarnessError("modo autônomo exige worktree isolado baseline/iteration-*") + checkout = ROOT.resolve() + if _within(worktree, checkout) or _within(checkout, worktree): + raise HarnessError("modo autônomo não pode gravar no checkout de origem") + marker = worktree / ".git" + if marker.is_symlink() or not marker.is_file(): + raise HarnessError("cwd autônomo não é um git worktree") + match = re.fullmatch(r"gitdir:\s*(.+)\s*", marker.read_text(encoding="utf-8")) + if not match: + raise HarnessError("ponteiro .git inválido no worktree autônomo") + gitdir = Path(match.group(1)) + if not gitdir.is_absolute(): + gitdir = worktree / gitdir + gitdir = gitdir.resolve() + if not gitdir.is_dir() or _within(gitdir, worktree): + raise HarnessError("gitdir do worktree autônomo ausente ou inseguro") + + +def _sandbox_path(cwd: str, env: dict[str, str]) -> str: + worktree = Path(cwd).resolve() + entries: list[str] = [] + seen: set[Path] = set() + for entry in env.get("PATH", "").split(os.pathsep): + path = Path(entry or cwd).expanduser() + if not path.is_absolute(): + path = worktree / path + try: + path = path.resolve() + except OSError: + continue + if not path.is_dir() or _within(path, worktree) or path in seen: + continue + seen.add(path) + entries.append(str(path)) + return os.pathsep.join(entries) + + +def _sandbox_command(tool: str, command: list[str], cwd: str, env: dict[str, str], + opencode_agent: str | None = None, temp_root: str | None = None + ) -> tuple[list[str], tuple[int, ...]]: + if env.get("DEV_WORKFLOWS_AUTONOMOUS") != "1": + return command, () + bwrap = shutil.which("bwrap", path="/usr/bin:/bin") + if not bwrap: + raise HarnessError("modo autônomo requer bwrap; despacho bloqueado") + if not temp_root: + raise HarnessError("sandbox sem diretório temporário privado") + temp_data = f"{temp_root}/data" + temp_config = f"{temp_root}/config" + args = [ + bwrap, "--unshare-all", "--share-net", "--die-with-parent", + "--ro-bind", "/", "/", "--proc", "/proc", "--dev", "/dev", + "--bind", temp_root, temp_root, + "--dir", f"{temp_root}/tmp", "--dir", f"{temp_root}/cache", + "--dir", f"{temp_root}/state", "--dir", temp_data, + ] + args.extend(("--setenv", "PATH", _sandbox_path(cwd, env))) + pass_fds: list[int] = [] + if tool == "opencode": + for directory in (temp_config, f"{temp_config}/opencode", f"{temp_config}/opencode/agents", + f"{temp_data}/opencode"): + args.extend(("--dir", directory)) + args.extend(("--setenv", "XDG_CONFIG_HOME", temp_config, + "--setenv", "XDG_DATA_HOME", temp_data, + "--setenv", "TMPDIR", f"{temp_root}/tmp")) + config_home = Path(env.get("XDG_CONFIG_HOME", str(Path.home() / ".config"))) + configured = env.get("OPENCODE_CONFIG") + if not configured: + for filename in ("opencode.json", "opencode.jsonc"): + candidate = config_home / "opencode" / filename + if candidate.is_file(): + configured = str(candidate) + break + if configured: + args.extend(("--setenv", "OPENCODE_CONFIG", configured)) + auth_home = Path(env.get("XDG_DATA_HOME", str(Path.home() / ".local" / "share"))) + auth_file = auth_home / "opencode" / "auth.json" + if auth_file.is_file(): + auth_target = f"{temp_data}/opencode/auth.json" + fd = _memfd(b"") + pass_fds.append(fd) + args.extend(("--file", str(fd), auth_target, "--ro-bind", str(auth_file), auth_target)) + if opencode_agent is not None: + role_name = command[command.index("--agent") + 1] + agent_target = f"{temp_config}/opencode/agents/{role_name}.md" + fd = _memfd(opencode_agent.encode("utf-8")) + pass_fds.append(fd) + args.extend(("--file", str(fd), agent_target)) + args.extend(("--setenv", "XDG_CACHE_HOME", f"{temp_root}/cache", + "--setenv", "XDG_STATE_HOME", f"{temp_root}/state")) + else: + args.extend(("--setenv", "TMPDIR", f"{temp_root}/tmp", + "--setenv", "XDG_CACHE_HOME", f"{temp_root}/cache", + "--setenv", "XDG_STATE_HOME", f"{temp_root}/state")) + if tool == "omp": + args.extend(("--dir", f"{temp_root}/omp-agent", + "--setenv", "PI_CODING_AGENT_DIR", f"{temp_root}/omp-agent")) + elif tool == "claude": + args.extend(("--setenv", "CLAUDE_CODE_SKIP_PROMPT_HISTORY", "1")) + marker = str(Path(cwd) / ".git") + args.extend(("--bind", cwd, cwd, "--ro-bind", marker, marker, "--chdir", cwd, "--", *command)) + # The network namespace is shared for provider APIs; it is not an API-only egress filter. + return args, tuple(pass_fds) + + +def _run_process(command: list[str], cwd: str, env: dict[str, str], timeout: float, tool: str, + opencode_agent: str | None = None) -> subprocess.CompletedProcess: + autonomous = env.get("DEV_WORKFLOWS_AUTONOMOUS") == "1" + temp_dir = None + pass_fds: tuple[int, ...] = () + try: + if autonomous: + safe_path = _sandbox_path(cwd, env) + if not safe_path: + raise HarnessError("PATH seguro vazio; despacho bloqueado") + executable = shutil.which(command[0], path=safe_path) + if not executable: + raise HarnessError(f"CLI {command[0]!r} não encontrado em PATH seguro") + if _within(Path(executable).resolve(), Path(cwd).resolve()): + raise HarnessError(f"CLI {command[0]!r} está dentro do worktree; despacho bloqueado") + command = [executable, *command[1:]] + temp_dir = tempfile.TemporaryDirectory(prefix="dev-workflows-harness-", dir="/tmp") + wrapped, pass_fds = _sandbox_command( + tool, command, cwd, env, opencode_agent, temp_dir.name if temp_dir else None + ) + return subprocess.run(wrapped, cwd=cwd, env=env, capture_output=True, text=True, + errors="replace", timeout=timeout, check=False, pass_fds=pass_fds) + finally: + for fd in pass_fds: + os.close(fd) + if temp_dir is not None: + temp_dir.cleanup() + + +def _claude_auth_status(cwd: str, env: dict[str, str], timeout: float) -> bool: + try: + result = _run_process(["claude", "auth", "status", "--json"], cwd, env, timeout, "claude") + except (OSError, subprocess.TimeoutExpired): + return False + if result.returncode != 0: + return False + try: + status = json.loads(result.stdout) + except json.JSONDecodeError: + return False + return isinstance(status, dict) and status.get("loggedIn") is True + + +def _omp_cli_model(row: dict[str, Any], model_key: str) -> tuple[str, str | None]: + variant = row.get("variant") + if variant is None: + if "#" in model_key: + raise HarnessError(f"model_key OMP variante sem metadado variant: {model_key!r}") + return model_key, None + if variant not in OMP_THINKING_LEVELS: + raise HarnessError(f"OMP CLI não oferece seleção exata da variante {variant!r}") + base_key, separator, key_variant = model_key.rpartition("#") + if not separator or key_variant != variant: + raise HarnessError(f"variante OMP não corresponde ao model_key canônico {model_key!r}") + provider, model_id = row.get("provider"), row.get("model_id") + if not isinstance(provider, str) or not isinstance(model_id, str) or base_key != f"{provider}/{model_id}": + raise HarnessError(f"provider/model_id do catálogo não confirma {model_key!r}") + return base_key, variant + + + + +def _local_models_command(tool: str, model_key: str, cwd: str, env: dict[str, str], + timeout: float, thinking: str | None = None) -> bool: + provider, model = model_key.split("/", 1) + base_model = model.split("#", 1)[0] + autonomous = env.get("DEV_WORKFLOWS_AUTONOMOUS") == "1" + if tool == "opencode": + command = ["opencode"] + (["--pure"] if autonomous else []) + ["models", provider] + else: + command = ["omp"] + (["--no-extensions"] if autonomous else []) + ["models", provider, "--json"] + try: + result = _run_process(command, cwd, env, timeout, tool) + except (OSError, subprocess.TimeoutExpired): + return False + if result.returncode != 0: + return False + if tool == "omp": + try: + payload = json.loads(result.stdout) + except json.JSONDecodeError: + return False + return _json_catalog_has_model(payload, provider, base_model, thinking) + for line in result.stdout.splitlines(): + candidate = line.strip().strip("`'\" ") + if candidate.startswith(provider + "/"): + candidate = candidate[len(provider) + 1:] + if candidate in {model, base_model}: + return True + return False + + +def _json_catalog_has_model(value: Any, provider: str, model_id: str, thinking: str | None = None) -> bool: + if isinstance(value, list): + return any(_json_catalog_has_model(item, provider, model_id, thinking) for item in value) + if not isinstance(value, dict): + return False + row_provider = value.get("provider") or value.get("providerID") or value.get("provider_id") + row_id = value.get("id") or value.get("modelID") or value.get("model_id") + full_id = value.get("selector") or value.get("model_key") or value.get("model") + matches = ((row_provider == provider and row_id == model_id) + or (isinstance(full_id, str) and full_id == f"{provider}/{model_id}")) + if matches and (thinking is None or + (isinstance(value.get("thinking"), list) and thinking in value["thinking"])): + return True + return any(_json_catalog_has_model(item, provider, model_id, thinking) for item in value.values() + if isinstance(item, (dict, list))) + + +def _omp_tools(meta: dict[str, str], autonomous: bool = False) -> list[str]: + if autonomous: + return [OMP_TOOLS[name] for name in _role_tools(meta)] + declared = {item.strip() for item in meta.get("tools", "").split(",") if item.strip()} + read_only = not bool(declared & {"Edit", "Write"}) + tool_order = ("Read", "Edit", "Write", "Bash", "Grep", "Glob", "WebFetch", "WebSearch", "NotebookEdit") + return list(dict.fromkeys( + OMP_TOOLS[name] for name in tool_order + if name in declared and not (read_only and name in {"Edit", "Write", "Bash"}) + )) + +def _omp_skills(meta: dict[str, str]) -> list[str]: + match = re.search(r"\[(.*?)\]", meta.get("skills", "")) + return re.findall(r"[\w-]+", match.group(1)) if match else [] + + +def _stderr_summary(stderr: str) -> str: + if not stderr: + return "" + lines = len(stderr.splitlines()) + return f"[stderr omitido por segurança: {lines} linha(s), {len(stderr)} caracteres]" + + +def _text(value: str | bytes | None) -> str: + if value is None: + return "" + if isinstance(value, bytes): + return value.decode("utf-8", errors="replace") + return value + + +def invoke(tool: str, role: str, prompt: str, cwd: str, model_key: str, + timeout: float = 600, env: Mapping[str, str] | None = None) -> dict[str, Any]: + """Invoke one role; only bounded autonomous OMP dispatch uses auto-approval. + + The local catalog must attest local listing, authentication and entitlement. + OpenCode/OMP local models are rechecked immediately before dispatch. Autonomous + calls require bwrap and a tool allowlist; no fallback or config rewrite occurs. + """ + if tool not in {"claude", "opencode", "omp"}: + raise HarnessError(f"ferramenta não suportada: {tool!r}") + if not isinstance(prompt, str) or not prompt: + raise HarnessError("prompt vazio") + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)) or timeout <= 0: + raise HarnessError("timeout deve ser positivo") + workdir = str(Path(cwd).resolve()) + if not Path(workdir).is_dir(): + raise HarnessError(f"cwd não existe ou não é diretório: {cwd!r}") + child_env = os.environ.copy() + if env is not None: + if any(not isinstance(key, str) or not isinstance(value, str) for key, value in env.items()): + raise HarnessError("env deve conter somente pares string/string") + child_env.update(env) + autonomous = child_env.get("DEV_WORKFLOWS_AUTONOMOUS") == "1" + if autonomous: + _validate_autonomous_worktree(workdir) + meta, role_prompt = _agent(role) + cli_model, catalog_key = _model_for_cli(tool, model_key, meta) + cache_dir = child_env.get("DEV_WORKFLOWS_MODEL_CACHE") + if tool == "claude": + if not _claude_auth_status(workdir, child_env, float(timeout)): + raise HarnessError("claude auth status não confirmou uma sessão autenticada; despacho bloqueado") + row = None if model_key in {"haiku", "sonnet", "opus", "inherit"} else _require_catalog_entry(tool, catalog_key, cache_dir) + else: + row = _require_catalog_entry(tool, catalog_key, cache_dir) + omp_thinking = None + local_model_key = catalog_key + if tool == "omp": + cli_model, omp_thinking = _omp_cli_model(row, catalog_key) + local_model_key = cli_model + if tool in {"opencode", "omp"} and not _local_models_command( + tool, local_model_key, workdir, child_env, float(timeout), omp_thinking): + raise HarnessError(f"provider/model indisponível na CLI local de {tool}: {catalog_key!r}; despacho bloqueado") + + opencode_agent = None + if tool == "claude": + if autonomous: + safe_tools = _role_tools(meta) + agent_config = json.dumps({ + role: {"description": meta["description"], "prompt": role_prompt, "tools": safe_tools} + }, ensure_ascii=False) + command = [ + "claude", "--bare", "--no-session-persistence", "-p", "--agent", role, + "--agents", agent_config, "--tools", ",".join(safe_tools), + "--permission-mode", "dontAsk", + ] + if cli_model is not None: + command.extend(("--model", cli_model)) + if safe_tools: + command.extend(("--allowedTools", *safe_tools)) + else: + command = ["claude", "-p", "--agent", role, "--permission-mode", "dontAsk"] + if cli_model is not None: + command.extend(("--model", cli_model)) + command.append(prompt) + elif tool == "opencode": + agent_name = f"dev-workflows-loop-{role}-{secrets.token_hex(8)}" if autonomous else role + command = ["opencode"] + (["--pure"] if autonomous else []) + ["run", "--agent", agent_name] + if cli_model is not None: + command.extend(("--model", cli_model)) + command.append(prompt) + if autonomous: + opencode_agent = _opencode_loop_agent(meta, role_prompt) + else: + command = ["omp", "-p", "--cwd", workdir] + if autonomous: + command.extend(("--no-extensions", "--no-lsp", "--no-session", "--auto-approve")) + else: + command.extend(("--approval-mode", "always-ask")) + if cli_model is not None: + command.extend(("--model", cli_model)) + if omp_thinking is not None: + command.extend(("--thinking", omp_thinking)) + command.extend(("--append-system-prompt", role_prompt)) + skills = _omp_skills(meta) + if skills: + command.extend(("--skills", ",".join(skills))) + tools = _omp_tools(meta, autonomous=autonomous) + if tools: + command.extend(("--tools", ",".join(tools))) + else: + command.append("--no-tools") + command.append(prompt) + try: + result = _run_process(command, workdir, child_env, float(timeout), tool, opencode_agent) + returncode = result.returncode + stdout, stderr = result.stdout, result.stderr + except subprocess.TimeoutExpired as exc: + returncode = 124 + stdout, stderr = _text(exc.stdout), _text(exc.stderr) + except OSError: + return {"returncode": 127, "output": f"CLI {tool} indisponível", "usage": None} + output = _text(stdout).rstrip() + summary = _stderr_summary(_text(stderr)) + if summary: + output = f"{output}\n{summary}" if output else summary + return {"returncode": returncode, "output": output, "usage": None} diff --git a/scripts/install.py b/scripts/install.py index 92c20d8..0d09026 100644 --- a/scripts/install.py +++ b/scripts/install.py @@ -1,14 +1,15 @@ #!/usr/bin/env python3 -"""Install dev-workflows into Claude Code, opencode or Hermes Agent. +"""Install dev-workflows into Claude Code, OpenCode, OMP or Hermes Agent. python3 scripts/install.py claude # plugin via local marketplace (skills, agents, hooks) python3 scripts/install.py opencode # skills + agents + plugin under ~/.config/opencode + python3 scripts/install.py omp # native skills + agents under ~/.omp/plugins/dev-workflows python3 scripts/install.py hermes # skills under ~/.hermes/skills (no agents/hooks there) -opencode and Hermes do not substitute ${CLAUDE_PLUGIN_ROOT}, so the copies point at this checkout: +OpenCode and Hermes do not substitute ${CLAUDE_PLUGIN_ROOT}, so their copies point at this checkout: keep it where it is, or re-run the install after moving it. """ -import argparse, os, re, shutil, subprocess, sys +import argparse, importlib.util, os, re, shutil, subprocess, sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) MARKER = ".dev-workflows-install" @@ -40,23 +41,74 @@ def install_skills(target_dir, force): def opencode_agent(path): - """Claude Code agent frontmatter -> opencode subagent frontmatter.""" - text = open(path, encoding="utf-8").read() + """Claude Code agent frontmatter -> OpenCode subagent with explicit permissions.""" + with open(path, encoding="utf-8") as source: + text = source.read() m = re.match(r"---\n(.*?)\n---\n(.*)", text, re.S) + if not m: + raise ValueError(f"frontmatter inválido em {path}") meta = dict(re.findall(r"^(\w+):\s*(.*)$", m.group(1), re.M)) tools = {t.strip() for t in meta.get("tools", "").split(",")} skills = re.findall(r"[\w-]+", meta.get("skills", "")) - perm = {"edit": "allow" if tools & {"Edit", "Write"} else "deny", - "bash": "allow" if "Bash" in tools else "deny", - "webfetch": "allow" if "WebFetch" in tools else "deny"} + read_only = not bool(tools & {"Edit", "Write"}) + allowed = { + "read": "Read" in tools, + "edit": not read_only and bool(tools & {"Edit", "Write"}), + "bash": not read_only and "Bash" in tools, + "glob": "Glob" in tools, + "grep": "Grep" in tools, + "list": False, + "task": False, + "external_directory": False, + "todowrite": False, + "webfetch": "WebFetch" in tools, + "websearch": "WebSearch" in tools, + "lsp": False, + "skill": "Skill" in tools, + "question": False, + "doom_loop": False, + } body = m.group(2).replace("${CLAUDE_PLUGIN_ROOT}", ROOT) if skills: body = f"Antes de começar, carregue a(s) skill(s): {', '.join(skills)}.\n\n" + body - head = ["# dev-workflows", f"description: {meta['description']}", "mode: subagent", "permission:"] - head += [f" {k}: {v}" for k, v in perm.items()] + head = ["# dev-workflows", f"description: {meta['description']}", "mode: subagent", "permission:", ' "*": deny'] + head += [f" {key}: {'allow' if value else 'deny'}" for key, value in allowed.items()] return "---\n" + "\n".join(head) + "\n---\n\n" + body +def _omp_builder(path=None): + path = path or os.path.join(ROOT, "adapters", "omp", "build.py") + spec = importlib.util.spec_from_file_location("dev_workflows_omp_build", path) + if spec is None or spec.loader is None: + raise ImportError(f"não foi possível carregar o adapter OMP: {path}") + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module.build_package + + +def install_omp(base=None, runner=None, build_package=None): + """Build and link the OMP package without replacing a foreign installation.""" + omp_root = os.path.realpath(base or os.path.expanduser("~/.omp")) + dest = os.path.join(omp_root, "plugins", "dev-workflows") + if os.path.lexists(dest) and (os.path.islink(dest) or not os.path.isdir(dest)): + sys.exit(f"{dest} existe e não é um pacote OMP deste checkout.") + build = build_package or _omp_builder() + try: + build(ROOT, dest) + except FileExistsError as exc: + sys.exit(str(exc)) + run = runner or subprocess.run + command = ["omp", "plugin", "link", dest] + try: + result = run(command, check=False) + except OSError: + sys.exit("OMP não encontrado; pacote gerado, mas `omp plugin link` não foi executado.") + if result.returncode != 0: + sys.exit(f"`omp plugin link {dest}` falhou com código {result.returncode}.") + print(f"plugin OMP instalado em {dest}") + + def install_opencode(force): base = os.path.expanduser("~/.config/opencode") install_skills(os.path.join(base, "skills"), force) @@ -83,13 +135,15 @@ def install_claude(): def main(): ap = argparse.ArgumentParser(description=__doc__.splitlines()[0]) - ap.add_argument("target", choices=["claude", "opencode", "hermes"]) - ap.add_argument("--force", action="store_true", help="sobrescreve pastas com o mesmo nome que não vieram deste script") + ap.add_argument("target", choices=["claude", "opencode", "hermes", "omp"]) + ap.add_argument("--force", action="store_true", help="sobrescreve skills/agentes externos; OMP nunca sobrescreve pacote não pertencente") a = ap.parse_args() if a.target == "claude": install_claude() elif a.target == "opencode": install_opencode(a.force) + elif a.target == "omp": + install_omp() else: install_skills(os.path.expanduser("~/.hermes/skills"), a.force) print("Hermes: agentes e hooks não se aplicam; rode o gate manualmente.")