DEV Community

Tamiz Uddin
Tamiz Uddin

Posted on Originally published at tamiz.pro

The Agency Stack in 2026: Lessons from Trueforge, OneCLI, and Lightdash on Building Production-Ready AI Agents

Originally published on tamiz.pro.

The era of the simple chatbot wrapper is over. By 2026, the distinction between "AI-powered feature" and "production-grade agent" has solidified into a rigorous architectural discipline. We are no longer just prompting; we are engineering sovereign systems that must handle state, observability, error recovery, and deterministic outcomes at scale.

This deep-dive examines the emerging Agency Stack—the layered architecture required to move agents from prototype to production—by analyzing patterns from three distinct verticals: Trueforge (developer tooling and code generation), OneCLI (local-first command-line agents), and Lightdash (structured data and analytics). These companies represent the three critical pressures of modern agent design: execution safety, user interaction models, and data integrity.

1. The Agency Stack: A Layered Architecture

Before dissecting the case studies, we must define the stack. In 2026, a production-ready agency stack typically consists of five layers:

  1. Interface Layer: How the user or system interacts (CLI, SDK, API).
  2. Orchestration Layer: The agent loop, state management, and tool routing.
  3. Execution Layer: Sandboxed environments for code/shell execution.
  4. Data & Memory Layer: Vector stores, SQL connections, and context windows.
  5. Observability & Evaluation Layer: Tracing, logging, and automated testing.

The failures of 2024 and 2025—hallucinated code, infinite loops, and unmonitored token spend—were largely due to missing or weak execution and observability layers. The lessons from Trueforge, OneCLI, and Lightdash address these gaps directly.

2. Trueforge: Execution Safety in Developer Workflows

Trueforge operates in the high-stakes environment of software development. When an agent generates code, executes tests, or modifies repositories, the cost of failure is not just a wrong answer—it’s broken builds and security vulnerabilities.

The Problem: Non-Deterministic Execution

LLMs are probabilistic. Developer tools require determinism. Trueforge’s approach demonstrates that agents must treat the LLM as a suggestion engine, not an authority.

Key Architectural Lessons:

  • Plan-Execute-Verify Loop: Trueforge implements a strict three-stage pipeline. The agent first generates a plan, then executes it in a sandboxed environment, and finally runs verification scripts (linters, type checkers, unit tests). If verification fails, the agent enters a self-correction loop with access to the error output.
  • Minimal Permissive Tools: Instead of giving agents broad filesystem access, Trueforge uses fine-grained tools (e.g., run_command, read_file, write_file) with explicit allowlists. This reduces the blast radius of any misdeed.
  • Stateful Session Management: Each development session maintains a full context of changes. Agents can rollback previous actions, a critical feature for CI/CD integration where one bad commit can disrupt pipelines.

Code Pattern: The Verification Wrapper

Below is a conceptual pattern used by Trueforge-like systems to enforce correctness after agent execution:

interface AgentAction {
  tool: string;
  args: Record<string, any>;
  planId: string;
}

async function executeWithVerification(
  action: AgentAction,
  verifier: Verifier,
  sandbox: SandboxEnvironment
): Promise<Result> {
  // 1. Execute in isolated sandbox
  const execution = await sandbox.run(action);

  // 2. Capture stdout, stderr, and exit code
  const { stdout, stderr, exitCode } = execution;

  // 3. Run verification (lint, type-check, tests)
  const verification = await verifier.run(execution.artifacts);

  // 4. Return rich result for agent self-correction
  return {
    success: exitCode === 0 && verification.passed,
    stdout,
    stderr,
    errors: verification.errors,
    nextSteps: verification.suggestions // Feed back to LLM
  };
}
Enter fullscreen mode Exit fullscreen mode

This pattern ensures that observability is baked into execution, not added as an afterthought. The verification.suggestions field is crucial—it transforms static test failures into dynamic learning opportunities for the agent.

3. OneCLI: Local-First Agents and Human-in-the-Loop

