feat(adapters): package OMP and confine autonomous runners

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)

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()