feat(automation): add autonomy guardrails and audit trail

Add runtime autonomy constraints (max steps, tool budget, required
confirmations) and a structured audit trail for plan execution.

New domain models:
- AutonomyGuardrails: enforces step limits, tool budgets, and
  confirmation gates with validators and check methods
- GuardrailAuditEntry: records each enforcement event with timestamp,
  event type, guard name, result, reason, and context
- GuardrailAuditTrail: ordered collection of audit entries persisted
  to plan metadata

New service:
- AutonomyGuardrailService: high-level service for configuring
  guardrails per plan, checking constraints, recording audit entries,
  and serializing/restoring state via plan metadata

Tests:
- Behave: 69 scenarios covering model validation, step/budget/
  confirmation checks, audit trail recording, and service operations
- Robot: 8 test cases for autonomy guardrail CLI flag smoke testing
- ASV: 6 benchmark suites measuring enforcement overhead

Documentation:
- Updated docs/reference/automation_profiles.md with guardrail fields,
  enforcement behavior, audit trail schema, and event type reference

ISSUES CLOSED: #204

benchmarks/autonomy_guardrails_bench.py

@@ -0,0 +1,285 @@
+"""ASV benchmarks for autonomy guardrails enforcement overhead.
+
+Measures the performance of:
+- AutonomyGuardrails model construction
+- Step limit checks
+- Tool budget checks
+- Confirmation checks
+- Wall-clock time limit checks
+- Actor tool-call limit checks
+- Audit trail entry creation, appending, and eviction
+- AutonomyGuardrailService operations
+- Metadata serialization and deserialization
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+try:
+    from cleveragents.application.services.autonomy_guardrail_service import (
+        AutonomyGuardrailService,
+    )
+    from cleveragents.domain.models.core.autonomy_guardrails import (
+        ActorLimits,
+        AutonomyGuardrails,
+        GuardrailAuditEntry,
+        GuardrailAuditTrail,
+        GuardrailEventType,
+        GuardrailResult,
+    )
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    from cleveragents.application.services.autonomy_guardrail_service import (
+        AutonomyGuardrailService,
+    )
+    from cleveragents.domain.models.core.autonomy_guardrails import (
+        ActorLimits,
+        AutonomyGuardrails,
+        GuardrailAuditEntry,
+        GuardrailAuditTrail,
+        GuardrailEventType,
+        GuardrailResult,
+    )
+
+
+class GuardrailsConstructionSuite:
+    """Benchmark AutonomyGuardrails model construction."""
+
+    def time_guardrails_construction_full(self) -> None:
+        """Benchmark fully-populated guardrails creation."""
+        AutonomyGuardrails(
+            max_steps=100,
+            tool_budget=500.0,
+            required_confirmations=["deploy", "delete", "write"],
+        )
+
+    def time_guardrails_construction_minimal(self) -> None:
+        """Benchmark minimal guardrails creation."""
+        AutonomyGuardrails()
+
+    def time_guardrails_construction_steps_only(self) -> None:
+        """Benchmark guardrails with only step limit."""
+        AutonomyGuardrails(max_steps=50)
+
+    def time_guardrails_construction_budget_only(self) -> None:
+        """Benchmark guardrails with only budget."""
+        AutonomyGuardrails(tool_budget=100.0)
+
+
+class StepLimitCheckSuite:
+    """Benchmark step limit enforcement checks."""
+
+    def setup(self) -> None:
+        """Create guardrails for benchmarking."""
+        self.limited = AutonomyGuardrails(max_steps=100)
+        self.unlimited = AutonomyGuardrails()
+
+    def time_step_limit_allowed(self) -> None:
+        """Benchmark step check when allowed."""
+        self.limited.step_count = 50
+        self.limited.check_step_limit()
+
+    def time_step_limit_denied(self) -> None:
+        """Benchmark step check when denied."""
+        self.limited.step_count = 100
+        self.limited.check_step_limit()
+
+    def time_step_limit_unlimited(self) -> None:
+        """Benchmark step check with no limit."""
+        self.unlimited.step_count = 10000
+        self.unlimited.check_step_limit()
+
+
+class BudgetCheckSuite:
+    """Benchmark tool budget enforcement checks."""
+
+    def setup(self) -> None:
+        """Create guardrails for benchmarking."""
+        self.budgeted = AutonomyGuardrails(tool_budget=100.0)
+        self.unlimited = AutonomyGuardrails()
+
+    def time_budget_allowed(self) -> None:
+        """Benchmark budget check when allowed."""
+        self.budgeted.budget_spent = 50.0
+        self.budgeted.check_tool_budget(10.0)
+
+    def time_budget_denied(self) -> None:
+        """Benchmark budget check when denied."""
+        self.budgeted.budget_spent = 95.0
+        self.budgeted.check_tool_budget(10.0)
+
+    def time_budget_unlimited(self) -> None:
+        """Benchmark budget check with no limit."""
+        self.unlimited.check_tool_budget(999.0)
+
+
+class ConfirmationCheckSuite:
+    """Benchmark confirmation requirement checks."""
+
+    def setup(self) -> None:
+        """Create guardrails for benchmarking."""
+        self.with_confirmations = AutonomyGuardrails(
+            required_confirmations=[
+                "deploy",
+                "delete",
+                "write",
+                "apply",
+                "revert",
+            ],
+        )
+        self.no_confirmations = AutonomyGuardrails()
+
+    def time_confirmation_required(self) -> None:
+        """Benchmark check when confirmation is required."""
+        self.with_confirmations.check_confirmation_required("deploy")
+
+    def time_confirmation_not_required(self) -> None:
+        """Benchmark check when confirmation is not required."""
+        self.with_confirmations.check_confirmation_required("read")
+
+    def time_confirmation_empty_list(self) -> None:
+        """Benchmark check with empty confirmation list."""
+        self.no_confirmations.check_confirmation_required("deploy")
+
+
+class AuditTrailSuite:
+    """Benchmark audit trail operations."""
+
+    def setup(self) -> None:
+        """Create audit trail for benchmarking."""
+        self.trail = GuardrailAuditTrail()
+        self.entry = GuardrailAuditEntry(
+            event_type=GuardrailEventType.STEP_ALLOWED,
+            guard_name="step_limit",
+            result=GuardrailResult.ALLOWED,
+        )
+
+    def time_audit_entry_creation(self) -> None:
+        """Benchmark audit entry creation."""
+        GuardrailAuditEntry(
+            event_type=GuardrailEventType.STEP_ALLOWED,
+            guard_name="step_limit",
+            result=GuardrailResult.ALLOWED,
+        )
+
+    def time_audit_trail_append(self) -> None:
+        """Benchmark appending to audit trail."""
+        self.trail.add_entry(self.entry)
+
+    def time_audit_trail_counts(self) -> None:
+        """Benchmark counting allowed/denied entries."""
+        _ = self.trail.allowed_count
+        _ = self.trail.denied_count
+
+
+class ServiceSuite:
+    """Benchmark AutonomyGuardrailService operations."""
+
+    def setup(self) -> None:
+        """Create service for benchmarking."""
+        self.service = AutonomyGuardrailService()
+        self.guardrails = AutonomyGuardrails(
+            max_steps=100,
+            tool_budget=500.0,
+            required_confirmations=["deploy"],
+        )
+        self.service.configure_guardrails("bench-plan", self.guardrails)
+
+    def time_service_check_step(self) -> None:
+        """Benchmark service step limit check."""
+        self.service.check_step_limit("bench-plan", 50)
+
+    def time_service_check_budget(self) -> None:
+        """Benchmark service budget check."""
+        self.service.check_tool_budget("bench-plan", 1.0)
+
+    def time_service_check_confirmation(self) -> None:
+        """Benchmark service confirmation check."""
+        self.service.check_confirmation_required("bench-plan", "deploy")
+
+    def time_service_get_metadata(self) -> None:
+        """Benchmark service metadata serialization."""
+        self.service.get_audit_trail_as_metadata("bench-plan")
+
+    def time_service_configure(self) -> None:
+        """Benchmark service guardrail configuration."""
+        self.service.configure_guardrails(
+            "new-plan",
+            AutonomyGuardrails(max_steps=50),
+        )
+
+    def time_service_check_wall_clock(self) -> None:
+        """Benchmark service wall-clock check."""
+        self.service.check_wall_clock("bench-plan")
+
+    def time_service_check_actor_calls(self) -> None:
+        """Benchmark service actor tool-call check."""
+        self.service.check_actor_tool_calls("bench-plan", 5)
+
+
+class WallClockCheckSuite:
+    """Benchmark wall-clock time limit checks."""
+
+    def setup(self) -> None:
+        """Create guardrails for benchmarking."""
+        self.with_limit = AutonomyGuardrails(max_wall_clock_seconds=3600.0)
+        self.with_limit.mark_started()
+        self.no_limit = AutonomyGuardrails()
+
+    def time_wall_clock_allowed(self) -> None:
+        """Benchmark wall-clock check when allowed."""
+        self.with_limit.check_wall_clock()
+
+    def time_wall_clock_no_limit(self) -> None:
+        """Benchmark wall-clock check with no limit."""
+        self.no_limit.check_wall_clock()
+
+
+class ActorLimitCheckSuite:
+    """Benchmark per-actor tool-call limit checks."""
+
+    def setup(self) -> None:
+        """Create guardrails for benchmarking."""
+        self.with_limit = AutonomyGuardrails(
+            actor_limits=ActorLimits(max_tool_calls_per_invocation=100),
+        )
+        self.no_limit = AutonomyGuardrails()
+
+    def time_actor_limit_allowed(self) -> None:
+        """Benchmark actor limit check when allowed."""
+        self.with_limit.check_actor_tool_calls(50)
+
+    def time_actor_limit_denied(self) -> None:
+        """Benchmark actor limit check when denied."""
+        self.with_limit.check_actor_tool_calls(100)
+
+    def time_actor_limit_unlimited(self) -> None:
+        """Benchmark actor limit check with no limit."""
+        self.no_limit.check_actor_tool_calls(10000)
+
+
+class AuditTrailEvictionSuite:
+    """Benchmark audit trail eviction under load."""
+
+    def setup(self) -> None:
+        """Create a full audit trail for benchmarking eviction."""
+        self.trail = GuardrailAuditTrail(max_entries=100)
+        for _ in range(100):
+            self.trail.add_entry(
+                GuardrailAuditEntry(
+                    event_type=GuardrailEventType.STEP_ALLOWED,
+                    guard_name="step_limit",
+                    result=GuardrailResult.ALLOWED,
+                )
+            )
+        self.entry = GuardrailAuditEntry(
+            event_type=GuardrailEventType.STEP_ALLOWED,
+            guard_name="step_limit",
+            result=GuardrailResult.ALLOWED,
+        )
+
+    def time_eviction_append(self) -> None:
+        """Benchmark appending to a full trail (triggers eviction)."""
+        self.trail.add_entry(self.entry)

docs/reference/automation_profiles.md

@@ -271,3 +271,123 @@ Legacy automation levels are mapped to built-in profiles:
 | `full_auto` | `full-auto` |
 
 Use `--automation-profile <name>` instead of `--automation-level`.
+
+## Autonomy Guardrails
+
+Autonomy guardrails extend the automation guard framework with runtime
+constraints that enforce step limits, tool budgets, and required human
+confirmations during plan execution. Unlike the phase-transition thresholds
+which gate entire phases, autonomy guardrails operate at the individual
+step and tool invocation level.
+
+### Guardrail Fields
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `max_steps` | `int \| None` | `None` | Maximum execution steps allowed. `None` means unlimited. |
+| `tool_budget` | `float \| None` | `None` | Maximum cumulative cost for tool invocations. `None` means unlimited. A value of `0.0` blocks any positive cost. |
+| `max_wall_clock_seconds` | `float \| None` | `None` | Maximum wall-clock time in seconds for plan execution. `None` means unlimited. |
+| `actor_limits` | `ActorLimits` | `ActorLimits()` | Per-actor runtime limits (see below). |
+| `required_confirmations` | `list[str]` | `[]` | Operation names requiring human confirmation before execution (case-insensitive). |
+| `step_count` | `int` | `0` | Current execution step counter (managed by the service). |
+| `budget_spent` | `float` | `0.0` | Current cumulative budget spent (managed by the service). |
+| `start_time` | `str \| None` | `None` | ISO-8601 timestamp of when plan execution started (for wall-clock tracking). |
+
+### Actor Limits
+
+Per-actor constraints that limit tool calls within a single actor invocation:
+
+| Field | Type | Default | Description |
+|-------|------|---------|-------------|
+| `max_tool_calls_per_invocation` | `int \| None` | `None` | Maximum tool calls allowed per actor invocation. `None` means unlimited. |
+| `max_retries_per_failure` | `int \| None` | `None` | Maximum retries per tool failure within an actor step. `None` means unlimited. |
+
+### Enforcement Behavior
+
+Guardrails are enforced at multiple points during plan execution:
+
+- **Pre-flight** — Before execution begins, `PlanExecutor._guard_execute()` calls `_enforce_guardrails()` which checks wall-clock time and the current step limit. If either check fails, execution is blocked with a `PlanError` before any decisions are processed.
+- **Per-decision** — Before each decision in the execution loop, `_enforce_guardrails_per_step()` increments the step counter, checks the step limit, and rechecks wall-clock time. If any check fails, execution halts with a `PlanError`.
+- **Start-time tracking** — On the first execution, `_enforce_guardrails()` automatically calls `mark_started()` on the guardrails model to record the wall-clock start time (if not already set).
+
+The individual guardrail checks are:
+
+1. **Step limit** — The `step_count` is compared against `max_steps`. If equal or exceeded, the step is **blocked** and an audit entry is recorded.
+2. **Tool budget** — The projected cost (`budget_spent + tool_cost`) is compared against `tool_budget`. If exceeded, the invocation is **blocked** and an audit entry is recorded. When allowed, the cost is recorded **before** the audit entry so the trail reflects the post-deduction budget state.
+3. **Wall-clock time** — If `max_wall_clock_seconds` is set and `start_time` is recorded, the elapsed wall-clock time is checked. If the limit is exceeded, the step is **blocked**. Malformed `start_time` values are treated as a block (defense-in-depth) rather than raising an unhandled exception.
+4. **Actor tool-call limit** — The actor's current tool call count is compared against `actor_limits.max_tool_calls_per_invocation`. If exceeded, the call is **blocked**.
+5. **Retry-per-failure limit** — The current retry count is compared against `actor_limits.max_retries_per_failure`. If equal or exceeded, the retry is **blocked** and an audit entry is recorded. When `max_retries_per_failure` is `None`, retries are unlimited.
+6. **Required confirmations** — The operation name is checked (case-insensitively) against `required_confirmations`. If matched, the operation is **blocked** pending human confirmation.
+
+If no guardrail blocks the invocation, it proceeds and the step counter and budget are updated.
+
+### Input Validation
+
+- `start_time` is validated on assignment to ensure it is a parseable ISO-8601 timestamp. Malformed strings are rejected with a `ValidationError`.
+- `max_retries_per_failure` must be non-negative (`>= 0`) if set; `None` means unlimited.
+- `max_steps` must be positive (`>= 1`) if set.
+- `tool_budget` must be non-negative (`>= 0.0`) if set.
+
+### Audit Trail
+
+Every guardrail enforcement event is recorded in a `GuardrailAuditTrail` which is persisted to plan metadata. The audit trail provides a bounded, ordered record of autonomy-related decisions. When the number of entries exceeds `max_entries` (default: 10,000), the oldest entries are evicted to prevent unbounded memory growth.
+
+#### Audit Entry Fields
+
+| Field | Type | Description |
+|-------|------|-------------|
+| `timestamp` | `str` | ISO-8601 timestamp of the enforcement event. |
+| `event_type` | `GuardrailEventType` | Type of guardrail event (see below). |
+| `guard_name` | `str` | Name of the guard that was evaluated. |
+| `result` | `GuardrailResult` | Outcome: `allowed` or `denied`. |
+| `reason` | `str \| None` | Human-readable reason for the decision. |
+| `context` | `dict` | Additional metadata about the enforcement event. |
+
+#### Event Types
+
+| Event Type | Description |
+|------------|-------------|
+| `step_blocked` | A step was blocked due to the step limit. |
+| `budget_blocked` | A tool invocation was blocked due to budget. |
+| `confirmation_required` | An operation was blocked pending confirmation. |
+| `step_allowed` | A step was allowed within the limit. |
+| `budget_allowed` | A tool invocation was allowed within budget. |
+| `confirmation_granted` | An operation was allowed (no confirmation needed). |
+| `time_blocked` | A step was blocked due to the wall-clock time limit. |
+| `time_allowed` | A step was allowed within the wall-clock time limit. |
+| `actor_limit_blocked` | An actor tool call was blocked due to per-invocation limits. |
+| `actor_limit_allowed` | An actor tool call was allowed within per-invocation limits. |
+| `retry_blocked` | A retry was blocked due to the per-failure retry limit. |
+| `retry_allowed` | A retry was allowed within the per-failure retry limit. |
+
+### Plan Metadata Persistence
+
+The guardrail state and audit trail are serialized to plan metadata using two keys:
+
+- `autonomy_guardrails` — The current guardrail configuration and counters.
+- `guardrail_audit_trail` — The ordered list of audit entries.
+
+This allows guardrail state to survive across plan restarts and provides
+a complete audit record for post-mortem analysis.
+
+### Example Configuration
+
+```yaml
+name: acme/guarded-auto
+description: Auto profile with autonomy guardrails
+schema_version: "1.0"
+auto_strategize: 0.0
+auto_execute: 0.0
+auto_apply: 1.0
+
+guards:
+  max_tool_calls_per_step: 10
+  max_total_cost: 100.0
+
+# Autonomy guardrails (configured via service or CLI)
+# max_steps: 50
+# tool_budget: 200.0
+# required_confirmations:
+#   - deploy
+#   - delete
+```

features/autonomy_guardrails.feature

@@ -0,0 +1,401 @@
+Feature: Autonomy Guardrails and Audit Trail
+  As a plan executor
+  I want autonomy guardrails to enforce step limits, tool budgets, and confirmations
+  So that plan execution stays within defined safety boundaries with a full audit trail
+
+  # ---- AutonomyGuardrails model validation ----
+
+  Scenario: Create guardrails with all fields
+    When I create autonomy guardrails with max_steps 10 and tool_budget 50.0
+    Then the guardrails should have max_steps 10
+    And the guardrails should have tool_budget 50.0
+    And the guardrails step_count should be 0
+    And the guardrails budget_spent should be 0.0
+
+  Scenario: Create guardrails with defaults
+    When I create autonomy guardrails with defaults
+    Then the guardrails should have max_steps None
+    And the guardrails should have tool_budget None
+    And the guardrails should have empty required_confirmations
+
+  Scenario: Guardrails reject negative max_steps
+    When I try to create guardrails with max_steps 0
+    Then a guardrails validation error should be raised
+    And the guardrails error should mention "max_steps"
+
+  Scenario: Guardrails reject negative tool_budget
+    When I try to create guardrails with tool_budget -1.0
+    Then a guardrails validation error should be raised
+    And the guardrails error should mention "tool_budget"
+
+  # ---- Step limit checks ----
+
+  Scenario: Step limit allows within bounds
+    Given guardrails with max_steps 5
+    When I check step limit at step 3
+    Then the step check should be allowed
+
+  Scenario: Step limit blocks at max
+    Given guardrails with max_steps 5
+    When I check step limit at step 5
+    Then the step check should be denied
+
+  Scenario: Step limit blocks above max
+    Given guardrails with max_steps 3
+    When I check step limit at step 10
+    Then the step check should be denied
+
+  Scenario: Unlimited steps always allowed
+    Given guardrails with no step limit
+    When I check step limit at step 1000
+    Then the step check should be allowed
+
+  # ---- Tool budget checks ----
+
+  Scenario: Budget allows within limit
+    Given guardrails with tool_budget 100.0
+    When I check tool budget for cost 50.0
+    Then the budget check should be allowed
+
+  Scenario: Budget blocks when exceeded
+    Given guardrails with tool_budget 10.0 and budget_spent 8.0
+    When I check tool budget for cost 5.0
+    Then the budget check should be denied
+
+  Scenario: Budget allows exact remaining
+    Given guardrails with tool_budget 10.0 and budget_spent 5.0
+    When I check tool budget for cost 5.0
+    Then the budget check should be allowed
+
+  Scenario: Unlimited budget always allowed
+    Given guardrails with no budget limit
+    When I check tool budget for cost 999.0
+    Then the budget check should be allowed
+
+  Scenario: Budget rejects negative tool cost
+    Given guardrails with tool_budget 100.0
+    When I try to check tool budget for cost -1.0
+    Then a guardrails validation error should be raised
+
+  # ---- Confirmation checks ----
+
+  Scenario: Confirmation required for listed operation
+    Given guardrails with required confirmations "deploy,delete"
+    When I check confirmation for operation "deploy"
+    Then confirmation should be required
+
+  Scenario: Confirmation not required for unlisted operation
+    Given guardrails with required confirmations "deploy,delete"
+    When I check confirmation for operation "read"
+    Then confirmation should not be required
+
+  Scenario: No confirmations required when list empty
+    Given guardrails with no required confirmations
+    When I check confirmation for operation "deploy"
+    Then confirmation should not be required
+
+  # ---- Step counter and budget tracking ----
+
+  Scenario: Increment step counter
+    Given guardrails with max_steps 10
+    When I increment the step counter 3 times
+    Then the step_count should be 3
+
+  Scenario: Record tool cost
+    Given guardrails with tool_budget 100.0
+    When I record a tool cost of 25.0
+    And I record a tool cost of 15.0
+    Then the budget_spent should be 40.0
+
+  Scenario: Record cost rejects negative cost
+    Given guardrails with tool_budget 100.0
+    When I try to record a negative cost
+    Then a guardrails validation error should be raised
+
+  # ---- Audit trail ----
+
+  Scenario: Audit trail records entries
+    Given an empty audit trail
+    When I add a step_allowed entry to the audit trail
+    And I add a budget_blocked entry to the audit trail
+    Then the audit trail should have 2 entries
+    And the audit trail allowed_count should be 1
+    And the audit trail denied_count should be 1
+
+  Scenario: Audit entry has correct fields
+    When I create an audit entry with event_type step_blocked and guard_name step_limit
+    Then the audit entry event_type should be "step_blocked"
+    And the audit entry guard_name should be "step_limit"
+    And the audit entry result should be "denied"
+    And the audit entry should have a timestamp
+
+  # ---- AutonomyGuardrailService ----
+
+  Scenario: Service configures and retrieves guardrails
+    Given an autonomy guardrail service
+    When I configure guardrails for plan "plan-001" with max_steps 5
+    Then the service should return guardrails for plan "plan-001"
+    And the service guardrails should have max_steps 5
+
+  Scenario: Service returns None for unconfigured plan
+    Given an autonomy guardrail service
+    Then the service should return None for plan "nonexistent"
+
+  Scenario: Service step limit check records audit trail
+    Given an autonomy guardrail service with guardrails for plan "plan-002"
+    When I check step limit via service for plan "plan-002" at step 3
+    Then the audit trail for plan "plan-002" should have 1 entry
+
+  Scenario: Service budget check records cost and audit trail
+    Given an autonomy guardrail service with budget guardrails for plan "plan-003"
+    When I check tool budget via service for plan "plan-003" with cost 5.0
+    Then the audit trail for plan "plan-003" should have 1 entry
+    And the service guardrails budget_spent should be 5.0
+
+  Scenario: Service confirmation check records audit trail
+    Given an autonomy guardrail service with confirmation guardrails for plan "plan-004"
+    When I check confirmation via service for plan "plan-004" operation "deploy"
+    Then the audit trail for plan "plan-004" should have 1 entry
+
+  Scenario: Service serializes audit trail to metadata
+    Given an autonomy guardrail service with guardrails for plan "plan-005"
+    When I check step limit via service for plan "plan-005" at step 1
+    Then the metadata for plan "plan-005" should contain the audit trail
+
+  Scenario: Service loads guardrails from metadata
+    Given an autonomy guardrail service
+    When I load guardrails from metadata for plan "plan-006"
+    Then the service should return guardrails for plan "plan-006"
+
+  Scenario: Service removes plan guardrails
+    Given an autonomy guardrail service with guardrails for plan "plan-007"
+    When I remove plan "plan-007" from the service
+    Then the service should return None for plan "plan-007"
+
+  Scenario: Service check_step_limit returns True for unconfigured plan
+    Given an autonomy guardrail service
+    When I check step limit via service for plan "unknown" at step 100
+    Then the step limit check should return True
+
+  Scenario: Service check_tool_budget returns True for unconfigured plan
+    Given an autonomy guardrail service
+    When I check tool budget via service for plan "unknown" with cost 999.0
+    Then the tool budget check should return True
+
+  Scenario: Service check_confirmation returns False for unconfigured plan
+    Given an autonomy guardrail service
+    When I check confirmation via service for plan "unknown" operation "deploy"
+    Then the confirmation check should return False
+
+  Scenario: Service configure_guardrails rejects empty plan_id
+    Given an autonomy guardrail service
+    When I try to configure guardrails with empty plan_id
+    Then a guardrails validation error should be raised
+
+  Scenario: GuardrailEventType enumerates all event types
+    Then GuardrailEventType should have at least 10 members
+
+  Scenario: GuardrailResult enumerates allowed and denied
+    Then GuardrailResult should have exactly 2 members
+
+  # ---- Wall-clock time checks ----
+
+  Scenario: Wall-clock allows when no limit set
+    Given guardrails with no wall-clock limit
+    When I check wall-clock time
+    Then the wall-clock check should be allowed
+
+  Scenario: Wall-clock allows when within limit
+    Given guardrails with wall-clock limit 60.0 seconds started recently
+    When I check wall-clock time
+    Then the wall-clock check should be allowed
+
+  Scenario: Wall-clock blocks when expired
+    Given guardrails with wall-clock limit that has already expired
+    When I check wall-clock time
+    Then the wall-clock check should be denied
+
+  Scenario: Service wall-clock check records audit trail
+    Given an autonomy guardrail service with wall-clock guardrails for plan "plan-wc"
+    When I check wall-clock via service for plan "plan-wc"
+    Then the audit trail for plan "plan-wc" should have 1 entry
+
+  # ---- Actor tool-call limits ----
+
+  Scenario: Actor tool-call limit allows within bounds
+    Given guardrails with actor tool-call limit 5
+    When I check actor tool calls with 3 current calls
+    Then the actor tool-call check should be allowed
+
+  Scenario: Actor tool-call limit blocks at max
+    Given guardrails with actor tool-call limit 5
+    When I check actor tool calls with 5 current calls
+    Then the actor tool-call check should be denied
+
+  Scenario: Actor tool-call limit unlimited
+    Given guardrails with no actor tool-call limit
+    When I check actor tool calls with 1000 current calls
+    Then the actor tool-call check should be allowed
+
+  Scenario: Service actor tool-call check records audit trail
+    Given an autonomy guardrail service with actor-limit guardrails for plan "plan-al"
+    When I check actor tool calls via service for plan "plan-al" with 3 calls
+    Then the audit trail for plan "plan-al" should have 1 entry
+
+  # ---- Bounded audit trail eviction ----
+
+  Scenario: Audit trail evicts oldest entry when full
+    Given an audit trail with max_entries 3
+    When I add 4 step_allowed entries to the audit trail
+    Then the audit trail should have 3 entries
+    And the audit trail allowed_count should be 3
+
+  # ---- Edge cases ----
+
+  Scenario: Budget blocks when tool_budget is zero
+    Given guardrails with tool_budget 0.0
+    When I check tool budget for cost 0.01
+    Then the budget check should be denied
+
+  Scenario: Budget allows zero cost when tool_budget is zero
+    Given guardrails with tool_budget 0.0
+    When I check tool budget for cost 0
+    Then the budget check should be allowed
+
+  Scenario: Case-insensitive confirmation matching
+    Given guardrails with required confirmations "Deploy,DELETE"
+    When I check confirmation for operation "deploy"
+    Then confirmation should be required
+
+  Scenario: Case-insensitive confirmation matching uppercase query
+    Given guardrails with required confirmations "deploy"
+    When I check confirmation for operation "DEPLOY"
+    Then confirmation should be required
+
+  Scenario: Service rejects oversized metadata entries
+    Given an autonomy guardrail service
+    When I try to load metadata with oversized audit trail for plan "plan-big"
+    Then a guardrails validation error should be raised
+
+  Scenario: Service rejects oversized confirmations in metadata
+    Given an autonomy guardrail service
+    When I try to load metadata with oversized confirmations for plan "plan-big2"
+    Then a guardrails validation error should be raised
+
+  Scenario: Metadata round-trip preserves guardrail state
+    Given an autonomy guardrail service with guardrails for plan "plan-rt"
+    When I check step limit via service for plan "plan-rt" at step 3
+    And I serialize and restore metadata for plan "plan-rt"
+    Then the restored guardrails for plan "plan-rt" should match the original
+
+  # ---- start_time validation ----
+
+  Scenario: start_time accepts valid ISO-8601 timestamp
+    When I create guardrails with a valid start_time
+    Then the guardrails start_time should be set
+
+  Scenario: start_time rejects malformed string
+    When I try to create guardrails with start_time "not-a-date"
+    Then a guardrails validation error should be raised
+    And the guardrails error should mention "start_time"
+
+  Scenario: start_time accepts None
+    When I create autonomy guardrails with defaults
+    Then the guardrails start_time should be None
+
+  Scenario: check_wall_clock handles malformed start_time gracefully
+    Given guardrails with a malformed start_time bypassing validation
+    When I check wall-clock time
+    Then the wall-clock check should be denied
+
+  # ---- Retry-per-failure checks ----
+
+  Scenario: Retry limit allows within bounds
+    Given guardrails with retry limit 3
+    When I check retries per failure with 2 current retries
+    Then the retry check should be allowed
+
+  Scenario: Retry limit blocks at max
+    Given guardrails with retry limit 3
+    When I check retries per failure with 3 current retries
+    Then the retry check should be denied
+
+  Scenario: Retry limit blocks above max
+    Given guardrails with retry limit 2
+    When I check retries per failure with 5 current retries
+    Then the retry check should be denied
+
+  Scenario: Unlimited retries always allowed
+    Given guardrails with no retry limit
+    When I check retries per failure with 1000 current retries
+    Then the retry check should be allowed
+
+  Scenario: Service retry check records audit trail
+    Given an autonomy guardrail service with retry-limit guardrails for plan "plan-rl"
+    When I check retries via service for plan "plan-rl" with 1 retries
+    Then the audit trail for plan "plan-rl" should have 1 entry
+
+  Scenario: Service retry check blocks and records audit trail
+    Given an autonomy guardrail service with retry-limit guardrails for plan "plan-rl2"
+    When I check retries via service for plan "plan-rl2" with 5 retries
+    Then the retry service check should return False
+    And the audit trail for plan "plan-rl2" should have 1 entry
+
+  Scenario: Successive retries stop after reaching max_retries_per_failure
+    Given a guardrail service with max_retries_per_failure 3 for plan "retry-stop"
+    When I check retries for plan "retry-stop" incrementing from 0 to 4
+    Then retry checks 0, 1, 2 should be allowed
+    And retry checks 3, 4 should be denied
+    And the audit trail for plan "retry-stop" should have 5 entry
+
+  # ---- PlanExecutor guardrail integration ----
+
+  Scenario: PlanExecutor enforces step limit during stub execution
+    Given a plan executor with guardrail service and step limit 2
+    And a plan in execute phase with 3 decisions
+    When I run execute on the plan
+    Then execution should fail with a guardrail step limit error
+
+  Scenario: PlanExecutor enforces wall-clock limit during execution
+    Given a plan executor with guardrail service and expired wall-clock
+    And a plan in execute phase with 1 decisions
+    When I run execute on the plan
+    Then execution should fail with a guardrail wall-clock error
+
+  Scenario: PlanExecutor runs without guardrails when service is None
+    Given a plan executor without guardrail service
+    And a plan in execute phase with 2 decisions
+    When I run execute on the plan
+    Then execution should succeed
+
+  Scenario: PlanExecutor marks start_time on first execution
+    Given a plan executor with guardrail service and no wall-clock limit
+    And a plan in execute phase with 1 decisions
+    When I run execute on the plan
+    Then the guardrails start_time should have been set
+
+  # ---- Additional validation edge cases for coverage ----
+
+  Scenario: Guardrails reject invalid max_tool_calls_per_invocation
+    When I try to create actor limits with max_tool_calls_per_invocation 0
+    Then a guardrails validation error should be raised
+    And the guardrails error should mention "max_tool_calls_per_invocation"
+
+  Scenario: Guardrails reject negative max_retries_per_failure
+    When I try to create actor limits with max_retries_per_failure -1
+    Then a guardrails validation error should be raised
+    And the guardrails error should mention "max_retries_per_failure"
+
+  Scenario: Guardrails reject non-positive max_wall_clock_seconds
+    When I try to create guardrails with max_wall_clock_seconds 0
+    Then a guardrails validation error should be raised
+    And the guardrails error should mention "max_wall_clock_seconds"
+
+  Scenario: Audit trail evicts denied entry when full
+    Given an audit trail with max_entries 2
+    When I add a budget_blocked entry to the audit trail
+    And I add a step_allowed entry to the audit trail
+    And I add a step_allowed entry to the audit trail
+    Then the audit trail should have 2 entries
+    And the audit trail denied_count should be 0
+    And the audit trail allowed_count should be 2

robot/autonomy_guardrails.robot

@@ -0,0 +1,87 @@
+*** Settings ***
+Documentation    Smoke tests for autonomy guardrails and audit trail
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_autonomy_guardrails.py
+
+*** Test Cases ***
+Guardrail Step Limit Allows Within Bounds
+    [Documentation]    Step limit allows when under limit
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    step-allow    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    step-allow-ok
+
+Guardrail Step Limit Blocks At Max
+    [Documentation]    Step limit blocks when at max steps
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    step-block    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    step-block-ok
+
+Guardrail Budget Allows Within Limit
+    [Documentation]    Budget allows when within limit
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    budget-allow    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    budget-allow-ok
+
+Guardrail Budget Blocks Exceeded
+    [Documentation]    Budget blocks when exceeded
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    budget-block    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    budget-block-ok
+
+Guardrail Confirmation Required
+    [Documentation]    Confirmation required for listed operation
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    confirm-required    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    confirm-required-ok
+
+Guardrail Audit Trail Records Events
+    [Documentation]    Audit trail records enforcement events
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    audit-trail    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    audit-trail-ok
+
+Guardrail Service Metadata Persistence
+    [Documentation]    Service can serialize and restore from metadata
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    metadata    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    metadata-ok
+
+Guardrail Wall Clock Allows Within Limit
+    [Documentation]    Wall-clock allows when within time limit
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    wall-clock-allow    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wall-clock-allow-ok
+
+Guardrail Wall Clock Blocks When Expired
+    [Documentation]    Wall-clock blocks when time limit exceeded
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    wall-clock-block    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wall-clock-block-ok
+
+Guardrail Actor Tool Call Limit
+    [Documentation]    Actor tool-call limit enforcement
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    actor-limit    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    actor-limit-ok
+
+Guardrail Audit Trail Eviction
+    [Documentation]    Audit trail evicts oldest entries when full
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    eviction    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    eviction-ok
+
+Guardrail Zero Budget Edge Case
+    [Documentation]    Zero budget blocks any positive cost
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    zero-budget    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    zero-budget-ok
+
+Guardrail CLI Flags Summary
+    [Documentation]    Summary of autonomy guardrail features
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    summary    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    summary-ok

robot/helper_autonomy_guardrails.py

@@ -0,0 +1,220 @@
+"""Helper script for Robot Framework autonomy guardrails tests."""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from cleveragents.application.services.autonomy_guardrail_service import (
+    AutonomyGuardrailService,
+)
+from cleveragents.domain.models.core.autonomy_guardrails import (
+    ActorLimits,
+    AutonomyGuardrails,
+    GuardrailAuditEntry,
+    GuardrailAuditTrail,
+    GuardrailEventType,
+    GuardrailResult,
+)
+
+
+def _test_step_allow() -> None:
+    """Step limit allows within bounds."""
+    g = AutonomyGuardrails(max_steps=10)
+    g.step_count = 5
+    allowed, _ = g.check_step_limit()
+    assert allowed is True
+    print("step-allow-ok")
+
+
+def _test_step_block() -> None:
+    """Step limit blocks at max."""
+    g = AutonomyGuardrails(max_steps=5)
+    g.step_count = 5
+    allowed, reason = g.check_step_limit()
+    assert allowed is False
+    assert reason is not None
+    assert "Step limit" in reason
+    print("step-block-ok")
+
+
+def _test_budget_allow() -> None:
+    """Budget allows within limit."""
+    g = AutonomyGuardrails(tool_budget=100.0)
+    allowed, _ = g.check_tool_budget(50.0)
+    assert allowed is True
+    print("budget-allow-ok")
+
+
+def _test_budget_block() -> None:
+    """Budget blocks when exceeded."""
+    g = AutonomyGuardrails(tool_budget=10.0, budget_spent=8.0)
+    allowed, reason = g.check_tool_budget(5.0)
+    assert allowed is False
+    assert reason is not None
+    assert "Budget exceeded" in reason
+    print("budget-block-ok")
+
+
+def _test_confirm_required() -> None:
+    """Confirmation required for listed operation."""
+    g = AutonomyGuardrails(required_confirmations=["deploy", "delete"])
+    required, reason = g.check_confirmation_required("deploy")
+    assert required is True
+    assert reason is not None
+    not_required, _ = g.check_confirmation_required("read")
+    assert not_required is False
+    print("confirm-required-ok")
+
+
+def _test_audit_trail() -> None:
+    """Audit trail records enforcement events."""
+    svc = AutonomyGuardrailService()
+    svc.configure_guardrails(
+        "test-plan",
+        AutonomyGuardrails(max_steps=5, tool_budget=50.0),
+    )
+    svc.check_step_limit("test-plan", 3)
+    svc.check_tool_budget("test-plan", 10.0)
+    trail = svc.get_audit_trail("test-plan")
+    assert len(trail.entries) == 2
+    assert trail.entries[0].guard_name == "step_limit"
+    assert trail.entries[1].guard_name == "tool_budget"
+    print("audit-trail-ok")
+
+
+def _test_metadata() -> None:
+    """Service can serialize and restore from metadata."""
+    svc = AutonomyGuardrailService()
+    svc.configure_guardrails(
+        "meta-plan",
+        AutonomyGuardrails(max_steps=10, tool_budget=25.0),
+    )
+    svc.check_step_limit("meta-plan", 2)
+
+    metadata = svc.get_audit_trail_as_metadata("meta-plan")
+    assert "guardrail_audit_trail" in metadata
+    assert "autonomy_guardrails" in metadata
+
+    svc2 = AutonomyGuardrailService()
+    svc2.load_from_metadata("meta-plan", metadata)
+    g = svc2.get_guardrails("meta-plan")
+    assert g is not None
+    assert g.max_steps == 10
+    print("metadata-ok")
+
+
+def _test_summary() -> None:
+    """Summary of autonomy guardrail features."""
+    g = AutonomyGuardrails(
+        max_steps=100,
+        tool_budget=500.0,
+        required_confirmations=["deploy", "delete"],
+    )
+    assert g.max_steps == 100
+    assert g.tool_budget == 500.0
+    assert len(g.required_confirmations) == 2
+    g.increment_step()
+    assert g.step_count == 1
+    g.record_cost(10.0)
+    assert g.budget_spent == 10.0
+    trail = GuardrailAuditTrail()
+    entry = GuardrailAuditEntry(
+        event_type=GuardrailEventType.STEP_ALLOWED,
+        guard_name="step_limit",
+        result=GuardrailResult.ALLOWED,
+    )
+    trail.add_entry(entry)
+    assert len(trail.entries) == 1
+    print("summary-ok")
+
+
+def _test_wall_clock_allow() -> None:
+    """Wall-clock allows when within limit."""
+    g = AutonomyGuardrails(max_wall_clock_seconds=600.0)
+    g.mark_started()
+    allowed, _ = g.check_wall_clock()
+    assert allowed is True
+    print("wall-clock-allow-ok")
+
+
+def _test_wall_clock_block() -> None:
+    """Wall-clock blocks when expired."""
+    g = AutonomyGuardrails(max_wall_clock_seconds=0.001)
+    g.start_time = (datetime.now(tz=UTC) - timedelta(seconds=10)).isoformat()
+    allowed, reason = g.check_wall_clock()
+    assert allowed is False
+    assert reason is not None
+    assert "Wall-clock" in reason
+    print("wall-clock-block-ok")
+
+
+def _test_actor_limit() -> None:
+    """Actor tool-call limit enforcement."""
+    g = AutonomyGuardrails(
+        actor_limits=ActorLimits(max_tool_calls_per_invocation=5),
+    )
+    allowed, _ = g.check_actor_tool_calls(3)
+    assert allowed is True
+    blocked, reason = g.check_actor_tool_calls(5)
+    assert blocked is False
+    assert reason is not None
+    assert "Actor" in reason
+    print("actor-limit-ok")
+
+
+def _test_eviction() -> None:
+    """Audit trail evicts oldest entries when full."""
+    trail = GuardrailAuditTrail(max_entries=3)
+    for i in range(5):
+        entry = GuardrailAuditEntry(
+            event_type=GuardrailEventType.STEP_ALLOWED,
+            guard_name=f"step_{i}",
+            result=GuardrailResult.ALLOWED,
+        )
+        trail.add_entry(entry)
+    assert len(trail.entries) == 3
+    # Oldest two (step_0, step_1) should be evicted
+    assert trail.entries[0].guard_name == "step_2"
+    assert trail.allowed_count == 3
+    print("eviction-ok")
+
+
+def _test_zero_budget() -> None:
+    """Zero budget blocks any positive cost."""
+    g = AutonomyGuardrails(tool_budget=0.0)
+    allowed, _ = g.check_tool_budget(0.0)
+    assert allowed is True
+    blocked, reason = g.check_tool_budget(0.01)
+    assert blocked is False
+    assert reason is not None
+    print("zero-budget-ok")
+
+
+if __name__ == "__main__":
+    cmd = sys.argv[1] if len(sys.argv) > 1 else "summary"
+    dispatch: dict[str, Any] = {
+        "step-allow": _test_step_allow,
+        "step-block": _test_step_block,
+        "budget-allow": _test_budget_allow,
+        "budget-block": _test_budget_block,
+        "confirm-required": _test_confirm_required,
+        "audit-trail": _test_audit_trail,
+        "metadata": _test_metadata,
+        "wall-clock-allow": _test_wall_clock_allow,
+        "wall-clock-block": _test_wall_clock_block,
+        "actor-limit": _test_actor_limit,
+        "eviction": _test_eviction,
+        "zero-budget": _test_zero_budget,
+        "summary": _test_summary,
+    }
+    fn = dispatch.get(cmd)
+    if fn:
+        fn()
+    else:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        sys.exit(1)

src/cleveragents/application/container.py

@@ -11,6 +11,9 @@ from dependency_injector import containers, providers
 from cleveragents.actor.registry import ActorRegistry
 from cleveragents.application.reactive_registry_adapter import register_registry_agents
 from cleveragents.application.services.actor_service import ActorService
+from cleveragents.application.services.autonomy_guardrail_service import (
+    AutonomyGuardrailService,
+)
 from cleveragents.application.services.context_service import ContextService
 from cleveragents.application.services.decision_service import DecisionService
 from cleveragents.application.services.plan_lifecycle_service import (
@@ -250,6 +253,11 @@ class Container(containers.DeclarativeContainer):
         database_url=database_url,
     )
 
+    # Autonomy Guardrail Service - Singleton so all callers share state
+    autonomy_guardrail_service = providers.Singleton(
+        AutonomyGuardrailService,
+    )
+
     # Reactive routing
     stream_router = providers.Singleton(ReactiveStreamRouter)
     langgraph_bridge = providers.Singleton(

src/cleveragents/application/services/__init__.py

@@ -3,6 +3,9 @@
 Contains service classes that orchestrate business operations.
 """
 
