Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions coworker/automation/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -183,6 +183,22 @@ def add_rule(self, tool: str, target: str) -> bool:
self.always_allowed_tools.append(entry)
return True

def add_tool_rule(self, tool: str) -> bool:
"""Grant a tool for this task with no target binding — "this routine may search
the web", as opposed to "…may message this one chat".

Needed because a target binding is not always available to bind. A tool that
takes no target argument can never earn a standing rule through add_rule, so a
routine using one asks again on every single run: the session-scoped "always"
dies with the run, and a scheduled task is a fresh session each time. Callers
gate which tools may take this path (see standing_tool_candidate).
"""
entry = rule_entry(tool)
if not tool or entry in self.always_allowed_tools:
return False
self.always_allowed_tools.append(entry)
return True

def revoke_rule(self, entry: str) -> bool:
if entry in self.always_allowed_tools:
self.always_allowed_tools.remove(entry)
Expand Down
20 changes: 20 additions & 0 deletions coworker/permissions.py
Original file line number Diff line number Diff line change
Expand Up @@ -278,6 +278,26 @@ def standing_rule_candidate(
return value or None


def standing_tool_candidate(
tool_name: str,
metadata: Any = None,
overrides: Optional[RiskOverrides] = None,
) -> bool:
"""Whether this tool may earn a task-scoped standing rule with NO target binding.

Egress only. The reasoning is the strictness table itself: egress (1) ranks below
external (2), and an external call already earns a standing rule when it names a
target. Refusing the lower class outright meant a routine could be trusted forever
to message a chat, but never to read a web page — so a daily sweep asked the same
question every morning and "Always" had nothing to bind to.

Write-local and exec stay out, as they do for target-bound rules: shell asks
forever. External stays target-bound — the target IS the safety there, and a
blanket "send anything anywhere" is a different proposition entirely.
"""
return classify(tool_name, metadata, overrides) is RiskClass.EGRESS


@dataclass
class PermissionEngine:
workspace_root: Path
Expand Down
31 changes: 24 additions & 7 deletions coworker/server/manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -4267,21 +4267,34 @@ def mint_task_rule(
) -> bool:
"""Persist a standing rule a human minted via "Allow every time" on a run's
approval card (§25's retrofit path). Server-side validation, not trust in the
card: the session must be an automation run and the call must be rule-eligible
(external risk, declared target argument, non-empty target). Also applies the
rule to the live engine so the run's next call auto-allows."""
from ..permissions import standing_rule_candidate
card: the session must be an automation run, and the call either names a target
(external risk, declared target argument) or is a tool eligible for a
target-less grant (egress — see standing_tool_candidate). Also applies the rule
to the live engine so the run's next call auto-allows."""
from ..permissions import standing_rule_candidate, standing_tool_candidate

task = self.task_store.task_for_run_session(session_id)
if task is None:
return False
target = standing_rule_candidate(tool_name, arguments or {}, metadata)
if not target or not task.add_rule(tool_name, target):
if target:
minted = task.add_rule(tool_name, target)
elif standing_tool_candidate(tool_name, metadata):
# No target to bind (web_search takes a query, not a destination). Grant the
# tool for this task instead of downgrading to a one-off — otherwise the
# routine asks the same question on every run and "Always" means nothing.
target, minted = None, task.add_tool_rule(tool_name)
else:
return False
if not minted:
return False
self.task_store.save(task)
engine = self._engines.get(session_id)
if engine is not None:
engine.permissions.task_rules.setdefault(tool_name, set()).add(target)
if target is None:
engine.permissions.allow_tool_for_session(tool_name)
else:
engine.permissions.task_rules.setdefault(tool_name, set()).add(target)
try:
self.audit_store.append(
{
Expand All @@ -4290,7 +4303,11 @@ def mint_task_rule(
"arguments": arguments or {},
"stage": "standing_rule_minted",
"status": "granted",
"reason": f"allow every time: {tool_name} → {target} (task {task.id})",
"reason": (
f"allow every time: {tool_name} → {target} (task {task.id})"
if target
else f"allow every time: {tool_name}, any arguments (task {task.id})"
),
}
)
except Exception:
Expand Down
80 changes: 80 additions & 0 deletions tests/test_task_tool_grants.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,80 @@
"""A routine can be trusted with a tool that has no target to bind.

"Always allow" on a run's approval card mints a standing rule on the task. That only
worked for external-risk calls naming a target, so a routine whose work is web search
asked the same question on every run: the session-scoped "always" dies with the run,
and each scheduled run is a fresh session. Egress ranks *below* external in the
engine's own strictness table, so the rule was being refused for the safer class.
"""

from __future__ import annotations

from coworker.automation.models import ScheduledTask, Schedule
from coworker.permissions import standing_tool_candidate
from coworker.providers import ModelCapabilities, ProviderClient
from coworker.server.manager import SessionManager


class NoTurnsProvider(ProviderClient):
def complete(self, *, model, messages, tools=None, **settings):
raise AssertionError("no model turns expected")

def capabilities(self, model):
return ModelCapabilities()


class _Request:
def __init__(self, tool_name, arguments=None):
self.tool_name = tool_name
self.arguments = arguments or {}
self.metadata = None


def test_only_egress_earns_a_target_less_grant():
assert standing_tool_candidate("web_search") is True
assert standing_tool_candidate("web_fetch") is True
# Exec and local writes ask forever, as they do for target-bound rules.
assert standing_tool_candidate("run_shell") is False
assert standing_tool_candidate("write_file") is False


def _task_manager(tmp_path):
manager = SessionManager(data_dir=tmp_path / "data", provider=NoTurnsProvider())
task = ScheduledTask(
title="Daily watch",
instructions="sweep",
agent="watcher",
workspace=str(tmp_path / "work"),
schedule=Schedule(kind="cron", cron="0 7 * * *"),
)
manager.task_store.save(task)
run_session = task.task_session_id
return manager, task, run_session


def test_web_search_grant_persists_on_the_task(tmp_path):
manager, task, _ = _task_manager(tmp_path)
from coworker.automation.models import TaskRun

run = TaskRun(task_id=task.id)
manager.task_store.add_run(run)

assert manager.mint_task_rule(run.session_id, "web_search", {"query": "anything"})

fresh = manager.task_store.get(task.id)
assert "web_search" in fresh.name_allowed_tools()
# A name-only entry is revocable like any other, and reads back on the API shape.
entry = [r for r in fresh.public()["always_allowed"] if r["tool"] == "web_search"][0]
assert entry["target"] is None
assert fresh.revoke_rule(entry["entry"]) is True


def test_shell_still_asks_every_time(tmp_path):
manager, task, _ = _task_manager(tmp_path)
from coworker.automation.models import TaskRun

run = TaskRun(task_id=task.id)
manager.task_store.add_run(run)

assert manager.mint_task_rule(run.session_id, "run_shell", {"command": "ls"}) is False
assert manager.task_store.get(task.id).always_allowed_tools == []