The First
Harness Native
Agent Language

// Nexa — Harness Native Agent Language
@tool("Search codebase with regex")
fn grep(pattern: string, path: string = "."): string {
    return std.shell.exec("grep -rn #{pattern} #{path}");
}

agent BugFixer(issue: string) uses grep {
    system `You are an expert engineer. Fix: #{issue}`

    // Harness C: compiler verifies eviction strategy exists
    context_policy { max_tokens: 100000, on_overflow: summarize }

    // Harness E: compiler verifies exit condition exists
    autoloop max_steps: 50 exit_when: "issue is resolved" {
        // Harness L: AI-native error self-correction
        try_agent { step(); }
        catch_correction(e: ToolError) { reflect `Error: #{e}. Retry.` }
    }
}

# Python + LangGraph — manual glue code
from langgraph.graph import StateGraph, END
from langchain.chat_models import ChatOpenAI
from langchain.tools import tool
import subprocess, json

# Manual tool schema — must match OpenAI format exactly
@tool
def grep(pattern: str, path: str = ".") -> str:
    """Search codebase with regex"""
    r = subprocess.run(["grep","-rn",pattern,path],capture_output=True)
    return r.stdout.decode()

# Manual context management — no auto-eviction
def manage_context(messages, max_tokens=100000):
    total = sum(len(m["content"])//4 for m in messages)
    if total > max_tokens * 0.8:
        return messages[:5] + messages[-3:]  # crude truncation
    return messages

# Manual ReAct loop — no exit condition verification
def bug_fixer(issue: str, max_steps=50):
    model = ChatOpenAI(model="minimax-m2.5")
    messages = [{"role":"system","content":f"Fix: {issue}"}]
    for step in range(max_steps):
        response = model.invoke(messages, tools=[grep])
        messages.append({"role":"assistant","content":response.content})
        if response.tool_calls:
            for tc in response.tool_calls:
                try:
                    result = tc.execute()
                    messages.append({"role":"tool","content":result})
                except Exception as e:
                    messages.append({"role":"tool","content":f"Error: {e}"})
        messages = manage_context(messages)
        if "resolved" in response.content.lower(): break
    return messages[-1]["content"]
# No Harness Validator. No sandbox. No trace. No HITL.

// Nexa — declarative DAG orchestration
agent Researcher { prompt: "Research the given topic." }
agent Analyst   { prompt: "Analyze findings quantitatively." }
agent Writer    { prompt: "Synthesize into a report." }
agent Reviewer  { prompt: "Review for accuracy." }

flow main(topic: string) {
    // Fork: parallel research + analysis → Merge: review → Pipeline: write
    result = topic |>> [Researcher, Analyst] &>> Reviewer >> Writer;
    print(result);
}

# Python + LangGraph — manual DAG construction
from langgraph.graph import StateGraph, END
from langchain.chat_models import ChatOpenAI
from typing import TypedDict
import concurrent.futures

model = ChatOpenAI(model="minimax-m2.5")

class State(TypedDict):
    topic: str; research: str; analysis: str; review: str; report: str

def research_node(state: State) -> State:
    state["research"] = model.invoke(f"Research: {state['topic']}")
    return state

def analysis_node(state: State) -> State:
    state["analysis"] = model.invoke(f"Analyze: {state['topic']}")
    return state

def review_node(state: State) -> State:
    combined = state["research"] + "\n" + state["analysis"]
    state["review"] = model.invoke(f"Review: {combined}")
    return state

def writer_node(state: State) -> State:
    state["report"] = model.invoke(f"Write report: {state['review']}")
    return state

# Manual graph construction — 15+ lines of wiring
graph = StateGraph(State)
graph.add_node("research", research_node)
graph.add_node("analysis", analysis_node)
graph.add_node("review", review_node)
graph.add_node("writer", writer_node)
graph.add_edge("research", "review")
graph.add_edge("analysis", "review")
graph.add_edge("review", "writer")
graph.set_entry_point("research")
graph.add_edge("writer", END)
app = graph.compile()
result = app.invoke({"topic": "Quantum Computing"})
# No compile-time topology verification