feat(plugin): a policy's gate rides in its plugin in every hook-carrying format, and its own skill files ride in its skill

.chock/bin/claude_code.py

@@ -844,6 +844,8 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N
 
 _RUNNER_PARTS = ('bin', 'gate.py')
 
+_PACKAGED_RUNNER = 'gate.py'
+
 _GIT = 'git'
 
 _DELETED = 'D'
@@ -867,7 +869,10 @@ def gate_path_from_argv(argv):
     return None
 
 def runner_for(gate):
-    """The vendored gate runner beside this compiled gate, or None when it is not there."""
+    """The gate runner: beside the gate in a plugin, under .chock/bin in a repository, else None."""
+    packaged = gate.resolve().parent / _PACKAGED_RUNNER
+    if packaged.exists():
+        return packaged
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
@@ -919,13 +924,13 @@ def writes_from_worktree(repo_root):
             continue
     return writes
 
-def run_gate(gate, writes, event):
+def run_gate(gate, writes, event, root=None):
     """Ask the vendored runner. Returns (outcome, message) and never decides for itself."""
     runner = runner_for(gate)
     if runner is None:
         return (GATE_ERRORED, 'the vendored gate runner is not installed beside this gate')
     try:
-        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False)
+        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False, cwd=str(root) if root is not None else None)
     except (OSError, _chock_subprocess.SubprocessError) as exc:
         return (GATE_ERRORED, str(exc))
     if proc.returncode == 0:
@@ -943,16 +948,25 @@ def root_for(gate):
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
+    if parents[_GATE_DEPTH_TO_CHOCK - 1].name != 'compiled' or parents[_GATE_DEPTH_TO_CHOCK].name != '.chock':
+        return None
     return parents[_GATE_DEPTH_TO_CHOCK].parent
 
+def repo_root_for(event, gate):
+    """The repository under judgement: the compiled layout's root, else where the agent works."""
+    root = root_for(gate)
+    if root is not None:
+        return root
+    cwd = getattr(event, 'cwd', None)
+    return _chock_Path(cwd) if cwd else _chock_Path.cwd()
+
 def writes_for(event, gate):
     """What this event puts under judgement: the call's own text, or what the turn left behind."""
     if event.event == PRE_TOOL:
         return writes_from_event(event)
     if (event.raw or {}).get('stop_hook_active'):
         return {}
-    root = root_for(gate)
-    return writes_from_worktree(root) if root is not None else {}
+    return writes_from_worktree(repo_root_for(event, gate))
 
 def evaluate_gate(argv, event):
     """The decision this event earns from a compiled gate, or None when it has nothing to say."""
@@ -963,7 +977,7 @@ def evaluate_gate(argv, event):
     writes = writes_for(event, gate)
     if not writes:
         return None
-    outcome, message = run_gate(gate, writes, name)
+    outcome, message = run_gate(gate, writes, name, repo_root_for(event, gate))
     if outcome == GATE_BLOCKED:
         return (VERDICT_DENY, message or f'Blocked by chock policy: {gate.parent.parent.name}')
     if outcome == GATE_ERRORED:

.chock/bin/codex_cli.py

@@ -830,6 +830,8 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N
 
 _RUNNER_PARTS = ('bin', 'gate.py')
 
+_PACKAGED_RUNNER = 'gate.py'
+
 _GIT = 'git'
 
 _DELETED = 'D'
@@ -853,7 +855,10 @@ def gate_path_from_argv(argv):
     return None
 
 def runner_for(gate):
-    """The vendored gate runner beside this compiled gate, or None when it is not there."""
+    """The gate runner: beside the gate in a plugin, under .chock/bin in a repository, else None."""
+    packaged = gate.resolve().parent / _PACKAGED_RUNNER
+    if packaged.exists():
+        return packaged
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
@@ -905,13 +910,13 @@ def writes_from_worktree(repo_root):
             continue
     return writes
 
-def run_gate(gate, writes, event):
+def run_gate(gate, writes, event, root=None):
     """Ask the vendored runner. Returns (outcome, message) and never decides for itself."""
     runner = runner_for(gate)
     if runner is None:
         return (GATE_ERRORED, 'the vendored gate runner is not installed beside this gate')
     try:
