Add immutable AgentCapabilities core data models

.agents/skills/python-design-patterns/SKILL.md

@@ -41,6 +41,7 @@ Wait until you have three instances before abstracting. Duplication is often bet
 # Instead of a factory/registry pattern:
 FORMATTERS = {"json": JsonFormatter, "csv": CsvFormatter}
 
+
 def get_formatter(name: str) -> Formatter:
     return FORMATTERS[name]()
 ```

.agents/skills/python-design-patterns/references/details.md

@@ -16,15 +16,17 @@ class OutputFormatterFactory:
         def decorator(formatter_cls):
             cls._formatters[name] = formatter_cls
             return formatter_cls
+
         return decorator
 
     @classmethod
     def create(cls, name: str) -> Formatter:
         return cls._formatters[name]()
 
+
 @OutputFormatterFactory.register("json")
-class JsonFormatter(Formatter):
-    ...
+class JsonFormatter(Formatter): ...
+
 
 # Simple: Just use a dictionary
 FORMATTERS = {
@@ -33,6 +35,7 @@ FORMATTERS = {
     "xml": XmlFormatter,
 }
 
+
 def get_formatter(name: str) -> Formatter:
     """Get formatter by name."""
     if name not in FORMATTERS:
@@ -60,12 +63,14 @@ class UserHandler:
         # Database access
         user = await db.execute(
             "INSERT INTO users (email, name) VALUES ($1, $2) RETURNING *",
-            data["email"], data["name"]
+            data["email"],
+            data["name"],
         )
 
         # Response formatting
         return Response({"id": user.id, "email": user.email}, status=201)
 
+
 # GOOD: Separated concerns
 class UserService:
     """Business logic only."""
@@ -78,6 +83,7 @@ class UserService:
         user = User(email=data.email, name=data.name)
         return await self._repo.save(user)
 
+
 class UserHandler:
     """HTTP concerns only."""
 
@@ -127,11 +133,10 @@ Each layer depends only on layers below it:
 # Repository: Data access
 class UserRepository:
     async def get_by_id(self, user_id: str) -> User | None:
-        row = await self._db.fetchrow(
-            "SELECT * FROM users WHERE id = $1", user_id
-        )
+        row = await self._db.fetchrow("SELECT * FROM users WHERE id = $1", user_id)
         return User(**row) if row else None
 
+
 # Service: Business logic
 class UserService:
     def __init__(self, repo: UserRepository) -> None:
@@ -143,6 +148,7 @@ class UserService:
             raise UserNotFoundError(user_id)
         return user
 
+
 # Handler: HTTP concerns
 @app.get("/users/{user_id}")
 async def get_user(user_id: str) -> UserResponse:
@@ -164,6 +170,7 @@ class EmailNotificationService(NotificationService):
     def notify(self, user: User, message: str) -> None:
         self._smtp.send(user.email, message)
 
+
 # Composition: Flexible and testable
 class NotificationService:
     """Send notifications via multiple channels."""
@@ -195,6 +202,7 @@ class NotificationService:
         if "push" in channels and self._push and user.device_token:
             await self._push.send(user.device_token, message)
 
+
 # Easy to test with fakes
 service = NotificationService(
     email_sender=FakeEmailSender(),
@@ -218,6 +226,7 @@ def process_orders(orders: list[Order]) -> list[Result]:
         results.append(result)
     return results
 
+
 def process_returns(returns: list[Return]) -> list[Result]:
     results = []
     for ret in returns:
@@ -226,6 +235,7 @@ def process_returns(returns: list[Return]) -> list[Result]:
         results.append(result)
     return results
 
+
 # These look similar, but wait! Are they actually the same?
 # Different validation, different processing, different errors...
 # Duplication is often better than the wrong abstraction
@@ -251,6 +261,7 @@ def process_order(order: Order) -> Result:
     # 20 lines of notification...
     pass
 
+
 # Better: Composed from focused functions
 def process_order(order: Order) -> Result:
     """Process a customer order through the complete workflow."""
@@ -268,14 +279,17 @@ Pass dependencies through constructors for testability.
 ```python
 from typing import Protocol
 
+
 class Logger(Protocol):
     def info(self, msg: str, **kwargs) -> None: ...
     def error(self, msg: str, **kwargs) -> None: ...
 