OneCLI represents the shift toward local-first, terminal-based agents. These tools run on the user’s machine, interact with existing CLI workflows, and require a different trust model than cloud-based assistants.

The Problem: Latency, Privacy, and Context

Cloud-based agents introduce latency and privacy concerns. OneCLI’s architecture prioritizes local execution and seamless integration with existing shell workflows.

Key Architectural Lessons:

  • Stdout/Stderr Parsing as a First-Class Citizen: OneCLI treats command-line output as structured data. It parses stdout and stderr using regex and schema-driven parsers to extract actionable information, rather than relying solely on the LLM to interpret raw text.
  • Interactive Confirmation Points: For destructive operations (e.g., git push --force, rm -rf), OneCLI pauses execution for explicit user confirmation. This human-in-the-loop design is essential for trust in local agents.
  • Contextual Shell Awareness: The agent maintains awareness of the current working directory, environment variables, and recent command history. This reduces the need for repetitive prompting and improves relevance.

Architecture: The Streaming Parser

OneCLI uses a streaming parser that processes command output in real-time, allowing the agent to react to partial results. For example, if a long-running process outputs progress bars or error messages, the agent can intervene early.

class StreamingCLIParser:
    def __init__(self, command: str):
        self.process = subprocess.Popen(
            command, shell=True, 
            stdout=subprocess.PIPE, 
            stderr=subprocess.PIPE,
            text=True
        )
        self.buffer = []

    def stream(self, callback: Callable):
        """Stream output and trigger callbacks on structural events."""
        while True:
            line = self.process.stdout.readline()
            if not line:
                break
            self.buffer.append(line)

            # Trigger callbacks on specific patterns
            if "error" in line.lower():
                callback("error_detected", line)
            elif "progress" in line.lower():
                progress = self.extract_progress(line)
                callback("progress_update", progress)

        return self.process.wait()
Enter fullscreen mode Exit fullscreen mode

This streaming approach allows for responsive agents that can adjust their behavior based on real-time feedback, a significant advantage over batch-processing architectures.

4. Lightdash: Data Integrity and Structured Outputs

Lightdash, an open-source BI platform, faces a unique challenge: generating SQL and analytics queries that are not only correct but also safe and performant. Hallucinated SQL can lead to incorrect business decisions or database loads.

The Problem: Schema Consistency and Query Performance

In analytics, correctness is non-negotiable. Lightdash’s agent stack emphasizes schema-aware generation and query validation.

Key Architectural Lessons:

  • Schema-First Generation: Instead of free-form SQL generation, Lightdash uses a schema-constrained approach. The agent first retrieves the relevant table schemas and column definitions, then generates SQL within those constraints. This reduces syntax errors and hallucinated columns.
  • Query Plan Analysis: Before executing generated queries, Lightdash analyzes the query execution plan to estimate cost and identify potential performance issues (e.g., full table scans). Agents are trained to prefer indexes and partition pruning.
  • Deterministic Output Formats: Analytics agents output structured data (JSON, CSV) with strict typing. This ensures downstream systems can rely on consistent data shapes, critical for dashboards and reports.

Technical Implementation: Schema-Guided SQL Generation

interface SchemaContext {
  tables: TableDefinition[];
  relationships: Relationship[];
  constraints: Constraint[];
}

function generateSQL(query: string, schema: SchemaContext): ValidatedSQL {
  // 1. Retrieve relevant schema portions based on query intent
  const relevantSchema = schema.extractRelevantTables(query);

  // 2. Generate SQL with schema constraints
  const sql = llm.generate(relevantSchema, query);

  // 3. Parse and validate against schema
  const parsed = parseSQL(sql);
  const validation = validateAgainstSchema(parsed, relevantSchema);

  if (!validation.valid) {
    throw new ValidationError(validation.errors);
  }

  // 4. Estimate cost and suggest optimizations
  const plan = analyzeQueryPlan(sql);
  return {
    sql,
    estimatedCost: plan.cost,
    suggestions: plan.optimizations
  };
}
Enter fullscreen mode Exit fullscreen mode

This schema-first, validate-then-execute pattern is essential for any agent interacting with relational data. It transforms the LLM from a guesser into a guided generator.

5. Cross-Cutting Patterns: The 2026 Agency Stack Blueprint

Analyzing Trueforge, OneCLI, and Lightdash reveals several common patterns that define the 2026 agency stack:

5.1. Observability as a Foundation, Not an Afterthought

All three companies treat tracing and logging as core infrastructure. Key components include:

  • Span-Level Tracing: Each agent action (tool call, LLM call, decision) is a traceable span.
  • Structured Logging: Logs are machine-readable (JSON) with consistent fields (timestamp, level, trace_id, agent_id).
  • Cost Tracking: Real-time monitoring of token usage and inference costs per session.

5.2. Evaluation-Driven Development

Production agents are continuously evaluated against benchmark datasets. Key practices:

  • Golden Test Sets: Curated inputs with expected outputs for regression testing.
  • A/B Testing Frameworks: Comparing agent versions in production with careful metrics tracking.
  • Automated Feedback Loops: User corrections and failures are fed back into training/finetuning pipelines.

5.3. State Management and Recovery

Agents must handle interruptions and failures gracefully:

  • Checkpointing: Saving agent state at key decision points.
  • Idempotent Operations: Designing tools so that retries don’t cause side effects.
  • Session Persistence: Resuming conversations from where they left off, with full context restored.

6. Building Your Own Agency Stack: A Practical Guide

If you’re building production-ready agents in 2026, consider this phased approach:

Phase 1: Define the Interface and Execution Context

  • Choose your interaction model (CLI, API, UI).
  • Design the execution environment (sandboxed, local, cloud).
  • Implement basic tool definitions with strict input/output schemas.

Phase 2: Implement the Orchestration Loop

  • Build the agent loop with support for multi-step reasoning.
  • Add error handling and retry logic with exponential backoff.
  • Integrate observability (tracing, logging) from day one.

Phase 3: Add Validation and Verification

  • Implement schema validation for all LLM outputs.
  • Add execution verification (tests, linting, query plans).
  • Create human-in-the-loop checkpoints for high-risk actions.

Phase 4: Scale with Evaluation and Monitoring

  • Build evaluation suites for your specific domain.
  • Set up cost and performance dashboards.
  • Implement feedback collection and continuous improvement pipelines.

7. The Future: Autonomous Agents and Self-Improvement

The next frontier is self-improving agents—systems that can learn from their mistakes and optimize their own configurations. Early research in 2026 shows promising results in:

  • Automatic Prompt Optimization: Agents that refine their own system prompts based on success/failure rates.
  • Tool Discovery: Agents that can suggest and integrate new tools based on task requirements.
  • Memory Synthesis: Agents that compress and summarize past interactions into reusable knowledge.

However, these advances come with increased complexity and risk. The lessons from Trueforge, OneCLI, and Lightdash remind us that reliability and safety must precede autonomy.

Frequently Asked Questions

Q1: How do I choose between local-first and cloud-based agent architectures?

A: Local-first (like OneCLI) is ideal for privacy-sensitive tasks, low-latency requirements, and integration with existing local workflows. Cloud-based architectures offer scalability, easier collaboration, and access to larger model ecosystems. Many production systems use a hybrid approach, with local execution for sensitive operations and cloud compute for heavy lifting.

Q2: What are the key metrics for evaluating production agent performance?

A: Beyond accuracy, track:

  • Task Success Rate: Percentage of tasks completed correctly without human intervention.
  • Latency: Time from user input to final output.
  • Cost Efficiency: Token usage and inference costs per task.
  • Safety Incidents: Number of violations, errors, or unauthorized actions.
  • User Satisfaction: Direct feedback and correction rates.

Q3: How can I implement effective human-in-the-loop workflows?

A: Use conditional checkpoints for high-risk actions (e.g., destructive commands, financial transactions). Provide clear explanations of the agent’s reasoning and options. Allow easy override and correction. Log all human interventions for analysis and improvement. Systems like OneCLI demonstrate that transparency and control build trust.


For more insights on AI engineering and production systems, visit Tamiz's Insights.


Deep Dive: Building a Production-Ready Agent Orchestrator

With the lessons from these three systems in mind, let's walk through constructing an agent orchestrator that balances autonomy with human oversight — the core pattern separating production-grade agents from experimental ones.

Architecture: The Three-Layer Agent Stack

Every robust agent system rests on three distinct layers:

  1. Execution Layer — The runtime where agents perform actions (tool calls, code execution, API requests)
  2. Coordination Layer — The brain that decides what to do next, manages context windows, and handles planning
  3. Observability Layer — The transparent lens through which humans can audit, intervene, and improve

Let's implement each layer.

Layer 1: The Execution Engine

The execution layer must be deterministic, sandboxed, and reversible. Trueforge taught us that agents executing arbitrary code without containment are a liability, not a feature.

# execution_layer.py
import asyncio
import json
import uuid
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any, Callable, Optional


class ActionStatus(str, Enum):
    PENDING = "pending"
    RUNNING = "running"
    SUCCESS = "success"
    FAILED = "failed"
    HUMAN_REVIEW = "human_review"
    REVERSED = "reversed"


@dataclass
class ActionLog:
    action_id: str
    action_type: str
    payload: dict[str, Any]
    status: ActionStatus
    result: Optional[dict[str, Any]] = None
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    human_interventions: list[dict] = field(default_factory=list)
    trace_context: dict[str, Any] = field(default_factory=dict)

    def to_execution_record(self) -> str:
        return json.dumps(
            {
                "id": self.action_id,
                "type": self.action_type,
                "status": self.status.value,
                "started": self.started_at.isoformat() if self.started_at else None,
                "completed": self.completed_at.isoformat() if self.completed_at else None,
                "result_summary": (
                    self.result.get("summary") if self.result else None
                ),
                "trace_id": self.trace_context.get("trace_id", "unknown"),
            },
            indent=2,
        )


class SandboxedExecutor:
    """
    Executes agent actions within controlled boundaries.
    Every action is logged, time-boxed, and interruptible.
    Inspired by OneCLI's explicit action-permission model.
    """

    def __init__(
        self,
        action_handlers: dict[str, Callable],
        timeout_seconds: int = 30,
        max_retries: int = 2,
    ):
        self._handlers = action_handlers
        self._timeout = timeout_seconds
        self._max_retries = max_retries
        self._action_logs: dict[str, ActionLog] = {}

    async def execute(
        self, action_type: str, payload: dict[str, Any]
    ) -> ActionLog:
        action_id = str(uuid.uuid4())[:8]
        log = ActionLog(
            action_id=action_id,
            action_type=action_type,
            payload=payload,
            status=ActionStatus.PENDING,
        )
        self._action_logs[action_id] = log

        handler = self._handlers.get(action_type)
        if handler is None:
            log.status = ActionStatus.FAILED
            log.result = {"error": f"No handler for action type: {action_type}"}
            return log

        for attempt in range(self._max_retries + 1):
            log.status = ActionStatus.RUNNING
            log.started_at = datetime.utcnow()
            try:
                result = await asyncio.wait_for(
                    handler(payload), timeout=self._timeout
                )
                log.status = ActionStatus.SUCCESS
                log.result = result
                log.completed_at = datetime.utcnow()
                return log
            except asyncio.TimeoutError:
                log.result = {"error": f"Action timed out after {self._timeout}s"}
                if attempt < self._max_retries:
                    continue
                log.status = ActionStatus.FAILED
                log.completed_at = datetime.utcnow()
                return log
            except Exception as e:
                log.result = {"error": str(e)}
                log.status = ActionStatus.FAILED
                log.completed_at = datetime.utcnow()
                return log

        return log

    def get_trace(self, action_id: str) -> ActionLog:
        return self._action_logs.get(action_id)

    def all_human_actions(self) -> list[ActionLog]:
        return [
            log
            for log in self._action_logs.values()
            if log.status == ActionStatus.HUMAN_REVIEW
        ]