-        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False)
+        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False, cwd=str(root) if root is not None else None)
     except (OSError, _chock_subprocess.SubprocessError) as exc:
         return (GATE_ERRORED, str(exc))
     if proc.returncode == 0:
@@ -929,16 +934,25 @@ def root_for(gate):
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
+    if parents[_GATE_DEPTH_TO_CHOCK - 1].name != 'compiled' or parents[_GATE_DEPTH_TO_CHOCK].name != '.chock':
+        return None
     return parents[_GATE_DEPTH_TO_CHOCK].parent
 
+def repo_root_for(event, gate):
+    """The repository under judgement: the compiled layout's root, else where the agent works."""
+    root = root_for(gate)
+    if root is not None:
+        return root
+    cwd = getattr(event, 'cwd', None)
+    return _chock_Path(cwd) if cwd else _chock_Path.cwd()
+
 def writes_for(event, gate):
     """What this event puts under judgement: the call's own text, or what the turn left behind."""
     if event.event == PRE_TOOL:
         return writes_from_event(event)
     if (event.raw or {}).get('stop_hook_active'):
         return {}
-    root = root_for(gate)
-    return writes_from_worktree(root) if root is not None else {}
+    return writes_from_worktree(repo_root_for(event, gate))
 
 def evaluate_gate(argv, event):
     """The decision this event earns from a compiled gate, or None when it has nothing to say."""
@@ -949,7 +963,7 @@ def evaluate_gate(argv, event):
     writes = writes_for(event, gate)
     if not writes:
         return None
-    outcome, message = run_gate(gate, writes, name)
+    outcome, message = run_gate(gate, writes, name, repo_root_for(event, gate))
     if outcome == GATE_BLOCKED:
         return (VERDICT_DENY, message or f'Blocked by chock policy: {gate.parent.parent.name}')
     if outcome == GATE_ERRORED:

.chock/bin/cursor.py

@@ -749,6 +749,8 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N
 
 _RUNNER_PARTS = ('bin', 'gate.py')
 
+_PACKAGED_RUNNER = 'gate.py'
+
 _GIT = 'git'
 
 _DELETED = 'D'
@@ -772,7 +774,10 @@ def gate_path_from_argv(argv):
     return None
 
 def runner_for(gate):
-    """The vendored gate runner beside this compiled gate, or None when it is not there."""
+    """The gate runner: beside the gate in a plugin, under .chock/bin in a repository, else None."""
+    packaged = gate.resolve().parent / _PACKAGED_RUNNER
+    if packaged.exists():
+        return packaged
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
@@ -824,13 +829,13 @@ def writes_from_worktree(repo_root):
             continue
     return writes
 
-def run_gate(gate, writes, event):
+def run_gate(gate, writes, event, root=None):
     """Ask the vendored runner. Returns (outcome, message) and never decides for itself."""
     runner = runner_for(gate)
     if runner is None:
         return (GATE_ERRORED, 'the vendored gate runner is not installed beside this gate')
     try:
-        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False)
+        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False, cwd=str(root) if root is not None else None)
     except (OSError, _chock_subprocess.SubprocessError) as exc:
         return (GATE_ERRORED, str(exc))
     if proc.returncode == 0:
@@ -848,16 +853,25 @@ def root_for(gate):
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
+    if parents[_GATE_DEPTH_TO_CHOCK - 1].name != 'compiled' or parents[_GATE_DEPTH_TO_CHOCK].name != '.chock':
+        return None
     return parents[_GATE_DEPTH_TO_CHOCK].parent
 
+def repo_root_for(event, gate):
+    """The repository under judgement: the compiled layout's root, else where the agent works."""
+    root = root_for(gate)
+    if root is not None:
+        return root
+    cwd = getattr(event, 'cwd', None)
+    return _chock_Path(cwd) if cwd else _chock_Path.cwd()
+
 def writes_for(event, gate):
     """What this event puts under judgement: the call's own text, or what the turn left behind."""
     if event.event == PRE_TOOL:
         return writes_from_event(event)
     if (event.raw or {}).get('stop_hook_active'):
         return {}
