Building Your First AI Control Framework: From Theory to Live Implementation
You've read the theory. You know *why* AI control frameworks matter.
Now the hard question: **how do you actually build one?**
Not the strategic version—the operational one. The one that works Monday morning when your AI agent needs to publish a document, send an email, or modify a database, and you need verification *before* it happens.
This guide walks you through building a control framework from scratch. No consulting deck. No aspirational architecture. Just the decisions, the code, and the mistakes you'll avoid.
---
The Control Framework: What You're Actually Building
A control framework is three layers:
- 1. **Pre-flight Gate**: Verify before action (not after).
- 2. **Approval Layer**: Human checkpoint with decision power.
- 3. **Execution + Evidence**: Do the thing, log everything.
Most teams implement this backwards. They log *after* the disaster. Here's a practical implementation approach.
---
Layer 1: Pre-flight Gate
Before your AI agent does anything external, it must verify:
- ✅ Does the output look valid? (length, format, content)
- ✅ Is this a duplicate action? (same task, same timeframe)
- ✅ Are all dependencies satisfied? (files exist, APIs reachable)
- ✅ Does the audience make sense? (sending to right person/system)
**Implementation:**
```python
def preflight_check(action: Action) -> PreflightResult:
"""Verify action is safe before submission."""
checks = {
"output_valid": validate_output(action.output),
"not_duplicate": not action_already_pending(action.hash),
"deps_satisfied": all_dependencies_ready(action.deps),
"audience_correct": audience_verified(action.recipient),
}
if not all(checks.values()):
failed = [k for k, v in checks.items() if not v]
return PreflightResult(
passed=False,
reason=f"Failed checks: {failed}",
action_blocked=True # Do NOT proceed
)
return PreflightResult(passed=True, reason="All checks passed")
```
**Real-world example:**
Your AI agent wants to publish a blog post. Preflight checks:
- ✅ Slug is valid URL format
- ✅ Post is >500 words (quality gate)
- ✅ Title is unique (no duplicate publish)
- ✅ Timestamp is on publish day (not random time)
- ❌ FAILS: Post references a previous post that doesn't exist yet
**Result:** Blocked. No publish attempt. Evidence logged. Human reviews.
---
Layer 2: Approval Layer
If preflight passes, the action goes to **human approval**. Not automagically. A person decides yes/no.
**Implementation:**
```python
class ApprovalQueue:
"""Queue actions pending human approval."""
def __init__(self, approval_file: Path):
self.queue_dir = approval_file
def enqueue(self, action: Action) -> ActionWithApprovalStatus:
"""Add action to queue, waiting for human decision."""
approval_status = {
"id": action.id,
"action": action.description,
"timestamp_utc": datetime.utcnow().isoformat(),
"approved_by": None, # Waiting for human
"approval_timestamp": None,
"status": "PENDING_APPROVAL"
}
Write to queue file
queue_file = self.queue_dir / f"{action.id}.json"
queue_file.write_text(json.dumps(approval_status, indent=2))
Notify human (Telegram, email, whatever)
notify_human(f"Action {action.id} waiting approval: {action.description}")
return ActionWithApprovalStatus(
action=action,
status="WAITING",
approval_file=queue_file
)
def is_approved(self, action_id: str) -> bool:
"""Check if human approved this action."""
queue_file = self.queue_dir / f"{action_id}.json"
if not queue_file.exists():
return False
data = json.loads(queue_file.read_text())
return data.get("approved_by") is not None
def get_approval_metadata(self, action_id: str) -> dict:
"""Get who approved, when, for audit trail."""
queue_file = self.queue_dir / f"{action_id}.json"
return json.loads(queue_file.read_text())
```
**Real-world flow:**
- 1. **09:00 UTC** — AI agent finishes post draft, passes preflight
- 2. **09:01 UTC** — Action queued, human notified
- 3. **09:15 UTC** — Human reviews draft (reads actual content, not summary)
- 4. **09:16 UTC** — Human approves (writes `approved_by: "Peter"` to queue file)
- 5. **09:17 UTC** — Executor checks approval, sees `approved_by = "Peter"`, proceeds
**Critical rule:** No approval file = no execution. Period.
---
Layer 3: Execution + Evidence
If approved, execute the action. Log everything.
**Implementation:**
```python
class ControlledExecutor:
"""Execute approved actions with full evidence trail."""
def __init__(self, approval_queue: ApprovalQueue, evidence_dir: Path):
self.queue = approval_queue
self.evidence_dir = evidence_dir
def execute_if_approved(self, action: Action) -> ExecutionResult:
"""Execute ONLY if approved. Log everything."""
Step 1: Verify approval exists
if not self.queue.is_approved(action.id):
result = ExecutionResult(
status="BLOCKED_NO_APPROVAL",
reason="Action not approved",
action_id=action.id
)
self._log_evidence(action, result)
return result
approval_meta = self.queue.get_approval_metadata(action.id)
Step 2: Execute the action
start_time = datetime.utcnow()
try:
execution_output = action.execute() # Do the actual thing
success = True
error = None
except Exception as e:
execution_output = None
success = False
error = str(e)
end_time = datetime.utcnow()
Step 3: Log evidence
evidence = {
"action_id": action.id,
"action_type": action.type,
"description": action.description,
Approval metadata
"approved_by": approval_meta["approved_by"],
"approval_timestamp": approval_meta["approval_timestamp"],
Execution metadata
"executed_at_utc": start_time.isoformat(),
"execution_duration_seconds": (end_time - start_time).total_seconds(),
"success": success,
"output": execution_output if success else None,
"error": error if not success else None,
}
Step 4: Write evidence
evidence_file = self.evidence_dir / f"execution-{action.id}-{start_time.timestamp()}.json"
evidence_file.write_text(json.dumps(evidence, indent=2))
Step 5: Return result
return ExecutionResult(
status="EXECUTED" if success else "FAILED",
action_id=action.id,
output=execution_output,
evidence_file=str(evidence_file)
)
def _log_evidence(self, action: Action, result: ExecutionResult):
"""Log failed executions too."""
evidence = {
"action_id": action.id,
"status": result.status,
"reason": result.reason,
"timestamp": datetime.utcnow().isoformat(),
}
evidence_file = self.evidence_dir / f"blocked-{action.id}.json"
evidence_file.write_text(json.dumps(evidence, indent=2))
```
**Real-world example:**
```python
Ghost publishing with controlled execution
action = PublishAction(
title="Building Your First AI Control Framework",
content="...",
slug="building-ai-control-framework-implementation"
)
Preflight
preflight = preflight_check(action)
if not preflight.passed:
log_and_alert(f"Publish blocked: {preflight.reason}")
exit(1)
Queue for approval
executor = ControlledExecutor(approval_queue, evidence_dir)
executor.execute_if_approved(action)
Result in evidence files:
- execution-action-123-1719653820.json (if success)
- blocked-action-123.json (if no approval)
```
---
The Kill Switch: Emergency Stop
Your framework needs an emergency stop button.
A file that says "nothing publishes until I say otherwise."
**Implementation:**
```python
PUBLISHING_DISABLED_FLAG = Path("/path/to/PUBLISHING_DISABLED.flag")
def is_publishing_enabled() -> bool:
"""Check if publishing is globally enabled."""
return not PUBLISHING_DISABLED_FLAG.exists()
def preflight_check(action: Action) -> PreflightResult:
"""Verify action is safe before submission."""
Check 1: Is publishing disabled?
if not is_publishing_enabled():
return PreflightResult(
passed=False,
reason="Publishing globally disabled",
action_blocked=True
)
... rest of checks ...
```
**Real-world usage:**
```bash
Everything running normally
$ python3 publish_agent.py # Works fine
Incident detected
$ touch /path/to/PUBLISHING_DISABLED.flag
Now:
$ python3 publish_agent.py # Blocked immediately
Output: FINAL_STATUS=PUBLISHING_BLOCKED_BY_GLOBAL_KILL_SWITCH
```
The flag is checked *before* any API calls, *before* any side effects. It's your circuit breaker.
---
Putting It Together: A Complete Workflow
Here's what a real publish cycle looks like:
```
09:00 UTC — Prepare Phase
├─ AI agent drafts blog post
├─ Preflight check runs
├─ ✅ Passes (format, length, no duplicates)
└─ Action queued, human notified
09:15 UTC — Review Phase
├─ Human reads draft (in review tool, not summary)
├─ Human approves: writes "approved_by: peter" to queue file
└─ Executor notified
09:17 UTC — Execute Phase
├─ Executor checks: is_approved? YES
├─ Executor checks: is_publishing_enabled? YES
├─ Ghost publish called
├─ HTTP 200 verified (actually live)
├─ Evidence logged with timestamp, approval meta, Ghost URL
├─ X publish triggered (only after Ghost HTTP 200)
└─ Newsletter sent (only if explicitly approved)
09:18 UTC — Audit Trail Complete
├─ All actions in evidence files
├─ Full chain: draft → approval → execution → public
└─ Disaster recovery possible: what happened, who approved, when
```
---
Common Mistakes to Avoid
- 1. **Logging after the fact** — You logged the failure. Good. But the damage is done. Preflight prevents damage.
- 2. **Approval as rubber-stamp** — If humans approve everything without reading, you haven't built a control framework, you've built a theater of approval. Make review easy. Ask for real decisions.
- 3. **Missing the dependencies** — Your AI publishes a post. It references another post that doesn't exist yet. Preflight checks catch this.
- 4. **No emergency stop** — You find a critical bug at 3 AM. You need everything to stop immediately. Build that button now, when there's no fire.
- 5. **Incomplete evidence** — You published something. Now you need to debug. Your evidence logs say "success" but not *what* succeeded, *who* approved, or *when*. Log generously.
---
Next Steps
- 1. **Pick a use case** — Start with one action type (Ghost publish, email send, database modify). Master it.
- 2. **Build preflight** — What must be true before this action is safe? Implement those checks.
- 3. **Add approval** — Queue the action, notify a human, wait for decision.
- 4. **Implement execution** — Execute only if approved. Log everything.
- 5. **Add the kill switch** — One file that stops everything. Check it first.
- 6. **Test it breaks** — Disable publishing. Run your workflow. Confirm it stops. Then enable and confirm it resumes.
You now have a control framework.
It won't prevent all mistakes. But it prevents the easy ones. It forces the hard ones into daylight where humans can see them.
That's the point.
---
Final Thought
AI agents are powerful. Control frameworks are the difference between powerful and reckless.
Build it right. Your customers will thank you.