proactive_tools

proactive_tools

Proactive agent tools — check/record permissions, queue and execute actions.

These tools are used exclusively by ProactiveAgent to manage the propose → approve → execute lifecycle for autonomous actions.

Permission key convention: "{action_type}:{context_key}"

Approval response parsing

When the user replies to a pending-actions notification, their message is expected to contain one or more tokens of the form:

``{action_id} yes``   or   ``{action_id} no``
``yes {action_id}``   or   ``no {action_id}``
``always yes {action_id}``  →  approve + remember
``always no {action_id}``   →  deny + remember
``yes all``  /  ``no all``  →  bulk approve/deny all pending

Call parse_approval_response(text, store) from any channel message handler to process these replies without running the full agent.

Classes

CheckPermissionTool

CheckPermissionTool(store: Optional[ApprovalStore] = None)

Bases: BaseTool

Look up whether the user has a remembered decision for a permission key.

Source code in src/diapason/tools/proactive_tools.py

def __init__(self, store: Optional[ApprovalStore] = None) -> None:
    self._store = store

QueueActionTool

QueueActionTool(store: Optional[ApprovalStore] = None)

Bases: BaseTool

Queue a proposed action for user approval or immediate execution.

Source code in src/diapason/tools/proactive_tools.py

def __init__(self, store: Optional[ApprovalStore] = None) -> None:
    self._store = store

GetPendingActionsTool

GetPendingActionsTool(
    store: Optional[ApprovalStore] = None,
)

Bases: BaseTool

Return all pending (not yet decided) actions as a JSON list.

Source code in src/diapason/tools/proactive_tools.py

def __init__(self, store: Optional[ApprovalStore] = None) -> None:
    self._store = store

RecordDecisionTool

RecordDecisionTool(store: Optional[ApprovalStore] = None)

Bases: BaseTool

Record a user approval or denial for a queued action.

Source code in src/diapason/tools/proactive_tools.py

def __init__(self, store: Optional[ApprovalStore] = None) -> None:
    self._store = store

ExecutePendingActionsTool

ExecutePendingActionsTool(
    store: Optional[ApprovalStore] = None,
    executor_fn: Optional[Any] = None,
)

Bases: BaseTool

Execute all approved (or trivial) actions and return a summary.

Source code in src/diapason/tools/proactive_tools.py

def __init__(
    self,
    store: Optional[ApprovalStore] = None,
    executor_fn: Optional[Any] = None,
) -> None:
    self._store = store
    # executor_fn(action: PendingAction) -> (success: bool, message: str)
    self._executor_fn = executor_fn

Functions:

parse_approval_response

parse_approval_response(
    text: str, store: Optional[ApprovalStore] = None
) -> List[Dict[str, Any]]

Parse a free-text message for approval tokens and update the store.

Returns a list of dicts describing each decision that was processed, for use in an acknowledgement message back to the user.

Call this from any channel message handler before routing the message to the main agent, e.g. inside the iMessage daemon or Telegram bot.

Source code in src/diapason/tools/proactive_tools.py

def parse_approval_response(
    text: str,
    store: Optional[ApprovalStore] = None,
) -> List[Dict[str, Any]]:
    """Parse a free-text message for approval tokens and update the store.

    Returns a list of dicts describing each decision that was processed,
    for use in an acknowledgement message back to the user.

    Call this from any channel message handler before routing the message
    to the main agent, e.g. inside the iMessage daemon or Telegram bot.
    """
    s = store or get_store()
    processed: List[Dict[str, Any]] = []

    # Notification template displays ids as `[abc123]`; users naturally reply
    # with `{abc123} yes`, `(abc123) yes`, etc.  Strip those surrounding
    # brackets/braces/parens before regex matching so the word-boundary
    # check sees a clean id.
    text = re.sub(r"[\[\]\{\}\(\)]", " ", text)

    for m in _APPROVAL_RE.finditer(text):
        target = (m.group("target") or m.group("target2") or "").lower()
        raw_decision = (m.group("decision") or m.group("decision2") or "").lower()
        always = bool(m.group("always") or m.group("always2"))

        approved = raw_decision in ("yes", "approve")

        if target == "all":
            pending = s.list_pending()
            for action in pending:
                new_status = STATUS_APPROVED if approved else STATUS_DENIED
                s.update_status(action.id, new_status)
                if always and action.tier in (TIER_LOW, TIER_MEDIUM):
                    decision = (
                        DECISION_ALWAYS_APPROVE if approved else DECISION_ALWAYS_DENY
                    )
                    s.set_permission(action.permission_key, decision, approved=approved)
                processed.append(
                    {"id": action.id, "approved": approved, "remembered": always}
                )
        else:
            action = s.get_action(target)
            if action is None:
                continue
            new_status = STATUS_APPROVED if approved else STATUS_DENIED
            s.update_status(target, new_status)
            remember = always and action.tier in (TIER_LOW, TIER_MEDIUM)
            if remember:
                decision = DECISION_ALWAYS_APPROVE if approved else DECISION_ALWAYS_DENY
                s.set_permission(action.permission_key, decision, approved=approved)
            processed.append(
                {"id": target, "approved": approved, "remembered": remember}
            )

    return processed