+from cleveragents.application.services.autonomy_guardrail_service import (
+    AutonomyGuardrailService,
+)
 from cleveragents.application.services.config_service import (
     ConfigEntry,
     ConfigLevel,
@@ -82,6 +85,7 @@ __all__ = [
     "ApplyValidationResult",
     "ApplyValidationSummary",
     "AttachmentScope",
+    "AutonomyGuardrailService",
     "BrokenReferenceRule",
     "ConfigEntry",
     "ConfigLevel",

src/cleveragents/application/services/autonomy_guardrail_service.py

@@ -0,0 +1,524 @@
+"""Autonomy guardrail enforcement service.
+
+Provides a high-level service interface for checking autonomy
+guardrails during plan execution and recording enforcement
+events to the plan's audit trail.
+
+The service integrates with the plan metadata to persist audit
+trail entries, providing a complete record of guardrail decisions.
+
+Thread-safe: all state mutations are protected by a reentrant lock
+so the service can be shared across threads (e.g. concurrent plan
+step evaluations).
+"""
+
+from __future__ import annotations
+
+import logging
+import threading
+from typing import Any
+
+from cleveragents.domain.models.core.autonomy_guardrails import (
+    AutonomyGuardrails,
+    GuardrailAuditEntry,
+    GuardrailAuditTrail,
+    GuardrailEventType,
+    GuardrailResult,
+)
+
+logger = logging.getLogger(__name__)
+
+# Key used in plan metadata to store the audit trail
+_AUDIT_TRAIL_KEY = "guardrail_audit_trail"
+
+# Key used in plan metadata to store guardrail state
+_GUARDRAILS_KEY = "autonomy_guardrails"
+
+# Hard ceiling for entries accepted through load_from_metadata to
+# prevent memory exhaustion from untrusted or corrupted payloads.
+_MAX_METADATA_ENTRIES = 50_000
+
+# Hard ceiling for required_confirmations list length.
+_MAX_CONFIRMATIONS = 500
+
+
+class AutonomyGuardrailService:
+    """Service for enforcing autonomy guardrails during plan execution.
+
+    Manages guardrail state and audit trail persistence via plan
+    metadata dictionaries. Each plan can have its own guardrail
+    configuration and audit trail stored in its metadata.
+
+    All public methods are thread-safe.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the autonomy guardrail service."""
+        self._guardrails: dict[str, AutonomyGuardrails] = {}
+        self._audit_trails: dict[str, GuardrailAuditTrail] = {}
+        self._lock = threading.RLock()
+
+    def configure_guardrails(
+        self,
+        plan_id: str,
+        guardrails: AutonomyGuardrails,
+    ) -> None:
+        """Configure guardrails for a plan.
+
+        Args:
+            plan_id: The plan identifier.
+            guardrails: The guardrail configuration to apply.
+        """
+        if not plan_id:
+            raise ValueError("plan_id must not be empty")
+        with self._lock:
+            self._guardrails[plan_id] = guardrails
+            if plan_id not in self._audit_trails:
+                self._audit_trails[plan_id] = GuardrailAuditTrail()
+        logger.debug("Configured guardrails for plan %s", plan_id)
+
+    def get_guardrails(self, plan_id: str) -> AutonomyGuardrails | None:
+        """Retrieve the guardrail configuration for a plan.
+
+        Args:
+            plan_id: The plan identifier.
+
+        Returns:
+            The guardrail configuration, or None if not configured.
+        """
+        with self._lock:
+            return self._guardrails.get(plan_id)
+
+    def check_step_limit(
+        self,
+        plan_id: str,
+        current_step: int,
+    ) -> bool:
+        """Check if the plan has exceeded its maximum step limit.
+
+        Args:
+            plan_id: The plan identifier.
+            current_step: The current step number.
+
+        Returns:
+            True if the step is allowed, False if blocked.
+        """
+        with self._lock:
+            guardrails = self._guardrails.get(plan_id)
+            if guardrails is None:
+                return True
+
+            guardrails.step_count = current_step
+            allowed, reason = guardrails.check_step_limit()
+
+            event_type = (
+                GuardrailEventType.STEP_ALLOWED
+                if allowed
+                else GuardrailEventType.STEP_BLOCKED
+            )
+            result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED
+
+            entry = GuardrailAuditEntry(
+                event_type=event_type,
+                guard_name="step_limit",
+                result=result,
+                reason=reason,
+                context={
+                    "current_step": current_step,
+                    "max_steps": guardrails.max_steps,
+                },
+            )
+            self._record_enforcement_locked(plan_id, entry)
+
+            if not allowed:
+                logger.warning(
+                    "Step limit exceeded for plan %s: %s",
+                    plan_id,
+                    reason,
+                )
+
+            return allowed
+
+    def check_tool_budget(
+        self,
+        plan_id: str,
+        tool_cost: float,
+    ) -> bool:
+        """Check if the plan's budget allows a tool invocation.
+
+        When allowed, the cost is recorded **before** the audit entry
+        so that the trail reflects the post-deduction budget state.
+
+        Args:
+            plan_id: The plan identifier.
+            tool_cost: The cost of the proposed tool invocation.
+
+        Returns:
+            True if the invocation is within budget, False if blocked.
+        """
+        with self._lock:
+            guardrails = self._guardrails.get(plan_id)
+            if guardrails is None:
+                return True
+
+            allowed, reason = guardrails.check_tool_budget(tool_cost)
+
+            # Record cost *before* audit entry so context is accurate
+            if allowed:
+                guardrails.record_cost(tool_cost)
+
+            event_type = (
+                GuardrailEventType.BUDGET_ALLOWED
+                if allowed
+                else GuardrailEventType.BUDGET_BLOCKED
+            )
+            result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED
+
+            entry = GuardrailAuditEntry(
+                event_type=event_type,
+                guard_name="tool_budget",
+                result=result,
+                reason=reason,
+                context={
+                    "tool_cost": tool_cost,
+                    "budget_spent": guardrails.budget_spent,
+                    "tool_budget": guardrails.tool_budget,
+                },
+            )
+            self._record_enforcement_locked(plan_id, entry)
+
+            if not allowed:
+                logger.warning(
+                    "Tool budget exceeded for plan %s: %s",
+                    plan_id,
+                    reason,
+                )
+
+            return allowed
+
+    def check_wall_clock(self, plan_id: str) -> bool:
+        """Check if the plan has exceeded its wall-clock time limit.
+
+        Args:
+            plan_id: The plan identifier.
+
+        Returns:
+            True if the plan is within the time limit, False if blocked.
+        """
+        with self._lock:
+            guardrails = self._guardrails.get(plan_id)
+            if guardrails is None:
+                return True
+
+            allowed, reason = guardrails.check_wall_clock()
+
+            event_type = (
+                GuardrailEventType.TIME_ALLOWED
+                if allowed
+                else GuardrailEventType.TIME_BLOCKED
+            )
+            result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED
+
+            entry = GuardrailAuditEntry(
+                event_type=event_type,
+                guard_name="wall_clock",
+                result=result,
+                reason=reason,
+                context={
+                    "max_wall_clock_seconds": guardrails.max_wall_clock_seconds,
+                },
+            )
+            self._record_enforcement_locked(plan_id, entry)
+
+            if not allowed:
+                logger.warning(
+                    "Wall-clock limit exceeded for plan %s: %s",
+                    plan_id,
+                    reason,
+                )
+
+            return allowed
+
+    def check_actor_tool_calls(
+        self,
+        plan_id: str,
+        current_calls: int,
+    ) -> bool:
+        """Check if the actor has exceeded its per-invocation tool call limit.
+
+        Args:
+            plan_id: The plan identifier.
+            current_calls: Number of tool calls made so far.
+
+        Returns:
+            True if allowed, False if blocked.
+        """
+        with self._lock:
+            guardrails = self._guardrails.get(plan_id)
+            if guardrails is None:
+                return True
+
+            allowed, reason = guardrails.check_actor_tool_calls(current_calls)
+
+            event_type = (
+                GuardrailEventType.ACTOR_LIMIT_ALLOWED
+                if allowed
+                else GuardrailEventType.ACTOR_LIMIT_BLOCKED
+            )
+            result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED
+
+            entry = GuardrailAuditEntry(
+                event_type=event_type,
+                guard_name="actor_tool_calls",
+                result=result,
+                reason=reason,
+                context={
+                    "current_calls": current_calls,
+                    "max_tool_calls_per_invocation": (
+                        guardrails.actor_limits.max_tool_calls_per_invocation
+                    ),
+                },
+            )
+            self._record_enforcement_locked(plan_id, entry)
+
+            if not allowed:
+                logger.warning(
+                    "Actor tool-call limit exceeded for plan %s: %s",
+                    plan_id,
+                    reason,
+                )
+
+            return allowed
+
+    def check_retries_per_failure(
+        self,
+        plan_id: str,
+        current_retries: int,
+    ) -> bool:
+        """Check if the retry count exceeds the per-failure retry limit.
+
+        Args:
+            plan_id: The plan identifier.
+            current_retries: Number of retries attempted so far.
+
+        Returns:
+            True if the retry is allowed, False if blocked.
+        """
+        with self._lock:
+            guardrails = self._guardrails.get(plan_id)
+            if guardrails is None:
+                return True
+
+            allowed, reason = guardrails.check_retries_per_failure(current_retries)
+
+            event_type = (
+                GuardrailEventType.RETRY_ALLOWED
+                if allowed
+                else GuardrailEventType.RETRY_BLOCKED
+            )
+            result = GuardrailResult.ALLOWED if allowed else GuardrailResult.DENIED
+
+            entry = GuardrailAuditEntry(
+                event_type=event_type,
+                guard_name="retries_per_failure",
+                result=result,
+                reason=reason,
+                context={
+                    "current_retries": current_retries,
+                    "max_retries_per_failure": (
+                        guardrails.actor_limits.max_retries_per_failure
+                    ),
+                },
+            )
+            self._record_enforcement_locked(plan_id, entry)
+
+            if not allowed:
+                logger.warning(
+                    "Retry limit exceeded for plan %s: %s",
+                    plan_id,
+                    reason,
+                )
+
+            return allowed
+
+    def check_confirmation_required(
+        self,
+        plan_id: str,
+        operation: str,
+    ) -> bool:
+        """Check if an operation requires human confirmation.
+
+        Args:
+            plan_id: The plan identifier.
+            operation: The operation name to check.
+
+        Returns:
+            True if confirmation is required, False otherwise.
+        """
+        with self._lock:
+            guardrails = self._guardrails.get(plan_id)
+            if guardrails is None:
+                return False
+
+            needs_confirmation, reason = guardrails.check_confirmation_required(
+                operation,
+            )
+
+            if needs_confirmation:
+                event_type = GuardrailEventType.CONFIRMATION_REQUIRED
+                result = GuardrailResult.DENIED
+            else:
+                event_type = GuardrailEventType.CONFIRMATION_GRANTED
+                result = GuardrailResult.ALLOWED
+
+            entry = GuardrailAuditEntry(
+                event_type=event_type,
+                guard_name="confirmation",
+                result=result,
+                reason=reason,
+                context={"operation": operation},
+            )
+            self._record_enforcement_locked(plan_id, entry)
+
+            if needs_confirmation:
+                logger.info(
+                    "Confirmation required for plan %s operation '%s': %s",
+                    plan_id,
+                    operation,
+                    reason,
+                )
+
+            return needs_confirmation
+
+    def record_enforcement(
+        self,
+        plan_id: str,
+        entry: GuardrailAuditEntry,
+    ) -> None:
+        """Persist an audit entry to the plan's audit trail.
+
+        Args:
+            plan_id: The plan identifier.
+            entry: The audit entry to record.
+        """
+        with self._lock:
+            self._record_enforcement_locked(plan_id, entry)
+
+    def get_audit_trail(self, plan_id: str) -> GuardrailAuditTrail:
+        """Retrieve the audit trail for a plan.
+
+        Args:
+            plan_id: The plan identifier.
+
+        Returns:
+            The audit trail for the plan. Returns an empty trail
+            if no events have been recorded.
+        """
+        with self._lock:
+            return self._audit_trails.get(plan_id, GuardrailAuditTrail())
+
+    def get_audit_trail_as_metadata(
+        self,
+        plan_id: str,
+    ) -> dict[str, Any]:
+        """Serialize the audit trail for persistence in plan metadata.
+
+        Args:
+            plan_id: The plan identifier.
+
+        Returns:
+            A dictionary suitable for storing in plan metadata.
+        """
+        with self._lock:
+            trail = self._audit_trails.get(plan_id, GuardrailAuditTrail())
+            guardrails = self._guardrails.get(plan_id)
+
+            metadata: dict[str, Any] = {
+                _AUDIT_TRAIL_KEY: trail.model_dump(),
+            }
+            if guardrails is not None:
+                metadata[_GUARDRAILS_KEY] = guardrails.model_dump()
+            return metadata
+
+    def load_from_metadata(
+        self,
+        plan_id: str,
+        metadata: dict[str, Any],
+    ) -> None:
+        """Restore guardrail state from plan metadata.
+
+        Applies size guards to prevent memory exhaustion from
+        oversized or malicious payloads.
+
+        Args:
+            plan_id: The plan identifier.
+            metadata: The plan metadata dictionary.
+
+        Raises:
+            ValueError: If the metadata contains oversized payloads.
+        """
+        with self._lock:
+            if _GUARDRAILS_KEY in metadata:
+                raw_guardrails = metadata[_GUARDRAILS_KEY]
+                if isinstance(raw_guardrails, dict):
+                    confirmations = raw_guardrails.get(
+                        "required_confirmations",
+                        [],
+                    )
+                    if (
+                        isinstance(confirmations, list)
+                        and len(confirmations) > _MAX_CONFIRMATIONS
+                    ):
+                        raise ValueError(
+                            f"required_confirmations exceeds maximum of "
+                            f"{_MAX_CONFIRMATIONS} entries"
+                        )
+                self._guardrails[plan_id] = AutonomyGuardrails.model_validate(
+                    raw_guardrails,
+                )
+
+            if _AUDIT_TRAIL_KEY in metadata:
+                raw_trail = metadata[_AUDIT_TRAIL_KEY]
+                if isinstance(raw_trail, dict):
+                    entries = raw_trail.get("entries", [])
+                    if (
+                        isinstance(entries, list)
+                        and len(entries) > _MAX_METADATA_ENTRIES
+                    ):
+                        raise ValueError(
+                            f"Audit trail exceeds maximum of "
+                            f"{_MAX_METADATA_ENTRIES} entries"
+                        )
+                self._audit_trails[plan_id] = GuardrailAuditTrail.model_validate(
+                    raw_trail
+                )
+
+    def remove_plan(self, plan_id: str) -> None:
+        """Remove all guardrail state for a plan.
+
+        Args:
+            plan_id: The plan identifier.
+        """
+        with self._lock:
+            self._guardrails.pop(plan_id, None)
+            self._audit_trails.pop(plan_id, None)
+
+    # ------------------------------------------------------------------
+    # Private helpers
+    # ------------------------------------------------------------------
+
+    def _record_enforcement_locked(
+        self,
+        plan_id: str,
+        entry: GuardrailAuditEntry,
+    ) -> None:
+        """Record an audit entry (caller must hold ``_lock``)."""
+        if plan_id not in self._audit_trails:
+            self._audit_trails[plan_id] = GuardrailAuditTrail()
+        self._audit_trails[plan_id].add_entry(entry)
+        logger.debug(
+            "Recorded guardrail event for plan %s: %s (%s)",
+            plan_id,
+            entry.event_type.value,
+            entry.result.value,
+        )

src/cleveragents/application/services/plan_executor.py

@@ -19,6 +19,9 @@ import structlog
 from pydantic import BaseModel, ConfigDict, Field
 from ulid import ULID
 
+from cleveragents.application.services.autonomy_guardrail_service import (
+    AutonomyGuardrailService,
+)
 from cleveragents.application.services.plan_execution_context import (
     PlanExecutionContext,
     RuntimeExecuteActor,
@@ -259,6 +262,7 @@ class PlanExecutor:
         execution_context: PlanExecutionContext | None = None,
         error_recovery_service: ErrorRecoveryService | None = None,
         checkpoint_manager: CheckpointManager | None = None,
+        guardrail_service: AutonomyGuardrailService | None = None,
     ) -> None:
         """Initialize the plan executor.
 
@@ -272,6 +276,8 @@ class PlanExecutor:
             checkpoint_manager: Optional checkpoint manager for
                 sandbox state snapshots during execute.  When ``None``,
                 checkpoint hooks are silently skipped.
+            guardrail_service: Optional autonomy guardrail service for
+                enforcing step limits, budgets, and wall-clock time.
         """
         if lifecycle_service is None:
             raise ValidationError("lifecycle_service must not be None")
@@ -281,6 +287,7 @@ class PlanExecutor:
         self._execution_context = execution_context
         self._error_recovery = error_recovery_service
         self._checkpoint_manager = checkpoint_manager
+        self._guardrail_service = guardrail_service
         self._strategize_actor = StrategizeStubActor()
         self._execute_actor = ExecuteStubActor()
         self._logger = logger.bind(service="plan_executor")
@@ -520,8 +527,49 @@ class PlanExecutor:
                 f"Plan {plan_id} has no decision tree. "
                 "Strategize must complete before Execute."
             )
+        self._enforce_guardrails(plan_id)
         return plan
 
+    def _enforce_guardrails(self, plan_id: str) -> None:
+        """Pre-flight guardrail checks before execution begins.
+
+        Marks the execution start time (for wall-clock tracking) if
+        not already set, then checks wall-clock time and current step
+        limit.  Raises ``PlanError`` if any guardrail blocks execution.
+        """
+        if self._guardrail_service is None:
+            return
+        guardrails = self._guardrail_service.get_guardrails(plan_id)
+        if guardrails is not None and guardrails.start_time is None:
+            guardrails.mark_started()
+        if not self._guardrail_service.check_wall_clock(plan_id):
+            raise PlanError(f"Guardrail wall-clock limit exceeded for plan {plan_id}")
+        if guardrails is not None:
+            current_step = guardrails.step_count
+            if not self._guardrail_service.check_step_limit(plan_id, current_step):
+                raise PlanError(
+                    f"Guardrail step limit already reached for plan {plan_id}"
+                )
+
+    def _enforce_guardrails_per_step(self, plan_id: str) -> None:
+        """Per-decision guardrail enforcement.
+
+        Increments the step counter and checks step limit and
+        wall-clock time. Raises ``PlanError`` if blocked.
+        """
+        if self._guardrail_service is None:
+            return
+        guardrails = self._guardrail_service.get_guardrails(plan_id)
+        if guardrails is None:
+            return  # pragma: no cover - defensive
+        next_step = guardrails.step_count + 1
+        if not self._guardrail_service.check_step_limit(plan_id, next_step):
+            raise PlanError(
+                f"Guardrail step limit reached for plan {plan_id} at step {next_step}"
+            )
+        if not self._guardrail_service.check_wall_clock(plan_id):
+            raise PlanError(f"Guardrail wall-clock limit exceeded for plan {plan_id}")
+
     def _run_execute_with_runtime(
         self,
         plan_id: str,
@@ -542,6 +590,10 @@ class PlanExecutor:
         self._lifecycle.start_execute(plan_id)
         self._try_create_checkpoint(plan_id, "pre_execute")
         try:
+            # Enforce per-step guardrails for each decision
+            for _decision in decisions:
+                self._enforce_guardrails_per_step(plan_id)
+
             result = runtime_actor.execute(
                 decisions=decisions, stream_callback=stream_callback
             )
@@ -600,6 +652,10 @@ class PlanExecutor:
 
         for attempt in range(max_attempts):
             try:
+                # Enforce per-step guardrails for each decision
+                for _decision in decisions:
+                    self._enforce_guardrails_per_step(plan_id)
+
                 result = self._execute_actor.execute(
                     plan_id=plan_id,
                     decisions=decisions,

src/cleveragents/domain/models/core/__init__.py

@@ -8,6 +8,14 @@ from cleveragents.domain.models.core.automation_profile import (
     GuardResult,
     get_builtin_profile,
 )
+from cleveragents.domain.models.core.autonomy_guardrails import (
+    ActorLimits,
+    AutonomyGuardrails,
+    GuardrailAuditEntry,
+    GuardrailAuditTrail,
+    GuardrailEventType,
+    GuardrailResult,
+)
 from cleveragents.domain.models.core.change import (
     Change,
     ChangeEntry,
@@ -224,8 +232,10 @@ __all__ = [
     "ROLE_PERMISSIONS",
     "ActionState",
     "Actor",
+    "ActorLimits",
     "AutomationGuard",
     "AutomationProfile",
+    "AutonomyGuardrails",
     "BindingMode",
     "BindingResult",
     "Change",
@@ -266,6 +276,10 @@ __all__ = [
     "ErrorRecord",
     "ErrorRecoveryPolicy",
     "GuardResult",
+    "GuardrailAuditEntry",
+    "GuardrailAuditTrail",
+    "GuardrailEventType",
+    "GuardrailResult",
     "InMemoryChangeSetStore",
     "InMemoryInvocationTracker",
     "Invariant",

src/cleveragents/domain/models/core/autonomy_guardrails.py

@@ -0,0 +1,458 @@
+"""Autonomy guardrails and audit trail models.
+
+Extends the automation guard framework with:
+
+- **AutonomyGuardrails** — runtime constraints for max steps, tool
+  budget, wall-clock time, per-actor limits, and required confirmations
+  during plan execution.
+- **GuardrailAuditEntry** — a single audit trail entry recording a
+  guardrail enforcement event with timestamp, event type, result, and
+  contextual metadata.
+- **GuardrailAuditTrail** — a bounded, ordered collection of audit
+  entries persisted to plan metadata with configurable maximum size
+  and oldest-first eviction.
+
+These models support the M6 autonomy hardening milestone by providing
+a structured, auditable record of every guardrail decision made during
+plan execution.
+
+Based on ``docs/specification.md`` Section "Automation Profiles"
+(autonomy guardrails and audit trail) and "Cost and Rate Limits"
+(per-plan budgets including wall-clock time, per-actor limits).
+"""
+
+from __future__ import annotations
+
+import threading
+from datetime import UTC, datetime
+from enum import StrEnum
+
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
+
+
+class GuardrailEventType(StrEnum):
+    """Types of guardrail enforcement events."""
+
+    STEP_BLOCKED = "step_blocked"
+    BUDGET_BLOCKED = "budget_blocked"
+    CONFIRMATION_REQUIRED = "confirmation_required"
+    STEP_ALLOWED = "step_allowed"
+    BUDGET_ALLOWED = "budget_allowed"
+    CONFIRMATION_GRANTED = "confirmation_granted"
+    TIME_BLOCKED = "time_blocked"
+    TIME_ALLOWED = "time_allowed"
+    ACTOR_LIMIT_BLOCKED = "actor_limit_blocked"
+    ACTOR_LIMIT_ALLOWED = "actor_limit_allowed"
+    RETRY_BLOCKED = "retry_blocked"
+    RETRY_ALLOWED = "retry_allowed"
+
+
+class GuardrailResult(StrEnum):
+    """Outcome of a guardrail enforcement check."""
+
+    ALLOWED = "allowed"
+    DENIED = "denied"
+
+
+class GuardrailAuditEntry(BaseModel):
+    """A single guardrail enforcement audit trail entry.
+
+    Records the outcome of a guardrail check including the event type,
+    which guard was evaluated, whether the action was allowed or denied,
+    and optional contextual metadata.
+    """
+
+    timestamp: str = Field(
+        default_factory=lambda: datetime.now(tz=UTC).isoformat(),
+        description="ISO-8601 timestamp of the enforcement event",
+    )
+    event_type: GuardrailEventType = Field(
+        ...,
+        description="Type of guardrail event",
+    )
+    guard_name: str = Field(
+        ...,
+        min_length=1,
+        description="Name of the guard that was evaluated",
+    )
+    result: GuardrailResult = Field(
+        ...,
+        description="Outcome of the guard check (allowed/denied)",
+    )
+    reason: str | None = Field(
+        default=None,
+        description="Human-readable reason for the decision",
+    )
+    context: dict[str, str | int | float | bool | None] = Field(
+        default_factory=dict,
+        description="Additional context about the enforcement event",
+    )
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+    )
+
+
+_DEFAULT_MAX_AUDIT_ENTRIES = 10_000
+
+
+class GuardrailAuditTrail(BaseModel):
+    """Bounded, ordered collection of guardrail audit entries.
+
+    Persisted to plan metadata to provide a complete record of
+    autonomy-related decisions during plan execution.
+
+    When the number of entries exceeds ``max_entries``, the oldest
+    entries are evicted to keep the trail bounded.
+    """
+
+    entries: list[GuardrailAuditEntry] = Field(
+        default_factory=list,
+        description="Ordered list of guardrail enforcement entries",
+    )
+    max_entries: int = Field(
+        default=_DEFAULT_MAX_AUDIT_ENTRIES,
+        ge=1,
+        le=1_000_000,
+        description="Maximum number of audit entries to retain",
+    )
+    _allowed_count: int = 0
+    _denied_count: int = 0
+
+    def model_post_init(self, _context: object) -> None:
+        """Recompute counters from pre-existing entries after init."""
+        self._allowed_count = sum(
+            1 for e in self.entries if e.result == GuardrailResult.ALLOWED
+        )
+        self._denied_count = sum(
+            1 for e in self.entries if e.result == GuardrailResult.DENIED
+        )
+
+    def add_entry(self, entry: GuardrailAuditEntry) -> None:
+        """Append an audit entry, evicting the oldest if at capacity."""
+        if len(self.entries) >= self.max_entries:
+            evicted = self.entries.pop(0)
+            if evicted.result == GuardrailResult.ALLOWED:
+                self._allowed_count -= 1
+            else:
+                self._denied_count -= 1
+        self.entries.append(entry)
+        if entry.result == GuardrailResult.ALLOWED:
+            self._allowed_count += 1
+        else:
+            self._denied_count += 1
+
+    @property
+    def denied_count(self) -> int:
+        """Count of denied enforcement events."""
+        return self._denied_count
+
+    @property
+    def allowed_count(self) -> int:
+        """Count of allowed enforcement events."""
+        return self._allowed_count
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+    )
+
+
+class ActorLimits(BaseModel):
+    """Per-actor runtime limits (spec line 27891).
+
+    Controls the maximum number of tool calls per actor invocation and
+    the maximum number of retries per tool failure within a single actor
+    step.
+    """
+
+    max_tool_calls_per_invocation: int | None = Field(
+        default=None,
+        description=(
+            "Maximum tool calls allowed per actor invocation. None means unlimited."
+        ),
+    )
+    max_retries_per_failure: int | None = Field(
+        default=None,
+        description=(
+            "Maximum retries per tool failure within an actor step. "
+            "None means unlimited."
+        ),
+    )
+
+    @field_validator("max_tool_calls_per_invocation")
+    @classmethod
+    def validate_max_tool_calls(
+        cls: type[ActorLimits],
+        v: int | None,
+    ) -> int | None:
+        """Validate max_tool_calls_per_invocation is positive if set."""
+        if v is not None and v < 1:
+            raise ValueError(f"max_tool_calls_per_invocation must be >= 1, got {v}")
+        return v
+
+    @field_validator("max_retries_per_failure")
+    @classmethod
+    def validate_max_retries(
+        cls: type[ActorLimits],
+        v: int | None,
+    ) -> int | None:
+        """Validate max_retries_per_failure is non-negative if set."""
+        if v is not None and v < 0:
+            raise ValueError(f"max_retries_per_failure must be >= 0, got {v}")
+        return v
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+    )
+
+
+class AutonomyGuardrails(BaseModel):
+    """Runtime autonomy constraints for plan execution.
+
+    Extends the automation guard framework with step limits, tool
+    budgets, wall-clock time limits, per-actor limits, and required
+    confirmation gates.  Tracks current step count and budget spent
+    for enforcement decisions.
+    """
+
+    max_steps: int | None = Field(
+        default=None,
+        description=(
+            "Maximum number of execution steps allowed. None means unlimited."
+        ),
+    )
+    tool_budget: float | None = Field(
+        default=None,
+        description=("Maximum cost budget for tool invocations. None means unlimited."),
+    )
+    max_wall_clock_seconds: float | None = Field(
+        default=None,
+        description=(
+            "Maximum wall-clock time in seconds for plan execution. "
+            "None means unlimited."
+        ),
+    )
+    actor_limits: ActorLimits = Field(
+        default_factory=ActorLimits,
+        description="Per-actor runtime limits",
+    )
+    required_confirmations: list[str] = Field(
+        default_factory=list,
+        description=(
+            "List of operation names requiring human confirmation before execution."
+        ),
+    )
+    step_count: int = Field(
+        default=0,
+        ge=0,
+        description="Current execution step counter",
+    )
+    budget_spent: float = Field(
+        default=0.0,
+        ge=0.0,
+        description="Current cumulative budget spent on tool invocations",
+    )
+    start_time: str | None = Field(
+        default=None,
+        description="ISO-8601 timestamp of when plan execution started",
+    )
+
+    @field_validator("start_time")
+    @classmethod
+    def validate_start_time(
+        cls: type[AutonomyGuardrails],
+        v: str | None,
+    ) -> str | None:
+        """Validate start_time is a parseable ISO-8601 string if set."""
+        if v is not None:
+            try:
+                datetime.fromisoformat(v)
+            except (ValueError, TypeError) as exc:
+                raise ValueError(
+                    f"start_time must be a valid ISO-8601 timestamp, got {v!r}"
+                ) from exc
+        return v
+
+    @field_validator("max_steps")
+    @classmethod
+    def validate_max_steps(
+        cls: type[AutonomyGuardrails],
+        v: int | None,
+    ) -> int | None:
+        """Validate max_steps is positive if set."""
+        if v is not None and v < 1:
+            raise ValueError(f"max_steps must be >= 1, got {v}")
+        return v
+
+    @field_validator("tool_budget")
+    @classmethod
+    def validate_tool_budget(
+        cls: type[AutonomyGuardrails],
+        v: float | None,
+    ) -> float | None:
+        """Validate tool_budget is non-negative if set."""
+        if v is not None and v < 0.0:
+            raise ValueError(f"tool_budget must be >= 0.0, got {v}")
+        return v
+
+    @field_validator("max_wall_clock_seconds")
+    @classmethod
+    def validate_max_wall_clock(
+        cls: type[AutonomyGuardrails],
+        v: float | None,
+    ) -> float | None:
+        """Validate max_wall_clock_seconds is positive if set."""
+        if v is not None and v <= 0.0:
+            raise ValueError(f"max_wall_clock_seconds must be > 0.0, got {v}")
+        return v
+
+    @model_validator(mode="after")
+    def _normalise_confirmations(self) -> AutonomyGuardrails:
+        """Normalise required_confirmations to lowercase for matching."""
+        lowered = [c.lower() for c in self.required_confirmations]
+        if lowered != self.required_confirmations:
+            object.__setattr__(self, "required_confirmations", lowered)
+        return self
+
+    def check_step_limit(self) -> tuple[bool, str | None]:
+        """Check if the current step count exceeds the max step limit.
+
+        Returns:
+            A tuple of (allowed, reason). If allowed is False, reason
+            contains a human-readable explanation.
+        """
+        if self.max_steps is not None and self.step_count >= self.max_steps:
+            return (
+                False,
+                f"Step limit reached: {self.step_count}/{self.max_steps}",
+            )
+        return (True, None)
+
+    def check_tool_budget(self, tool_cost: float) -> tuple[bool, str | None]:
+        """Check if the budget allows a tool invocation of the given cost.
+
+        Args:
+            tool_cost: The cost of the proposed tool invocation.
+
+        Returns:
+            A tuple of (allowed, reason).
+        """
+        if tool_cost < 0.0:
+            raise ValueError(f"tool_cost must be >= 0.0, got {tool_cost}")
+        if self.tool_budget is not None:
+            projected = self.budget_spent + tool_cost
+            if projected > self.tool_budget:
+                return (
+                    False,
+                    f"Budget exceeded: {projected:.2f} > {self.tool_budget:.2f}",
+                )
+        return (True, None)
+
+    def check_wall_clock(self) -> tuple[bool, str | None]:
+        """Check if the plan has exceeded its wall-clock time limit.
+
+        Returns:
+            A tuple of (allowed, reason).
+        """
+        if self.max_wall_clock_seconds is None or self.start_time is None:
+            return (True, None)
+        try:
+            elapsed = (
+                datetime.now(UTC) - datetime.fromisoformat(self.start_time)
+            ).total_seconds()
+        except (ValueError, TypeError):
+            return (
+                False,
+                f"Wall-clock check failed: malformed start_time {self.start_time!r}",
+            )
+        if elapsed >= self.max_wall_clock_seconds:
+            return (
+                False,
+                f"Wall-clock limit reached: {elapsed:.1f}s >= "
+                f"{self.max_wall_clock_seconds:.1f}s",
+            )
+        return (True, None)
+
+    def check_actor_tool_calls(
+        self,
+        current_calls: int,
+    ) -> tuple[bool, str | None]:
+        """Check if the actor has exceeded its per-invocation tool call limit.
+
+        Args:
+            current_calls: Number of tool calls made so far in this invocation.
+
+        Returns:
+            A tuple of (allowed, reason).
+        """
+        limit = self.actor_limits.max_tool_calls_per_invocation
+        if limit is not None and current_calls >= limit:
+            return (
+                False,
+                f"Actor tool-call limit reached: {current_calls}/{limit}",
+            )
+        return (True, None)
+
+    def check_retries_per_failure(
+        self,
+        current_retries: int,
+    ) -> tuple[bool, str | None]:
+        """Check if the current retry count exceeds the per-failure retry limit.
+
+        Args:
+            current_retries: Number of retries attempted so far for this failure.
+
+        Returns:
+            A tuple of (allowed, reason).
+        """
+        limit = self.actor_limits.max_retries_per_failure
+        if limit is not None and current_retries >= limit:
+            return (
+                False,
+                f"Retry limit reached: {current_retries}/{limit}",
+            )
+        return (True, None)
+
+    def check_confirmation_required(self, operation: str) -> tuple[bool, str | None]:
+        """Check if an operation requires human confirmation.
+
+        Args:
+            operation: The operation name to check (case-insensitive).
+
+        Returns:
+            A tuple of (confirmation_required, reason). If True, the
+            operation needs human confirmation before proceeding.
+        """
+        if operation.lower() in self.required_confirmations:
+            return (
+                True,
+                f"Operation '{operation}' requires human confirmation",
+            )
+        return (False, None)
+
+    def increment_step(self) -> None:
+        """Increment the step counter by one."""
+        self.step_count += 1
+
+    def record_cost(self, cost: float) -> None:
+        """Record a tool invocation cost.
+
+        Args:
+            cost: The cost to add to the budget spent.
+        """
+        if cost < 0.0:
+            raise ValueError(f"cost must be >= 0.0, got {cost}")
+        self.budget_spent += cost
+
+    def mark_started(self) -> None:
+        """Record the plan execution start time for wall-clock tracking."""
+        self.start_time = datetime.now(UTC).isoformat()
+
+    model_config = ConfigDict(
+        str_strip_whitespace=True,
+        validate_assignment=True,
+    )
+
+
+# Re-export threading lock for service-level thread safety
+GuardrailLock = threading.Lock

vulture_whitelist.py

@@ -413,3 +413,45 @@ is_local_mode  # noqa: B018, F821
 get_role_bindings  # noqa: B018, F821
 add_binding  # noqa: B018, F821
 remove_binding  # noqa: B018, F821
+
+# Autonomy guardrails — public API surface
+ActorLimits  # noqa: B018, F821
+AutonomyGuardrails  # noqa: B018, F821
+GuardrailAuditEntry  # noqa: B018, F821
+GuardrailAuditTrail  # noqa: B018, F821
+GuardrailEventType  # noqa: B018, F821
+GuardrailResult  # noqa: B018, F821
+GuardrailLock  # noqa: B018, F821
+AutonomyGuardrailService  # noqa: B018, F821
+autonomy_guardrail_service  # noqa: B018, F821
+check_step_limit  # noqa: B018, F821
+check_tool_budget  # noqa: B018, F821
+check_wall_clock  # noqa: B018, F821
+check_actor_tool_calls  # noqa: B018, F821
+check_confirmation_required  # noqa: B018, F821
+record_enforcement  # noqa: B018, F821
+get_audit_trail  # noqa: B018, F821
+get_audit_trail_as_metadata  # noqa: B018, F821
+load_from_metadata  # noqa: B018, F821
+remove_plan  # noqa: B018, F821
+configure_guardrails  # noqa: B018, F821
+increment_step  # noqa: B018, F821
+record_cost  # noqa: B018, F821
+mark_started  # noqa: B018, F821
+denied_count  # noqa: B018, F821
+allowed_count  # noqa: B018, F821
+add_entry  # noqa: B018, F821
+max_entries  # noqa: B018, F821
+max_wall_clock_seconds  # noqa: B018, F821
+actor_limits  # noqa: B018, F821
+max_tool_calls_per_invocation  # noqa: B018, F821
+max_retries_per_failure  # noqa: B018, F821
+start_time  # noqa: B018, F821
+_DEFAULT_MAX_AUDIT_ENTRIES  # noqa: B018, F821
+check_retries_per_failure  # noqa: B018, F821
+RETRY_BLOCKED  # noqa: B018, F821
+RETRY_ALLOWED  # noqa: B018, F821
+validate_start_time  # noqa: B018, F821
+_enforce_guardrails  # noqa: B018, F821
+_enforce_guardrails_per_step  # noqa: B018, F821
+guardrail_service  # noqa: B018, F821