From 6561dd674329140378b56b82dbdeebf3235fbf6e Mon Sep 17 00:00:00 2001 From: ciregenz Date: Wed, 24 Jun 2026 22:36:17 -0700 Subject: [PATCH] [eric] agents: convention fixes on merge surface (public cross-file helpers, pydantic approval memory, resolve_ask @typechecked) --- .../permissions/build_effective_tool_lists.py | 2 +- .../agents/manager/permissions/gate_hooks.py | 26 ++++++------ .../agents/manager/permissions/path_gate.py | 4 +- .../manager/permissions/workflow_approval.py | 41 +++++++++---------- .../manager/session/history_compaction.py | 4 +- backend/tests/test_gate_hooks.py | 2 +- backend/tests/test_schedule_gate.py | 12 +++--- backend/tests/test_swarm_bundle.py | 6 ++- 8 files changed, 49 insertions(+), 48 deletions(-) diff --git a/backend/apps/agents/manager/permissions/build_effective_tool_lists.py b/backend/apps/agents/manager/permissions/build_effective_tool_lists.py index f4dc9c45..3d6eaf77 100644 --- a/backend/apps/agents/manager/permissions/build_effective_tool_lists.py +++ b/backend/apps/agents/manager/permissions/build_effective_tool_lists.py @@ -101,7 +101,7 @@ def build_effective_tool_lists( effective_disallowed.append(wt_name) # Claude's internal Cron* scheduler is denied in favour of the visible native # one; withhold it from the SDK so the model doesn't even reach for it. - for bt in path_gate.p_CLAUDE_INTERNAL_SCHEDULER_TOOLS: + for bt in path_gate.CLAUDE_INTERNAL_SCHEDULER_TOOLS: if bt not in effective_disallowed: effective_disallowed.append(bt) return effective_allowed, effective_disallowed diff --git a/backend/apps/agents/manager/permissions/gate_hooks.py b/backend/apps/agents/manager/permissions/gate_hooks.py index 486577bc..b038eb31 100644 --- a/backend/apps/agents/manager/permissions/gate_hooks.py +++ b/backend/apps/agents/manager/permissions/gate_hooks.py @@ -23,9 +23,9 @@ from backend.apps.agents.manager.prompt.prompt_context import ( ) from backend.apps.agents.manager.streaming.HookContext import HookContext from backend.apps.agents.manager.permissions.workflow_approval import ( - p_is_claude_schedule_skill, - p_note_tool_used, - p_resolve_ask, + is_claude_schedule_skill, + note_tool_used, + resolve_ask, ) logger = logging.getLogger(__name__) @@ -35,8 +35,8 @@ logger = logging.getLogger(__name__) async def can_use_tool( ctx: HookContext, tool_name: str, input_data: object, context: object ) -> Union[PermissionResultAllow, PermissionResultDeny]: - if p_is_claude_schedule_skill(tool_name, input_data): - p_note_tool_used(ctx.session_id, tool_name, False) + if is_claude_schedule_skill(tool_name, input_data): + note_tool_used(ctx.session_id, tool_name, False) return PermissionResultDeny( message="Use the openswarm-schedule MCP tools instead of Claude's internal schedule skill." ) @@ -46,13 +46,13 @@ async def can_use_tool( effective_policy(tool_name, ctx.builtin_perms, ctx.policy_defaults), tool_name, input_data ) if policy == "always_allow": - p_note_tool_used(ctx.session_id, tool_name, True) + note_tool_used(ctx.session_id, tool_name, True) return PermissionResultAllow(updated_input=input_data) if policy == "deny": - p_note_tool_used(ctx.session_id, tool_name, False) + note_tool_used(ctx.session_id, tool_name, False) return PermissionResultDeny(message="Tool denied by permission policy") - decision = await p_resolve_ask(ctx, tool_name, input_data, sensitive_pattern) + decision = await resolve_ask(ctx, tool_name, input_data, sensitive_pattern) if decision.behavior == "allow": return PermissionResultAllow( updated_input=decision.updated_input if decision.updated_input is not None else input_data @@ -136,8 +136,8 @@ async def pre_tool_hook(ctx: HookContext, input_data: dict, tool_use_id: Optiona if tool_name and tool_name != "AskUserQuestion": tool_input = input_data.get("tool_input", {}) - if p_is_claude_schedule_skill(tool_name, tool_input): - p_note_tool_used(ctx.session_id, tool_name, False) + if is_claude_schedule_skill(tool_name, tool_input): + note_tool_used(ctx.session_id, tool_name, False) return { "hookSpecificOutput": { "hookEventName": hook_event, @@ -150,10 +150,10 @@ async def pre_tool_hook(ctx: HookContext, input_data: dict, tool_use_id: Optiona ) if policy == "always_allow": - p_note_tool_used(ctx.session_id, tool_name, True) + note_tool_used(ctx.session_id, tool_name, True) if policy == "deny": - p_note_tool_used(ctx.session_id, tool_name, False) + note_tool_used(ctx.session_id, tool_name, False) return { "hookSpecificOutput": { "hookEventName": hook_event, @@ -163,7 +163,7 @@ async def pre_tool_hook(ctx: HookContext, input_data: dict, tool_use_id: Optiona } if policy == "ask": - decision = await p_resolve_ask(ctx, tool_name, tool_input, sensitive_pattern) + decision = await resolve_ask(ctx, tool_name, tool_input, sensitive_pattern) if decision.behavior == "allow": if tool_use_id: diff --git a/backend/apps/agents/manager/permissions/path_gate.py b/backend/apps/agents/manager/permissions/path_gate.py index c98ea8b4..c3a4e055 100644 --- a/backend/apps/agents/manager/permissions/path_gate.py +++ b/backend/apps/agents/manager/permissions/path_gate.py @@ -170,7 +170,7 @@ p_SCHEDULE_GATED = { "mcp__openswarm-schedule__DeleteScheduledWorkflow", "mcp__openswarm-schedule__PauseAllWorkflows", } -p_CLAUDE_INTERNAL_SCHEDULER_TOOLS = ("CronCreate", "CronList", "CronDelete") +CLAUDE_INTERNAL_SCHEDULER_TOOLS = ("CronCreate", "CronList", "CronDelete") @typechecked @@ -182,7 +182,7 @@ def maybe_override_policy(policy: str, tool_name: str, tool_input: object) -> Tu future writes to it pass through silently.""" if tool_name == "Bash" and looks_like_os_scheduling(tool_input): return "ask", None - if tool_name in p_CLAUDE_INTERNAL_SCHEDULER_TOOLS: + if tool_name in CLAUDE_INTERNAL_SCHEDULER_TOOLS: return "deny", None # Committing or mutating a native recurring schedule is the in-app twin of the # crontab gate above: real, user-visible, hard-to-undo, so it goes through diff --git a/backend/apps/agents/manager/permissions/workflow_approval.py b/backend/apps/agents/manager/permissions/workflow_approval.py index e5d3597e..38410747 100644 --- a/backend/apps/agents/manager/permissions/workflow_approval.py +++ b/backend/apps/agents/manager/permissions/workflow_approval.py @@ -10,6 +10,7 @@ agent_manager re-exports the setters for the executor's convenience. import logging from typing import Callable, Dict, Optional +from pydantic import BaseModel, ConfigDict from typeguard import typechecked from backend.apps.agents.manager.permissions.ApprovalDecision import ApprovalDecision @@ -19,23 +20,18 @@ from backend.apps.agents.manager.streaming.HookContext import HookContext logger = logging.getLogger(__name__) -class WorkflowApprovalMemory: +class WorkflowApprovalMemory(BaseModel): """A workflow run's approval context, pushed in by the executor.""" - def __init__( - self, - decisions: Dict[str, str], - step_usage: Dict[str, Dict[str, bool]], - remember: Optional[Callable[[str, str], None]], - ask_timeout: float, - ) -> None: - self.decisions = decisions # workflow-level: tool -> "allow"/"deny" - self.step_usage = step_usage # per-step record: step_id -> {tool: approved} - self.remember = remember # persist a workflow-level decision to disk - self.ask_timeout = ask_timeout - # The executor bumps this as it advances steps so the gate can record - # which tools each step touched. None on test runs that don't thread it. - self.current_step_id: Optional[str] = None + model_config = ConfigDict(validate_assignment=True) + + decisions: Dict[str, str] # workflow-level: tool -> "allow"/"deny" + step_usage: Dict[str, Dict[str, bool]] # per-step record: step_id -> {tool: approved} + remember: Optional[Callable[[str, str], None]] # persist a workflow-level decision to disk + ask_timeout: float + # The executor bumps this as it advances steps so the gate can record which + # tools each step touched. None on test runs that don't thread it. + current_step_id: Optional[str] = None p_approval_memory: Dict[str, WorkflowApprovalMemory] = {} @@ -51,7 +47,7 @@ def set_workflow_approval_memory( ask_timeout: float, ) -> None: p_approval_memory[session_id] = WorkflowApprovalMemory( - decisions, step_usage, remember, ask_timeout + decisions=decisions, step_usage=step_usage, remember=remember, ask_timeout=ask_timeout ) @@ -74,14 +70,14 @@ def get_workflow_step_usage(session_id: str) -> Dict[str, Dict[str, bool]]: @typechecked -def p_is_claude_schedule_skill(tool_name: str, tool_input: object) -> bool: +def is_claude_schedule_skill(tool_name: str, tool_input: object) -> bool: if tool_name != "Skill" or not isinstance(tool_input, dict): return False return str(tool_input.get("skill") or "").strip().lower() == "schedule" @typechecked -def p_note_tool_used(session_id: str, tool_name: str, approved: bool) -> None: +def note_tool_used(session_id: str, tool_name: str, approved: bool) -> None: # Record which tools each step touched (in-memory; the executor/test path # persists step_usage once at run end). Captures every tool the gate sees so # a step's tool set is complete, not only the ones that prompted. @@ -91,7 +87,8 @@ def p_note_tool_used(session_id: str, tool_name: str, approved: bool) -> None: mem.step_usage.setdefault(mem.current_step_id, {})[tool_name] = approved -async def p_resolve_ask( +@typechecked +async def resolve_ask( ctx: HookContext, tool_name: str, tool_input: object, sensitive_pattern: Optional[str] ) -> ApprovalDecision: """Resolve an 'ask' policy. On a workflow run, reuse a remembered decision @@ -114,10 +111,10 @@ async def p_resolve_ask( return ApprovalDecision(behavior="deny", message="Denied by a remembered workflow permission") prior = mem.decisions.get(tool_name) if prior == "allow": - p_note_tool_used(ctx.session_id, tool_name, True) + note_tool_used(ctx.session_id, tool_name, True) return ApprovalDecision(behavior="allow") if prior == "deny": - p_note_tool_used(ctx.session_id, tool_name, False) + note_tool_used(ctx.session_id, tool_name, False) return ApprovalDecision(behavior="deny", message="Denied by a remembered workflow permission") timeout = mem.ask_timeout if mem is not None else 600.0 decision = await request_user_approval( @@ -127,7 +124,7 @@ async def p_resolve_ask( if rememberable and decision.behavior in ("allow", "deny"): behavior = decision.behavior mem.decisions[tool_name] = behavior - p_note_tool_used(ctx.session_id, tool_name, behavior == "allow") + note_tool_used(ctx.session_id, tool_name, behavior == "allow") if mem.remember: try: mem.remember(tool_name, behavior) diff --git a/backend/apps/agents/manager/session/history_compaction.py b/backend/apps/agents/manager/session/history_compaction.py index 46246642..1f191328 100644 --- a/backend/apps/agents/manager/session/history_compaction.py +++ b/backend/apps/agents/manager/session/history_compaction.py @@ -34,7 +34,7 @@ def wrap_platform_note(body: str) -> str: return f"{PLATFORM_NOTE_OPEN}\n{PLATFORM_NOTE_PREAMBLE}\n{body}\n{PLATFORM_NOTE_CLOSE}" -_SENTINEL_TAG_RE = re.compile(r"]*>") +P_SENTINEL_TAG_RE = re.compile(r"]*>") def strip_forged_sentinels(text: str) -> str: @@ -42,7 +42,7 @@ def strip_forged_sentinels(text: str) -> str: user input) so attacker-supplied content can't pose as trusted platform context.""" if "openswarm_platform_note" not in text and "openswarm_session_recap" not in text: return text - return _SENTINEL_TAG_RE.sub(lambda m: m.group(0).replace("<", "<").replace(">", ">"), text) + return P_SENTINEL_TAG_RE.sub(lambda m: m.group(0).replace("<", "<").replace(">", ">"), text) def p_recap_tool_call_line(content: object) -> str: diff --git a/backend/tests/test_gate_hooks.py b/backend/tests/test_gate_hooks.py index 29ce1561..bfd36724 100644 --- a/backend/tests/test_gate_hooks.py +++ b/backend/tests/test_gate_hooks.py @@ -46,7 +46,7 @@ async def test_can_use_tool_deny_returns_deny(): async def test_can_use_tool_ask_routes_through_approval(): ctx = p_ctx() with patch.object(gate_hooks.path_gate, "maybe_override_policy", return_value=("ask", None)), \ - patch.object(gate_hooks, "p_resolve_ask", new=AsyncMock(return_value=ApprovalDecision(behavior="allow"))): + patch.object(gate_hooks, "resolve_ask", new=AsyncMock(return_value=ApprovalDecision(behavior="allow"))): result = await gate_hooks.can_use_tool(ctx, "Write", {"file_path": "/x"}, None) assert isinstance(result, PermissionResultAllow) diff --git a/backend/tests/test_schedule_gate.py b/backend/tests/test_schedule_gate.py index 514f27fc..37a58a05 100644 --- a/backend/tests/test_schedule_gate.py +++ b/backend/tests/test_schedule_gate.py @@ -5,7 +5,7 @@ the scheduled-tasks PR shipped without a test, and the exact path most at risk i agent-manager decomposition (the gating moved from agent_manager into path_gate).""" from backend.apps.agents.manager.permissions import path_gate -from backend.apps.agents.manager.permissions.workflow_approval import p_is_claude_schedule_skill +from backend.apps.agents.manager.permissions.workflow_approval import is_claude_schedule_skill def test_schedule_commit_tools_force_ask_even_when_always_allow(): @@ -26,11 +26,11 @@ def test_claude_internal_cron_tools_denied(): def test_claude_schedule_skill_detected(): - assert p_is_claude_schedule_skill("Skill", {"skill": "schedule"}) - assert p_is_claude_schedule_skill("Skill", {"skill": "Schedule"}) - assert not p_is_claude_schedule_skill("Skill", {"skill": "other"}) - assert not p_is_claude_schedule_skill("Bash", {"skill": "schedule"}) - assert not p_is_claude_schedule_skill("Skill", "not a dict") + assert is_claude_schedule_skill("Skill", {"skill": "schedule"}) + assert is_claude_schedule_skill("Skill", {"skill": "Schedule"}) + assert not is_claude_schedule_skill("Skill", {"skill": "other"}) + assert not is_claude_schedule_skill("Bash", {"skill": "schedule"}) + assert not is_claude_schedule_skill("Skill", "not a dict") def test_normal_tool_unaffected_by_schedule_gate(): diff --git a/backend/tests/test_swarm_bundle.py b/backend/tests/test_swarm_bundle.py index 1c341560..0a7a9d90 100644 --- a/backend/tests/test_swarm_bundle.py +++ b/backend/tests/test_swarm_bundle.py @@ -166,6 +166,7 @@ def test_workflow_round_trips_through_the_store(isolated_workflows_data): # from before the workflow store was on eric/dev. from backend.apps.swarm.entities.workflows import WorkflowExportable from backend.apps.swarm.exportable import RemapTable + from backend.apps.workflows import storage assert WorkflowExportable.load("nonexistent") is None new_id = WorkflowExportable.import_( {"title": "Shared WF", "schedule": {"enabled": True}}, {}, RemapTable() @@ -174,7 +175,10 @@ def test_workflow_round_trips_through_the_store(isolated_workflows_data): loaded = WorkflowExportable.load(new_id) assert loaded is not None assert loaded.name == "Shared WF" - assert loaded.p_data["schedule"]["enabled"] is False + # Read the persisted row back through the store's public API (not the entity's + # private data) to confirm the schedule was forced off on import. + saved = storage.get_workflow(new_id) + assert saved is not None and saved.schedule.enabled is False def test_session_export_carries_transcript_drops_runtime_and_secrets():