152 lines
5.9 KiB
Python
152 lines
5.9 KiB
Python
"""Configuration for the WriteGuard middleware.
|
|
|
|
The WriteGuard middleware intercepts external-write tool calls so they can be
|
|
recorded for human approval **before** the underlying tool executes. It is
|
|
activated per-run via a ``maf_run_id`` flag in the run context, so legacy
|
|
single-agent runs are completely unaffected (the middleware no-ops).
|
|
|
|
This module is harness-only: it defines the config schema and the
|
|
``WriteGuardPolicy`` / ``WriteGuardInterceptor`` Protocols. The concrete
|
|
implementations live in the application layer (``app.maf_roundtable``) and are
|
|
resolved at runtime via ``resolve_variable`` from the ``use`` dotted path.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Protocol, runtime_checkable
|
|
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
class WriteGuardProviderConfig(BaseModel):
|
|
"""Configuration for the application-layer WriteGuard implementations.
|
|
|
|
``policy_use`` resolves to a callable/class implementing
|
|
``WriteGuardPolicy``; ``interceptor_use`` resolves to one implementing
|
|
``WriteGuardInterceptor``. Both are resolved lazily by the harness via
|
|
``deerflow.reflection.resolve_variable`` so the harness never imports
|
|
``app.*``.
|
|
"""
|
|
|
|
policy_use: str = Field(
|
|
description="Dotted path to the WriteGuardPolicy factory, e.g. 'app.maf_roundtable.write_guard_bridge:make_policy'"
|
|
)
|
|
interceptor_use: str = Field(
|
|
description="Dotted path to the WriteGuardInterceptor factory, e.g. 'app.maf_roundtable.write_guard_bridge:make_interceptor'"
|
|
)
|
|
config: dict = Field(default_factory=dict, description="Implementation-specific settings passed as kwargs")
|
|
|
|
|
|
class WriteGuardConfig(BaseModel):
|
|
"""Configuration for the WriteGuard middleware.
|
|
|
|
When ``enabled`` and a ``provider`` is configured, the harness appends a
|
|
``WriteGuardMiddleware`` to the runtime middleware chain. The middleware
|
|
itself only acts when the run context carries a ``maf_run_id`` flag —
|
|
every other run is a pass-through.
|
|
"""
|
|
|
|
enabled: bool = Field(default=False, description="Enable the WriteGuard middleware in the chain")
|
|
provider: WriteGuardProviderConfig | None = Field(
|
|
default=None, description="Application-layer policy/interceptor factories"
|
|
)
|
|
|
|
|
|
_write_guard_config: WriteGuardConfig | None = None
|
|
|
|
|
|
def get_write_guard_config() -> WriteGuardConfig:
|
|
"""Get the write-guard config, returning defaults if not loaded."""
|
|
global _write_guard_config
|
|
if _write_guard_config is None:
|
|
_write_guard_config = WriteGuardConfig()
|
|
return _write_guard_config
|
|
|
|
|
|
def load_write_guard_config_from_dict(data: dict) -> WriteGuardConfig:
|
|
"""Load write-guard config from a dict (called during AppConfig loading)."""
|
|
global _write_guard_config
|
|
_write_guard_config = WriteGuardConfig.model_validate(data)
|
|
return _write_guard_config
|
|
|
|
|
|
def reset_write_guard_config() -> None:
|
|
"""Reset the cached config instance. Used in tests to prevent singleton leaks."""
|
|
global _write_guard_config
|
|
_write_guard_config = None
|
|
|
|
|
|
# ── Protocols (harness ↔ application contract) ─────────────────────────────
|
|
|
|
|
|
@runtime_checkable
|
|
class WriteGuardPolicy(Protocol):
|
|
"""Decides whether a given tool call must be intercepted for approval.
|
|
|
|
Implementations are constructed by the application layer and passed into
|
|
the harness ``WriteGuardMiddleware``. They must NOT import deerflow.*
|
|
(the harness owns this Protocol definition).
|
|
"""
|
|
|
|
def should_intercept(self, tool_name: str, args: Any, run_context: dict | None = None) -> bool: # noqa: ANN401
|
|
"""Return True if the tool call must be recorded for approval before execution.
|
|
|
|
``run_context`` (the shared run-context dict) lets the policy distinguish
|
|
reads from writes using the frozen capability snapshot + write policy, so
|
|
read tools are never intercepted (they pass straight through, including
|
|
on the sync path) while writes are gated by the active write policy.
|
|
"""
|
|
...
|
|
|
|
|
|
@runtime_checkable
|
|
class WriteGuardInterceptor(Protocol):
|
|
"""Records an intercepted write tool call (action + approval + audit event).
|
|
|
|
The interceptor is responsible for the full create-pending-action flow
|
|
(see ``store.create_pending_action_bundle``). It receives the run context
|
|
dict (the same one ``request.runtime.context`` exposes) so it can resolve
|
|
the current ``session_id`` / ``run_id`` / ``agent_id`` / ``turn_index``
|
|
and the write policy. It returns a dict describing the interception
|
|
outcome so the middleware can build the ToolMessage the model sees in
|
|
place of the real tool result.
|
|
|
|
Expected return shape::
|
|
|
|
{
|
|
"status": "approval_required" | "allowed",
|
|
"action_id": "<id when approval_required>",
|
|
"approval_id": "<id when approval_required>",
|
|
"tool_name": "<name>",
|
|
}
|
|
"""
|
|
|
|
async def intercept(
|
|
self,
|
|
*,
|
|
tool_name: str,
|
|
args: Any, # noqa: ANN401
|
|
tool_call_id: str,
|
|
run_context: dict,
|
|
) -> dict[str, Any]:
|
|
"""Record the tool call for approval. Never executes the real tool."""
|
|
...
|
|
|
|
async def validate_approved_action(
|
|
self,
|
|
*,
|
|
tool_name: str,
|
|
run_context: dict,
|
|
) -> bool:
|
|
"""Validate a post-approval deterministic execution before it runs.
|
|
|
|
Called on the async path when the run context binds ``maf_approved_tool``
|
|
+ ``maf_approved_action_id``. Must query the stored action and confirm:
|
|
its status is ``running`` (the executor already occupied it), it belongs
|
|
to the current ``run_id`` / ``session_id`` / ``agent_id``, and its
|
|
``tool_name`` matches. Returns True only when all hold — the middleware
|
|
fails closed (blocks the call) otherwise. Field-existence alone is NOT
|
|
sufficient.
|
|
"""
|
|
...
|