-    root = root_for(gate)
-    return writes_from_worktree(root) if root is not None else {}
+    return writes_from_worktree(repo_root_for(event, gate))
 
 def evaluate_gate(argv, event):
     """The decision this event earns from a compiled gate, or None when it has nothing to say."""
@@ -868,7 +882,7 @@ def evaluate_gate(argv, event):
     writes = writes_for(event, gate)
     if not writes:
         return None
-    outcome, message = run_gate(gate, writes, name)
+    outcome, message = run_gate(gate, writes, name, repo_root_for(event, gate))
     if outcome == GATE_BLOCKED:
         return (VERDICT_DENY, message or f'Blocked by chock policy: {gate.parent.parent.name}')
     if outcome == GATE_ERRORED:

.chock/bin/devin.py

@@ -812,6 +812,8 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N
 
 _RUNNER_PARTS = ('bin', 'gate.py')
 
+_PACKAGED_RUNNER = 'gate.py'
+
 _GIT = 'git'
 
 _DELETED = 'D'
@@ -835,7 +837,10 @@ def gate_path_from_argv(argv):
     return None
 
 def runner_for(gate):
-    """The vendored gate runner beside this compiled gate, or None when it is not there."""
+    """The gate runner: beside the gate in a plugin, under .chock/bin in a repository, else None."""
+    packaged = gate.resolve().parent / _PACKAGED_RUNNER
+    if packaged.exists():
+        return packaged
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
@@ -887,13 +892,13 @@ def writes_from_worktree(repo_root):
             continue
     return writes
 
-def run_gate(gate, writes, event):
+def run_gate(gate, writes, event, root=None):
     """Ask the vendored runner. Returns (outcome, message) and never decides for itself."""
     runner = runner_for(gate)
     if runner is None:
         return (GATE_ERRORED, 'the vendored gate runner is not installed beside this gate')
     try:
-        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False)
+        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False, cwd=str(root) if root is not None else None)
     except (OSError, _chock_subprocess.SubprocessError) as exc:
         return (GATE_ERRORED, str(exc))
     if proc.returncode == 0:
@@ -911,16 +916,25 @@ def root_for(gate):
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
+    if parents[_GATE_DEPTH_TO_CHOCK - 1].name != 'compiled' or parents[_GATE_DEPTH_TO_CHOCK].name != '.chock':
+        return None
     return parents[_GATE_DEPTH_TO_CHOCK].parent
 
+def repo_root_for(event, gate):
+    """The repository under judgement: the compiled layout's root, else where the agent works."""
+    root = root_for(gate)
+    if root is not None:
+        return root
+    cwd = getattr(event, 'cwd', None)
+    return _chock_Path(cwd) if cwd else _chock_Path.cwd()
+
 def writes_for(event, gate):
     """What this event puts under judgement: the call's own text, or what the turn left behind."""
     if event.event == PRE_TOOL:
         return writes_from_event(event)
     if (event.raw or {}).get('stop_hook_active'):
         return {}
-    root = root_for(gate)
-    return writes_from_worktree(root) if root is not None else {}
+    return writes_from_worktree(repo_root_for(event, gate))
 
 def evaluate_gate(argv, event):
     """The decision this event earns from a compiled gate, or None when it has nothing to say."""
@@ -931,7 +945,7 @@ def evaluate_gate(argv, event):
     writes = writes_for(event, gate)
     if not writes:
         return None
-    outcome, message = run_gate(gate, writes, name)
+    outcome, message = run_gate(gate, writes, name, repo_root_for(event, gate))
     if outcome == GATE_BLOCKED:
         return (VERDICT_DENY, message or f'Blocked by chock policy: {gate.parent.parent.name}')
     if outcome == GATE_ERRORED:

.chock/bin/gate.py

@@ -454,6 +454,17 @@ def _log_outcome(gate_path: Path, event: str, spec: dict, result: GateResult) ->
 
 AGENT_EVENTS = ("pre-tool-use", "stop")
 
