---
title: AI agent steering and control patterns
description: Place deterministic Memrail control points around model reasoning, tool calls, human escalation, and material outcomes without replacing the agent framework.
eyebrow: AGENT ARCHITECTURE
keywords: AI agent steering, AI agent control, agent tool call policy, deterministic guardrails
last_updated: 2026-09-10
---

# AI agent steering and control patterns

Memrail works beside an agent framework. The agent still observes, reasons, and proposes; Memrail gives the surrounding application a deterministic way to steer context and control execution at named seams.

## The authority boundary

The strongest integration separates three responsibilities:

```text
MODEL                         MEMRAIL                       APPLICATION
interpret input              evaluate explicit policy     validate arguments
classify into fixed tags  →  select policy outcome      → authorize and execute tool
propose an action             trace the decision            emit result event
```

Model inference is useful evidence. It becomes unsafe authority when an unbounded output, hidden confidence threshold, or prompt instruction is allowed to mutate external state directly.

Use fixed taxonomies for model-derived tags:

```python
from memrail.atoms import tag

REFUND_INTENTS = {"duplicate_charge", "service_failure", "fraud", "other", "unknown"}

# Validate actual runtime output; a type annotation alone does not do this.
if classified_intent not in REFUND_INTENTS:
    raise ValueError("Invalid refund intent")
context.append(tag("refund_intent", classified_intent, source="ml"))
```

Combine classifications with trusted state and event history, and give `unknown` an explicit safe outcome. The SDK preserves source metadata. Label model-derived values accurately and prevent model output from impersonating trusted application facts.

Each example invokes a named `decision_point`. Bind its policies to that exact name and follow the [integration invariants](/concepts/#integration-invariants).

`your_app.execution_contract`, `your_app.review_queue`, `model`, and executor helpers below are application-owned integration points, not SDK functions. The contract receives the full selection, including policy and lifecycle metadata, and must deny unauthorized modes/states, verify the reviewed action, and enforce caller permissions. An EMU-key allowlist alone is insufficient for consequential policies. The [Python](/python/#handle-selected-actions) and [TypeScript](/typescript/#dispatch-selected-actions) references also describe SDK-managed dispatch.

## Pattern 1: steer before reasoning

Invoke Memrail before the model generates its next response. Use `context_directive` actions to add applicable protocol, safety, or customer context.

```python
from memrail.models import InvokeOptions

evaluation_only = True  # Change only through the application's rollout control.
guidance = await client.decide(
    decision_point="agent.before_response",
    context=[
        state("customer.tier", customer.tier),
        state("conversation.turn_count", turn_count),
        tag("intent", intent, source="ml"),
    ],
    options=InvokeOptions(dry_run=evaluation_only),
)

directives = [
    item.action["directive"]
    for item in guidance.selected
    if not evaluation_only
    and your_app.execution_contract.allows(item, principal=current_user)
    and item.action and item.action.get("type") == "context_directive"
]

response = await model.generate(
    system="\n".join([base_instructions, *directives]),
    messages=messages,
)
```

Use this for response protocols and context shaping. Do not treat a directive as a hard execution barrier; the model can still produce unexpected text. Put consequential actions behind a separate control point.

## Pattern 2: control before a tool call

When the model proposes a tool, validate its proposed arguments and translate them into typed state and tags, then ask Memrail for a policy outcome. Verify monetary amounts and entitlement against trusted domain records, not merely the model's extraction.

```python
proposal = await model.propose_tool(messages)
amount = validate_refund_amount(proposal.arguments["amount"], customer)

decision = await client.decide(
    decision_point="agent.before_tool_call",
    context=[
        state("agent.tool_name", proposal.name),
        state("customer.id", customer.id),
        state("customer.tier", customer.tier),
        state("refund.amount", amount),
        state("refund.request_id", request_id),
        tag("refund_intent", classified_intent, source="ml"),
    ],
    options=InvokeOptions(dry_run=evaluation_only),
)

for item in decision.selected:
    if evaluation_only:
        continue  # Dry run may contain ordinary actionable selections.
    if not your_app.execution_contract.allows(item, principal=current_user):
        continue
    if item.action and item.action.get("type") == "decision_prompt":
        await your_app.review_queue.create(item.action)
    elif item.action and item.action.get("type") == "tool_call":
        await your_app.execute_authorized_tool(item.action["tool"])
```

The executor uses the Memrail-selected tool specification, not the original free-form model proposal. It still performs ordinary schema validation and authorization close to the underlying service.

Design the default explicitly. For a side-effecting proposal, an empty selection, timeout, malformed context, or unavailable control service should normally stop execution or route to a human. It should not fall through to “let the model decide.”

## Pattern 3: deterministic tool selection

When policy can choose the tool directly, bypass model tool selection for the final step:

```json
{
  "emu_key": "billing.large_refund_review",
  "decision_point": "agent.before_tool_call",
  "trigger": "state.refund.request_id EXISTS AND (state.refund.amount > 500 OR tag.refund_intent == 'fraud')",
  "action": {
    "type": "tool_call",
    "intent": "CREATE_REFUND_REVIEW",
    "tool": {
      "tool_id": "refund_review_queue",
      "version": "1.0.0",
      "args": {
        "request_id": "{{refund.request_id}}"
      }
    }
  },
  "policy": {
    "mode": "auto",
    "priority": 9,
    "cooldown": {
      "seconds": 3600,
      "gate": "ack"
    },
    "idempotency": {
      "enabled": true,
      "boundary": "workspace",
      "roles": "auto",
      "scope": [
        "refund.request_id"
      ]
    }
  },
  "expected_utility": 0.95,
  "confidence": 0.95,
  "state": "shadow"
}
```

The trigger already guards the interpolated request ID with `EXISTS`. Idempotency `scope` contains literal state keys, not `{{...}}` templates. This one-hour cooldown is shared by the EMU across its workspace/project, not per refund request; omit or shorten it if independent requests must create review tasks immediately, and enforce request-level duplication in the queue. Priority 9 only breaks score ties; it does not guarantee this rule wins.

## Pattern 4: require a human

Use an advisory `decision_prompt` to request review while keeping the consequential operation blocked. After authenticating approval for that operation, pass `InvokeOptions(human_consent=True)` in Python or `options: { human_consent: true }` in TypeScript. Combined SDK helpers propagate consent to selection and execution. `require_human` applies to every action type and does not create an approval queue automatically. Never infer consent from model output.

Good human handoffs contain:

- the decision and bounded options;
- the verified facts that caused the escalation;
- the relevant trace or invocation identifier;
- an expiry and safe timeout outcome;
- an idempotency key so retries do not create duplicate work.

Human review is an execution state, not a log message. Do not continue the consequential branch while approval is pending.

## Pattern 5: record the outcome

The selection trace says what policy prescribed. Emit a material event after the executor reports what happened:

```python
from datetime import datetime, timezone
from memrail.atoms import event

if evaluation_only or not your_app.execution_contract.allows(item, principal=current_user):
    raise PermissionError("Execution is not authorized")

result = await your_app.execute_authorized_tool(item.action["tool"])

if result.success:
    await client.emit_event(
        event(
            "agent.completed.refund_review",
            ts=datetime.now(timezone.utc),
            attributes={
                "request_id": request_id,
                "activation_id": item.activation_id,
            },
        )
    )
    await client.ack(item.activation_id, "acknowledged", code="SUCCESS")
```

Event emission is not an activation ACK. An ACK-gated cooldown starts only after a successful `acknowledged` status. ACK uses a receipt saved before the decision response, independently of tracing. Reconcile delivery failures without rerunning the side effect. Persist outcome/event/ACK work durably (for example, in an outbox) so a failure in one does not lose the others.

Scoped event attributes allow future policy to ask about the current entity:

```dsl
state.refund.request_id EXISTS AND NOT event.agent.completed.refund_review
  WHERE request_id == '{{refund.request_id}}'
  IN 'P7D'
```

Without `request_id` in the event attributes, that guard cannot distinguish one refund request from another. Without the state ID, an unresolved placeholder may match no events and make the negation true; keep the presence guard. Missing producers and expired event history also invalidate an absence-based duplicate check.

## Place control points by consequence

| Seam | Typical action | Default on failure | Useful context |
|---|---|---|---|
| before retrieval | choose allowed corpus or route | return bounded public context | identity, purpose, data class |
| before response | add protocol directives | use base instructions | user tier, locale, intent |
| before tool call | allow, reroute, or require review | deny side effect | tool, amount, entity, risk tag |
| between workflow steps | authorize transition | hold current state | current state, prerequisites |
| after execution | acknowledge and emit outcome | retry or reconcile | activation, result, entity IDs |

Use multiple narrow points when consequences differ. A policy that controls read-only search should not share an ambiguous contract with one that sends money or deletes data.

## Control failures to test

- **Missing atom:** a positive predicate evaluates false, but `NOT` can invert it and `OR` can match another branch. Test explicit presence guards.
- **Ghost negative event:** a `NOT event...` guard is always true because no producer emits that event.
- **Unscoped event:** an event from another user satisfies the current user’s policy.
- **Unknown classifier output:** free-form or new labels bypass expected branches.
- **Disconnected action:** an EMU selects a tool or route the application cannot handle.
- **Project mismatch:** a schema fallback hides that the expected project-scoped executor is absent; verify actual connectivity, not only validation output.
- **Unsafe fallback:** Memrail is unavailable and execution falls back to model authority.
- **Evaluation leak:** dry-run/advisory selections reach a live dispatcher, or comparison code and legacy logic both execute a side effect.
- **Metadata loss:** a custom adapter drops policy or lifecycle information before authorization.
- **No outcome event:** selection is recorded, but completion and failure cannot be distinguished.

Build tests around these failure classes, not only the happy-path policy match.

## Framework integration rule

Keep the Memrail call in an adapter around the framework's proposal/execution seam. Do not bury it inside the model prompt or a generic callback where the application cannot enforce the result. The adapter builds typed context, invokes the control point, and dispatches only supported, explicitly authorized live actions.