Enter fullscreen mode Exit fullscreen mode

Layer 2: The Coordinator

The coordinator manages the agent's reasoning loop. Drawing from Lightdash's approach, it maintains a structured context window and surfaces uncertainty — not pretending to know when it doesn't.

# coordinator_layer.py
import json
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Optional


@dataclass
class Message:
    role: str  # "user" | "assistant" | "tool" | "system"
    content: str
    metadata: dict[str, Any] = field(default_factory=dict)
    timestamp: datetime = field(default_factory=datetime.utcnow)
    confidence: Optional[float] = None  # 0.0–1.0, surfaced for transparency

    def to_dict(self) -> dict:
        return {
            "role": self.role,
            "content": self.content,
            "confidence": self.confidence,
            "timestamp": self.timestamp.isoformat(),
            "metadata": self.metadata,
        }


@dataclass
class PlanningState:
    goal: str
    subtasks: list[dict[str, Any]] = field(default_factory=list)
    completed: list[str] = field(default_factory=list)
    blocked: list[str] = field(default_factory=list)
    need_human_input: bool = False
    human_questions: list[str] = field(default_factory=list)
    current_step: Optional[str] = None
    uncertainty_flags: list[str] = field(default_factory=list)

    def summary(self) -> str:
        return f"""
Planning State:
  Goal: {self.goal}
  Subtasks: {len(self.subtasks)}
  Completed: {len(self.completed)}
  Blocked: {len(self.blocked)}
  Needs Human Input: {self.need_human_input}
  Uncertainty Flags: {', '.join(self.uncertainty_flags) or 'None'}
""".strip()


class AgentCoordinator:
    """
    Manages the reasoning loop: observe → plan → execute → reflect.
    Implements Lightdash-style uncertainty surfacing and OneCLI-style
    explicit permission gates on high-stakes actions.
    """

    def __init__(
        self,
        llm_client,
        executor: SandboxedExecutor,
        max_steps: int = 20,
        human_gate_threshold: float = 0.7,
    ):
        self._llm = llm_client
        self._executor = executor
        self._max_steps = max_steps
        self._human_gate = human_gate_threshold
        self._messages: list[Message] = []
        self._plan: PlanningState = PlanningState(goal="")
        self._step_count = 0

    async def run(self, goal: str, initial_context: dict[str, Any] = None) -> dict:
        self._plan = PlanningState(goal=goal)
        self._messages = [
            Message(
                role="system",
                content=self._build_system_prompt(initial_context or {}),
            )
        ]

        while self._step_count < self._max_steps:
            self._step_count += 1
            self._messages.append(
                Message(
                    role="user",
                    content=f"[Step {self._step_count}] Continue working toward: {goal}",
                )
            )

            response = await self._llm.generate(
                messages=[m.to_dict() for m in self._messages]
            )

            # Extract tool calls from LLM response
            tool_calls = response.get("tool_calls", [])
            uncertainty = response.get("uncertainty_flags", [])

            if uncertainty:
                self._plan.uncertainty_flags.extend(uncertainty)

            if not tool_calls:
                # No more actions — we're done
                break

            for call in tool_calls:
                action_type = call.get("type")
                payload = call.get("parameters", {})
                confidence = call.get("confidence", 1.0)

                # OneCLI-inspired human gate
                if confidence < self._human_gate:
                    self._plan.need_human_input = True
                    self._plan.human_questions.append(
                        f"Agent proposes action '{action_type}' with low confidence "
                        f"({confidence:.2f}). Approve? (yes/no/modify)"
                    )
                    self._messages.append(
                        Message(
                            role="assistant",
                            content=f"Seeking human approval for: {action_type} "
                                    f"(confidence: {confidence:.2f})",
                            confidence=confidence,
                        )
                    )
                    continue

                result_log = await self._executor.execute(action_type, payload)
                self._messages.append(
                    Message(
                        role="assistant",
                        content=f"Executed {action_type}: {result_log.status.value}",
                    )
                )
                self._messages.append(
                    Message(
                        role="tool",
                        content=json.dumps(result_log.result or {}),
                        metadata={"action_id": result_log.action_id},
                    )
                )
                self._plan.completed.append(f"{action_type}{result_log.status.value}")

        return {
            "plan": self._plan.summary(),
            "final_context": self._messages[-5:],
            "steps_taken": self._step_count,
            "human_interventions_needed": self._plan.human_questions,
            "uncertainty_flags": self._plan.uncertainty_flags,
        }

    def _build_system_prompt(self, context: dict) -> str:
        return f"""You are a production-grade AI agent. Your design principles:

1. ALWAYS surface uncertainty — never pretend confidence you don't have.
2. For high-stakes actions (data modification, deletions, external API calls),
   explicitly flag when human review is needed.
3. Keep actions atomic and reversible where possible.
4. Log every decision and its rationale for auditability.

Current context: {json.dumps(context, indent=2)}
""".strip()
Enter fullscreen mode Exit fullscreen mode