+#: `script_base` value naming the gate file's own directory as where `params.script` lives.
+SCRIPT_BASE_GATE = "gate"
+
+
+def _params(gate_path: Path, spec: dict) -> dict:
+    """The gate's params, with a packaged script gate's program located beside the gate file."""
+    params = dict(spec.get("params", {}))
+    if spec.get("script_base") == SCRIPT_BASE_GATE:
+        params["script"] = str(gate_path.resolve().parent / str(params.get("script", "")))
+    return params
+
 
 def _context(
     event: str,
@@ -518,7 +529,7 @@ def run(
             file=sys.stderr,
         )
         return 2
-    result = kind(ctx, spec.get("params", {}), name)
+    result = kind(ctx, _params(gate_path, spec), name)
     _log_outcome(gate_path, name, spec, result)
     if not result.allowed:
         print(result.message or spec.get("message", ""), file=sys.stderr)

.chock/bin/gemini_cli.py

@@ -812,6 +812,8 @@ def evaluate(argv: list[str], command: str, tool: str='') -> tuple[str, str] | N
 
 _RUNNER_PARTS = ('bin', 'gate.py')
 
+_PACKAGED_RUNNER = 'gate.py'
+
 _GIT = 'git'
 
 _DELETED = 'D'
@@ -835,7 +837,10 @@ def gate_path_from_argv(argv):
     return None
 
 def runner_for(gate):
-    """The vendored gate runner beside this compiled gate, or None when it is not there."""
+    """The gate runner: beside the gate in a plugin, under .chock/bin in a repository, else None."""
+    packaged = gate.resolve().parent / _PACKAGED_RUNNER
+    if packaged.exists():
+        return packaged
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
@@ -887,13 +892,13 @@ def writes_from_worktree(repo_root):
             continue
     return writes
 
-def run_gate(gate, writes, event):
+def run_gate(gate, writes, event, root=None):
     """Ask the vendored runner. Returns (outcome, message) and never decides for itself."""
     runner = runner_for(gate)
     if runner is None:
         return (GATE_ERRORED, 'the vendored gate runner is not installed beside this gate')
     try:
-        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False)
+        proc = _chock_subprocess.run([sys.executable, str(runner), 'run', '--gate', str(gate), '--event', event], input=json.dumps({'writes': writes}), capture_output=True, text=True, timeout=_GATE_TIMEOUT_SECONDS, check=False, cwd=str(root) if root is not None else None)
     except (OSError, _chock_subprocess.SubprocessError) as exc:
         return (GATE_ERRORED, str(exc))
     if proc.returncode == 0:
@@ -911,16 +916,25 @@ def root_for(gate):
     parents = gate.resolve().parents
     if len(parents) <= _GATE_DEPTH_TO_CHOCK:
         return None
+    if parents[_GATE_DEPTH_TO_CHOCK - 1].name != 'compiled' or parents[_GATE_DEPTH_TO_CHOCK].name != '.chock':
+        return None
     return parents[_GATE_DEPTH_TO_CHOCK].parent
 
+def repo_root_for(event, gate):
+    """The repository under judgement: the compiled layout's root, else where the agent works."""
+    root = root_for(gate)
+    if root is not None:
+        return root
+    cwd = getattr(event, 'cwd', None)
+    return _chock_Path(cwd) if cwd else _chock_Path.cwd()
+
 def writes_for(event, gate):
     """What this event puts under judgement: the call's own text, or what the turn left behind."""
     if event.event == PRE_TOOL:
         return writes_from_event(event)
     if (event.raw or {}).get('stop_hook_active'):
         return {}
-    root = root_for(gate)
-    return writes_from_worktree(root) if root is not None else {}
+    return writes_from_worktree(repo_root_for(event, gate))
 
 def evaluate_gate(argv, event):
     """The decision this event earns from a compiled gate, or None when it has nothing to say."""
@@ -931,7 +945,7 @@ def evaluate_gate(argv, event):
     writes = writes_for(event, gate)
     if not writes:
         return None
-    outcome, message = run_gate(gate, writes, name)
+    outcome, message = run_gate(gate, writes, name, repo_root_for(event, gate))
     if outcome == GATE_BLOCKED:
         return (VERDICT_DENY, message or f'Blocked by chock policy: {gate.parent.parent.name}')
     if outcome == GATE_ERRORED: