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

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
  };
}

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()

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
  };
}

# 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
        ]

# 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()

# 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

# 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())