+
 class Cache(Protocol):
     async def get(self, key: str) -> str | None: ...
     async def set(self, key: str, value: str, ttl: int) -> None: ...
 
+
 class UserService:
     """Service with injected dependencies."""
 
@@ -303,6 +317,7 @@ class UserService:
 
         return user
 
+
 # Production
 service = UserService(
     repository=PostgresUserRepository(db),
@@ -328,6 +343,7 @@ service = UserService(
 def get_user(id: str) -> UserModel:  # SQLAlchemy model
     return db.query(UserModel).get(id)
 
+
 # GOOD: Use response schemas
 @app.get("/users/{id}")
 def get_user(id: str) -> UserResponse:
@@ -344,6 +360,7 @@ def calculate_discount(user_id: str) -> float:
     orders = db.query("SELECT * FROM orders WHERE user_id = ?", user_id)
     # Business logic mixed with data access
 
+
 # GOOD: Repository pattern
 def calculate_discount(user: User, order_history: list[Order]) -> float:
     # Pure business logic, easily testable

.agents/skills/python-observability/SKILL.md

@@ -60,6 +60,7 @@ Configure structlog for JSON output with consistent fields.
 import logging
 import structlog
 
+
 def configure_logging(log_level: str = "INFO") -> None:
     """Configure structured logging for the application."""
     structlog.configure(
@@ -71,14 +72,13 @@ def configure_logging(log_level: str = "INFO") -> None:
             structlog.processors.format_exc_info,
             structlog.processors.JSONRenderer(),
         ],
-        wrapper_class=structlog.make_filtering_bound_logger(
-            getattr(logging, log_level.upper())
-        ),
+        wrapper_class=structlog.make_filtering_bound_logger(getattr(logging, log_level.upper())),
         context_class=dict,
         logger_factory=structlog.PrintLoggerFactory(),
         cache_logger_on_first_use=True,
     )
 
+
 # Initialize at application startup
 configure_logging("INFO")
 logger = structlog.get_logger()
@@ -97,6 +97,7 @@ correlation_id: ContextVar[str] = ContextVar("correlation_id", default="")
 
 logger = structlog.get_logger()
 
+
 def process_request(request: Request) -> Response:
     """Process request with structured logging."""
     logger.info(
@@ -174,16 +175,19 @@ import structlog
 
 correlation_id: ContextVar[str] = ContextVar("correlation_id", default="")
 
+
 def set_correlation_id(cid: str | None = None) -> str:
     """Set correlation ID for current context."""
     cid = cid or str(uuid.uuid4())
     correlation_id.set(cid)
     structlog.contextvars.bind_contextvars(correlation_id=cid)
     return cid
 
+
 # FastAPI middleware example
 from fastapi import Request
 
+
 async def correlation_middleware(request: Request, call_next):
     """Middleware to set and propagate correlation ID."""
     # Use incoming header or generate new
@@ -200,6 +204,7 @@ Propagate to outbound requests:
 ```python
 import httpx
 
+
 async def call_downstream_service(endpoint: str, data: dict) -> dict:
     """Call downstream service with correlation ID."""
     async with httpx.AsyncClient() as client:

.agents/skills/python-observability/references/details.md

@@ -44,8 +44,10 @@ Instrument your endpoints:
 import time
 from functools import wraps
 
+
 def track_request(func):
     """Decorator to track request metrics."""
+
     @wraps(func)
     async def wrapper(request: Request, *args, **kwargs):
         method = request.method
@@ -67,7 +69,9 @@ def track_request(func):
         finally:
             duration = time.perf_counter() - start
             REQUEST_COUNT.labels(method=method, endpoint=endpoint, status=status).inc()
-            REQUEST_LATENCY.labels(method=method, endpoint=endpoint, status=status).observe(duration)
+            REQUEST_LATENCY.labels(method=method, endpoint=endpoint, status=status).observe(
+                duration
+            )
 
     return wrapper
 ```
@@ -105,6 +109,7 @@ import structlog
 
 logger = structlog.get_logger()
 
+
 @contextmanager
 def timed_operation(name: str, **extra_fields):
     """Context manager for timing and logging operations."""
@@ -132,6 +137,7 @@ def timed_operation(name: str, **extra_fields):
             **extra_fields,
         )
 
+
 # Usage
 with timed_operation("fetch_user_orders", user_id=user.id):
     orders = await order_repository.get_by_user(user.id)
@@ -149,15 +155,18 @@ from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
 
+
 def configure_tracing(service_name: str, otlp_endpoint: str) -> None:
     """Configure OpenTelemetry tracing."""
     provider = TracerProvider()
     processor = BatchSpanProcessor(OTLPSpanExporter(endpoint=otlp_endpoint))
     provider.add_span_processor(processor)
     trace.set_tracer_provider(provider)
 
+
 tracer = trace.get_tracer(__name__)
 
+
 async def process_order(order_id: str) -> Order:
     """Process order with tracing."""
     with tracer.start_as_current_span("process_order") as span:

README.md

@@ -281,8 +281,8 @@ from agentwatch.reasoning.auditor import ReasoningAuditor
 auditor = ReasoningAuditor()
 audit = await auditor.audit_step(step.step_number, step)
 
-print(audit.score)        # 0.0 – 1.0 confidence in the step
-print(audit.rationale)    # why the auditor scored it this way
+print(audit.score)  # 0.0 – 1.0 confidence in the step
+print(audit.rationale)  # why the auditor scored it this way
 ```
 
 When the score drops below your threshold, the next action is **held — not logged after the fact.** An alert fires. You decide what happens next.

agentwatch/core/__init__.py

@@ -1,3 +1,4 @@
+from agentwatch.core.capabilities import AgentCapabilities, Capability
 from agentwatch.core.event_bus import EventBus, EventFilter, get_event_bus
 from agentwatch.core.safety import (
     DEFAULT_POLICY,

agentwatch/core/capabilities.py

@@ -0,0 +1,25 @@
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from pathlib import Path
+
+
+@dataclass(frozen=True)
+class Capability:
+    """Represents a single capability permission."""
+
+    name: str
+
+
+@dataclass(frozen=True)
+class AgentCapabilities:
+    """Immutable collection of agent capabilities."""
+
+    read_paths: frozenset[Path] = field(default_factory=frozenset)
+    write_paths: frozenset[Path] = field(default_factory=frozenset)
+    network_domains: frozenset[str] = field(default_factory=frozenset)
+    db_tables: frozenset[tuple[str, str]] = field(default_factory=frozenset)
+    exec_binaries: frozenset[str] = field(default_factory=frozenset)
+
+
+__all__ = ["Capability", "AgentCapabilities"]

docs/adapters/langchain.md

@@ -18,19 +18,16 @@ from langchain.agents import AgentExecutor
 
 # Initialize the handler
 handler = AgentWatchCallbackHandler(
-    session_id="my-session-123", # Optional: defaults to a new UUID
-    agent_id="my-agent-456"      # Optional: helpful for multi-agent setups
+    session_id="my-session-123",  # Optional: defaults to a new UUID
+    agent_id="my-agent-456",  # Optional: helpful for multi-agent setups
 )
 
 # Use with AgentExecutor
-agent = AgentExecutor(
-    agent=..., 
-    tools=..., 
-    callbacks=[handler]
-)
+agent = AgentExecutor(agent=..., tools=..., callbacks=[handler])
 
 # Use with an LLM directly
 from langchain_openai import ChatOpenAI
+
 llm = ChatOpenAI(model="gpt-4", callbacks=[handler])
 ```
 

docs/custom_adapters_tutorial.md

@@ -13,6 +13,7 @@ def custom_execution_wrapper(func):
         result = func(*args, **kwargs)
         # 3. Publish completion event
         return result
+
     return wrapper
 ```
 

docs/getting_started_extended.md

@@ -13,10 +13,12 @@ Verify setup by wrapping a dummy model agent:
 ```python
 from agentwatch import watch
 
+
 class TestAgent:
     def run(self, query: str) -> str:
         return f"Response to: {query}"
 
+
 agent = watch(TestAgent())
 agent.run("Verify connection parameters.")
 ```

tests/test_audit_persistence.py

@@ -23,9 +23,7 @@ async def test_append_persists_and_chains():
     store = InMemoryAuditStore()
     log = PersistentAuditLog(store)
 
-    first = await log.append(
-        "role.change", "u1", actor="admin", details={"to": "admin"}
-    )
+    first = await log.append("role.change", "u1", actor="admin", details={"to": "admin"})
     second = await log.append("policy.set", "team-1")
 
     assert first.prev_hash == GENESIS_HASH