Layer 3: The Observability Bridge

This is where Trueforge and OneCLI converge most powerfully. Observability isn't an afterthought — it's the mechanism that makes human intervention practical and trust durable.

# observability_layer.py
import json
import os
from datetime import datetime
from pathlib import Path
from typing import Any, Optional


class AuditTrail:
    """
    Immutable, append-only log of every agent decision.
    This is the foundation of agent transparency.
    Inspired by Trueforge's immutable audit log pattern.
    """

    def __init__(self, log_dir: str = "./agent_audits"):
        self._log_dir = Path(log_dir)
        self._log_dir.mkdir(parents=True, exist_ok=True)

    def record(self, session_id: str, event_type: str, data: dict):
        timestamp = datetime.utcnow().isoformat() + "Z"
        entry = {
            "ts": timestamp,
            "session": session_id,
            "event": event_type,
            "data": data,
        }
        log_file = self._log_dir / f"{session_id}.jsonl"
        with open(log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def query(
        self, session_id: str, event_type: Optional[str] = None
    ) -> list[dict]:
        log_file = self._log_dir / f"{session_id}.jsonl"
        if not log_file.exists():
            return []
        entries = []
        with open(log_file) as f:
            for line in f:
                entry = json.loads(line)
                if event_type and entry["event"] != event_type:
                    continue
                entries.append(entry)
        return entries


class HumanInterventionPortal:
    """
    A lightweight web interface for reviewing agent actions in real-time.
    OneCLI demonstrated that making interventions frictionless is what
    enables humans to stay in the loop at scale.
    """

    def __init__(self, audit: AuditTrail, port: int = 8080):
        self._audit = audit
        self._port = port
        self._sessions: dict[str, list[dict]] = {}

    def load_session(self, session_id: str) -> dict:
        events = self._audit.query(session_id)
        return {
            "session_id": session_id,
            "total_events": len(events),
            "events": events,
            "human_actions": [e for e in events if e["event"] == "human_intervention"],
            "agent_actions": [e for e in events if e["event"] in ("tool_call", "plan_step")],
        }

    def approve_action(self, session_id: str, action_id: str) -> dict:
        intervention = {
            "session": session_id,
            "event": "human_intervention",
            "data": {"action_id": action_id, "decision": "approved"},
        }
        self._audit.record(session_id, "human_intervention", intervention["data"])
        return {"status": "approved", "action_id": action_id}

    def modify_action(self, session_id: str, action_id: str, new_params: dict) -> dict:
        intervention = {
            "session": session_id,
            "event": "human_intervention",
            "data": {
                "action_id": action_id,
                "decision": "modified",
                "new_parameters": new_params,
            },
        }
        self._audit.record(session_id, "human_intervention", intervention["data"])
        return {"status": "modified", "action_id": action_id}


def build_session_dashboard(sessions: list[str]) -> dict:
    """
    Aggregated view across all agent sessions — what Lightdash
    would show a data analyst reviewing their workflow.
    """
    audit = AuditTrail()
    overview = {}
    for sid in sessions:
        events = audit.query(sid)
        overview[sid] = {
            "event_count": len(events),
            "human_interventions": len(
                [e for e in events if e["event"] == "human_intervention"]
            ),
            "agent_decisions": len(
                [e for e in events if e["event"] in ("tool_call", "plan_step")]
            ),
            "duration_estimate": (
                f"{events[-1]['ts'][:19]} - {events[0]['ts'][:19]}"
                if len(events) > 1
                else "single-event"
            ),
        }
    return overview
Enter fullscreen mode Exit fullscreen mode

Putting It All Together: A Complete Agent Session

Here's how the three layers compose into a single, production-ready agent loop:

# main_agent.py
import asyncio
import json
from typing import Any


async def sample_tool_handler(payload: dict) -> dict:
    """Example tool: read a configuration file."""
    path = payload.get("path", "")
    # Simulated file read
    return {"content": f"Contents of {path}", "lines": 42}


async def main():
    executor = SandboxedExecutor(
        action_handlers={"read_config": sample_tool_handler},
        timeout_seconds=15,
        max_retries=1,
    )

    # Mock LLM client for demonstration
    class MockLLM:
        async def generate(self, messages):
            return {
                "tool_calls": [
                    {
                        "type": "read_config",
                        "parameters": {"path": "/etc/app/config.yaml"},
                        "confidence": 0.92,
                    }
                ],
                "uncertainty_flags": [],
            }

    coordinator = AgentCoordinator(
        llm_client=MockLLM(),
        executor=executor,
        max_steps=10,
        human_gate_threshold=0.7,
    )

    audit = AuditTrail()
    session_id = "sess-2026-001"
    audit.record(session_id, "session_start", {"goal": "Load application config"})

    result = await coordinator.run(
        goal="Load and validate application configuration",
        initial_context={"app": "production-api", "env": "staging"},
    )

    audit.record(session_id, "session_complete", result)

    print(f"Session: {session_id}")
    print(f"Steps taken: {result['steps_taken']}")
    print(f"Plan summary:\n{result['plan']}")
    if result["human_interventions_needed"]:
        print(f"\n⚠ Human interventions required:")
        for q in result["human_interventions_needed"]:
            print(f"{q}")
    if result["uncertainty_flags"]:
        print(f"\n📊 Uncertainty flags: {result['uncertainty_flags']}")

    portal = HumanInterventionPortal(audit)
    dashboard = portal.load_session(session_id)
    print(f"\n📋 Session dashboard — {dashboard['total_events']} total events")


if __name__ == "__main__":
    asyncio.run(main())
Enter fullscreen mode Exit fullscreen mode

Conclusion: The Pattern That Unites Trueforge, OneCLI, and Lightdash

Across all three systems, one architectural pattern emerges consistently: trust is earned through visibility, not asserted through autonomy.

Trueforge proves that immutable audit trails aren't overhead — they're the product. When every decision is recorded and reversible, humans can delegate confidently because the consequences of error are bounded and traceable.

OneCLI demonstrates that permission gates shouldn't be barriers but conversational interfaces. The best human-in-the-loop designs don't interrupt — they invite collaboration, making the agent's reasoning legible enough that a human can say "yes," "no," or "try this instead" in seconds.

Lightdash reminds us that uncertainty surfacing is a feature, not a bug. Agents that admit when they're unsure earn higher long-term trust than agents that confidently produce wrong answers. A 0.6-confidence flag with a clear explanation is worth more than a 0.95-confidence hallucination every time.

The agency stack of 2026 isn't about building smarter models — it's about building more honest systems. The code above gives you the scaffolding. The culture of transparency around it is what makes it production-ready.


For more insights on AI engineering and production systems, visit Tamiz's Insights.

Top comments (0)