The Death of Frameworks, The Rise of the Harness: Why Software Architecture is Moving to Upstream Agentic Runtimes

Agent = Model + Harness

"""
A minimal Agentic Runtime ("Harness").
 
The harness owns the event loop, the memory hierarchy, the sandbox, and the
loop-detection heuristics. The model is treated as a swappable component.
"""
 
from __future__ import annotations
 
import asyncio
import json
import logging
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Any, Awaitable, Callable, Protocol
 
logger = logging.getLogger("harness")
 
 
# ---------------------------------------------------------------------------
# Model interface — swappable, NOT a framework dependency
# ---------------------------------------------------------------------------
 
class ModelClient(Protocol):
    """Anything that can take a list of messages + tool schemas and return
    a structured action. The harness does not care which provider it is."""
 
    async def complete(
        self,
        messages: list[dict[str, Any]],
        tools: list[dict[str, Any]],
    ) -> dict[str, Any]: ...
 
 
# ---------------------------------------------------------------------------
# Hybrid memory: working / episodic / semantic
# ---------------------------------------------------------------------------
 
@dataclass
class HybridMemory:
    working: deque[dict[str, Any]] = field(default_factory=lambda: deque(maxlen=40))
    episodic: list[dict[str, Any]] = field(default_factory=list)
    semantic: dict[str, str] = field(default_factory=dict)
    max_working_tokens: int = 8000
 
    def append(self, message: dict[str, Any]) -> None:
        self.working.append(message)
        self.episodic.append({"ts": time.time(), **message})
 
    def compact(self, summarizer: Callable[[list[dict[str, Any]]], str]) -> None:
        """Compaction policy: when working memory is too large, summarize the
        oldest half into a single semantic note and drop those raw turns."""
        size = sum(len(json.dumps(m)) for m in self.working)
        if size < self.max_working_tokens * 4:  # rough char->token ratio
            return
        half = len(self.working) // 2
        old = [self.working.popleft() for _ in range(half)]
        summary = summarizer(old)
        key = f"summary:{int(time.time())}"
        self.semantic[key] = summary
        self.working.appendleft({"role": "system", "content": f"[memory] {summary}"})
        logger.info("compacted %d turns into semantic note %s", half, key)
 
    def render(self) -> list[dict[str, Any]]:
        sem = [{"role": "system", "content": f"[fact] {v}"}
               for v in list(self.semantic.values())[-5:]]
        return sem + list(self.working)
 
 
# ---------------------------------------------------------------------------
# Sandbox: every tool call is wrapped in a controlled environment
# ---------------------------------------------------------------------------
 
ToolFn = Callable[[dict[str, Any]], Awaitable[Any]]
 
 
@dataclass
class Sandbox:
    timeout_s: float = 20.0
    max_output_chars: int = 16_000
    deny_list: tuple[str, ...] = ("subprocess.Popen", "os.system", "eval", "exec")
 
    async def run(self, tool: ToolFn, args: dict[str, Any]) -> Any:
        # 1) static argument inspection
        flat = json.dumps(args, default=str).lower()
        for bad in self.deny_list:
            if bad.lower() in flat:
                raise PermissionError(f"sandbox: blocked unsafe argument '{bad}'")
 
        # 2) bounded execution
        try:
            result = await asyncio.wait_for(tool(args), timeout=self.timeout_s)
        except asyncio.TimeoutError as e:
            raise TimeoutError(f"sandbox: tool exceeded {self.timeout_s}s") from e
 
        # 3) output truncation — protects context window
        s = json.dumps(result, default=str)
        if len(s) > self.max_output_chars:
            s = s[: self.max_output_chars] + "\n…[truncated]"
            return json.loads(s) if s.startswith("{") else s
        return result
 
 
# ---------------------------------------------------------------------------
# The Harness itself — the event loop owns everything
# ---------------------------------------------------------------------------
 
@dataclass
class Harness:
    model: ModelClient
    memory: HybridMemory = field(default_factory=HybridMemory)
    sandbox: Sandbox = field(default_factory=Sandbox)
    tools: dict[str, ToolFn] = field(default_factory=dict)
    schemas: list[dict[str, Any]] = field(default_factory=list)
    max_steps: int = 25
    loop_window: int = 4  # detect identical actions within last N steps
 
    def register_tool(self, name: str, fn: ToolFn, schema: dict[str, Any]) -> None:
        self.tools[name] = fn
        self.schemas.append(schema)
 
    def _summarize(self, msgs: list[dict[str, Any]]) -> str:
        # Real harnesses call the model here; we keep it deterministic.
        return f"Earlier the agent performed {len(msgs)} steps."
 
    def _detect_loop(self, recent: list[str]) -> bool:
        tail = recent[-self.loop_window:]
        return len(tail) == self.loop_window and len(set(tail)) == 1
 
    async def run(self, goal: str) -> str:
        self.memory.append({"role": "user", "content": goal})
        action_log: list[str] = []
 
        for step in range(self.max_steps):
            self.memory.compact(self._summarize)
            decision = await self.model.complete(
                messages=self.memory.render(),
                tools=self.schemas,
            )
 
            # Terminal: model returned a final answer
            if decision.get("type") == "final":
                self.memory.append({"role": "assistant", "content": decision["content"]})
                return decision["content"]
 
            # Tool call path
            name = decision["name"]
            args = decision.get("args", {})
            signature = f"{name}:{json.dumps(args, sort_keys=True, default=str)}"
            action_log.append(signature)
 
            if self._detect_loop(action_log):
                logger.warning("loop detected on %s — forcing reflection", name)
                self.memory.append({
                    "role": "system",
                    "content": "You repeated the same action. Reflect and choose differently.",
                })
                continue
 
            if name not in self.tools:
                self.memory.append({"role": "tool", "name": name,
                                    "content": f"unknown tool '{name}'"})
                continue
 
            try:
                result = await self.sandbox.run(self.tools[name], args)
                self.memory.append({"role": "tool", "name": name,
                                    "content": json.dumps(result, default=str)})
            except Exception as e:
                logger.exception("tool %s failed", name)
                self.memory.append({"role": "tool", "name": name,
                                    "content": f"ERROR: {type(e).__name__}: {e}"})
 
        return "step budget exhausted"

from langgraph.graph import StateGraph, END
from typing import TypedDict
 
class RefundState(TypedDict):
    order_id: str
    eligibility: str | None
    refund_id: str | None
    notification_sent: bool
 
def check_eligibility(state: RefundState) -> RefundState:
    state["eligibility"] = orders_db.check(state["order_id"])  # custom client
    return state
 
def issue_refund(state: RefundState) -> RefundState:
    if state["eligibility"] != "approved":
        raise ValueError("not eligible")
    state["refund_id"] = payments.refund(state["order_id"])  # another client
    return state
 
def notify(state: RefundState) -> RefundState:
    email.send(state["order_id"], state["refund_id"])         # yet another client
    state["notification_sent"] = True
    return state
 
graph = StateGraph(RefundState)
graph.add_node("check",  check_eligibility)
graph.add_node("refund", issue_refund)
graph.add_node("notify", notify)
graph.add_edge("check",  "refund")
graph.add_edge("refund", "notify")
graph.add_edge("notify", END)
graph.set_entry_point("check")
app = graph.compile()

{
  "name": "refund_customer",
  "description": "Issue a refund for a customer order with eligibility check and notification.",
  "triggers": ["refund", "return", "chargeback", "reimburse"],
  "instructions": "1) Call internal-crm.get_order to fetch the order. 2) Verify eligibility per refund-policy. If not eligible, propose goodwill credit instead. 3) Call payments.issue_refund. 4) Call comms.send_email with template 'refund_confirmation'. Stop and ask the user before any refund > $500.",
  "required_mcp_tools": [
    "internal-crm.get_order",
    "payments.issue_refund",
    "comms.send_email"
  ],
  "approval_threshold_usd": 500,
  "success_count": 47,
  "failure_count": 1
}

"""
A persistent skill registry. Skills are reusable, abstract capabilities the
agent has learned. They are NOT prompts — they are structured artifacts the
harness retrieves on demand and injects only when relevant.
"""
 
from __future__ import annotations
 
import hashlib
import json
import time
from dataclasses import asdict, dataclass, field
from pathlib import Path
 
 
@dataclass
class Skill:
    name: str
    description: str           # used for retrieval — short, intent-focused
    triggers: list[str]        # keywords / task patterns that activate this skill
    instructions: str          # how to perform the task
    code: str | None = None    # optional executable companion (validated before use)
    success_count: int = 0
    failure_count: int = 0
    created_at: float = field(default_factory=time.time)
 
    @property
    def confidence(self) -> float:
        total = self.success_count + self.failure_count
        return self.success_count / total if total else 0.5
 
    def fingerprint(self) -> str:
        return hashlib.sha256(
            (self.name + self.instructions + (self.code or "")).encode()
        ).hexdigest()[:12]
 
 
class SkillRegistry:
    """File-backed registry: one JSON file per skill, retrieval by trigger match."""
 
    def __init__(self, root: str | Path = "./skills") -> None:
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)
 
    # --- persistence -------------------------------------------------------
 
    def save(self, skill: Skill) -> Path:
        path = self.root / f"{skill.name}.json"
        path.write_text(json.dumps(asdict(skill), indent=2), encoding="utf-8")
        return path
 
    def all(self) -> list[Skill]:
        skills: list[Skill] = []
        for p in self.root.glob("*.json"):
            data = json.loads(p.read_text(encoding="utf-8"))
            skills.append(Skill(**data))
        return skills
 
    # --- retrieval ---------------------------------------------------------
 
    def find(self, query: str, top_k: int = 3, min_confidence: float = 0.4) -> list[Skill]:
        q = query.lower()
        scored: list[tuple[float, Skill]] = []
        for s in self.all():
            if s.confidence < min_confidence:
                continue
            score = sum(1 for t in s.triggers if t.lower() in q)
            if score:
                scored.append((score + s.confidence, s))
        scored.sort(key=lambda x: x[0], reverse=True)
        return [s for _, s in scored[:top_k]]
 
    # --- learning loop -----------------------------------------------------
 
    def record_outcome(self, name: str, success: bool) -> None:
        path = self.root / f"{name}.json"
        if not path.exists():
            return
        data = json.loads(path.read_text(encoding="utf-8"))
        if success:
            data["success_count"] += 1
        else:
            data["failure_count"] += 1
        path.write_text(json.dumps(data, indent=2), encoding="utf-8")
 
    def promote(self, skill: Skill) -> Skill:
        """Called by the harness after a successful novel task: stores the
        distilled skill so the next agent run starts smarter."""
        existing = {s.name for s in self.all()}
        if skill.name in existing:
            self.record_outcome(skill.name, success=True)
            return skill
        self.save(skill)
        return skill

"""
A minimal MCP-style client. In production you would use an official MCP SDK,
but the contract is small enough to illustrate from scratch:
 
    GET  /mcp/tools         -> list of tool schemas
    POST /mcp/call          -> { name, args } -> { result | error }
 
The harness registers EVERY discovered tool through its sandbox, so MCP servers
get the same loop/timeout/output-cap protections as in-process tools.
"""
 
from __future__ import annotations
 
import asyncio
from dataclasses import dataclass
from typing import Any
 
import httpx
 
 
@dataclass
class MCPServer:
    name: str
    url: str
    auth_header: str | None = None  # e.g. "Bearer …" — usually delegated by harness
 
 
class MCPClient:
    def __init__(self, servers: list[MCPServer], timeout_s: float = 15.0) -> None:
        self.servers = servers
        self._client = httpx.AsyncClient(timeout=timeout_s)
 
    async def aclose(self) -> None:
        await self._client.aclose()
 
    def _headers(self, srv: MCPServer) -> dict[str, str]:
        h = {"Content-Type": "application/json"}
        if srv.auth_header:
            h["Authorization"] = srv.auth_header
        return h
 
    # --- discovery ---------------------------------------------------------
 
    async def discover(self) -> list[dict[str, Any]]:
        """Returns a flat list of tool schemas, namespaced by server name so
        two servers can expose tools with the same local name without conflict."""
        async def _one(srv: MCPServer) -> list[dict[str, Any]]:
            r = await self._client.get(f"{srv.url}/mcp/tools",
                                       headers=self._headers(srv))
            r.raise_for_status()
            tools = r.json().get("tools", [])
            for t in tools:
                t["name"] = f"{srv.name}.{t['name']}"
                t["__server__"] = srv.name
            return tools
 
        results = await asyncio.gather(*[_one(s) for s in self.servers],
                                       return_exceptions=True)
        flat: list[dict[str, Any]] = []
        for r in results:
            if isinstance(r, Exception):
                continue
            flat.extend(r)
        return flat
 
    # --- invocation --------------------------------------------------------
 
    async def call(self, namespaced_name: str, args: dict[str, Any]) -> Any:
        server_name, _, local = namespaced_name.partition(".")
        srv = next((s for s in self.servers if s.name == server_name), None)
        if srv is None:
            raise ValueError(f"unknown MCP server '{server_name}'")
        r = await self._client.post(
            f"{srv.url}/mcp/call",
            headers=self._headers(srv),
            json={"name": local, "args": args},
        )
        r.raise_for_status()
        payload = r.json()
        if "error" in payload:
            raise RuntimeError(payload["error"])
        return payload.get("result")

async def wire_mcp(harness: Harness, mcp: MCPClient) -> None:
    """Attach all MCP tools to the harness. The harness sandbox still applies."""
    schemas = await mcp.discover()
    for schema in schemas:
        name = schema["name"]
 
        async def _call(args, _name=name):
            return await mcp.call(_name, args)
 
        harness.register_tool(name, _call, schema)

model:
  provider: anthropic
  name: claude-sonnet-4-5
  temperature: 0.2
 
mcp_servers:
  - name: internal-crm
    url: https://mcp.internal/crm
    auth: oidc
 
  - name: code-executor
    url: https://mcp.internal/sandbox
    auth: mtls
 
skills:
  registry: ./skills
  auto_learn: true
  min_confidence: 0.4
 
memory:
  episodic: postgres://memory@db/episodic
  semantic: qdrant://memory:6333/agent-1
  max_working_tokens: 8000
 
runtime:
  max_steps: 25
  step_timeout_s: 20
  loop_window: 4

Primary Harness
     │
     ├──► [Spawn] Code Review Subagent (ephemeral container)
     │         └──► Returns: { verdict: "approved", comments: [...] }
     │
     ├──► [Spawn] Data Extraction Subagent (ephemeral container)
     │         └──► Returns: { records: [...], confidence: 0.94 }
     │
     └──► [Synthesize] Final result from subagent outputs

"""
Runtime-level multi-agent: spawn a fresh harness in an isolated container,
talk to it via a clean API, then tear it down. No shared memory, no shared
context window, no leakage between primary and subagent.
"""
 
from __future__ import annotations
 
import asyncio
import json
import secrets
from dataclasses import dataclass
from typing import Any
 
import httpx
 
 
@dataclass
class SubagentSpec:
    image: str = "ghcr.io/onemancrew/hermes-agent:latest"
    role: str = "generic"             # passed as env var; selects skill subset
    timeout_s: float = 120.0
    cpu: str = "1"
    memory_mb: int = 1024
 
 
class SubagentSpawner:
    """Thin wrapper over a container runtime. Replace the docker calls with
    the orchestrator of your choice — the contract is identical."""
 
    def __init__(self, network: str = "agent-net") -> None:
        self.network = network
        self._client = httpx.AsyncClient(timeout=30)
 
    async def _docker(self, *args: str) -> str:
        proc = await asyncio.create_subprocess_exec(
            "docker", *args,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )
        stdout, stderr = await proc.communicate()
        if proc.returncode != 0:
            raise RuntimeError(f"docker {args[0]} failed: {stderr.decode()}")
        return stdout.decode().strip()
 
    async def run_task(self, spec: SubagentSpec, goal: str,
                       inputs: dict[str, Any] | None = None) -> dict[str, Any]:
        name = f"subagent-{secrets.token_hex(4)}"
        token = secrets.token_urlsafe(24)
 
        # 1) start the ephemeral harness
        await self._docker(
            "run", "-d", "--rm",
            "--name", name,
            "--network", self.network,
            "--cpus", spec.cpu,
            "--memory", f"{spec.memory_mb}m",
            "-e", f"AGENT_ROLE={spec.role}",
            "-e", f"AGENT_TOKEN={token}",
            spec.image,
        )
 
        try:
            # 2) wait for the harness to come up
            url = f"http://{name}:8080"
            for _ in range(30):
                try:
                    r = await self._client.get(f"{url}/health", timeout=2)
                    if r.status_code == 200:
                        break
                except httpx.HTTPError:
                    await asyncio.sleep(0.5)
            else:
                raise TimeoutError(f"subagent {name} never became ready")
 
            # 3) submit the task
            r = await self._client.post(
                f"{url}/run",
                headers={"Authorization": f"Bearer {token}"},
                content=json.dumps({"goal": goal, "inputs": inputs or {}}),
                timeout=spec.timeout_s,
            )
            r.raise_for_status()
            return r.json()
        finally:
            # 4) tear down — context, memory, sandbox, all gone
            await self._docker("rm", "-f", name)

import asyncio
import logging
 
from harness_runtime import Harness, HybridMemory, Sandbox
from mcp_client import MCPClient, MCPServer
from skill_registry import SkillRegistry
from subagent_spawner import SubagentSpawner, SubagentSpec
from wire_mcp import wire_mcp
 
 
class FakeModel:
    """Stand-in model — in production this is Claude / GPT / Llama via API."""
    async def complete(self, messages, tools):
        # In a real run, the model picks a tool from `tools`, or returns final.
        return {"type": "final", "content": "demo: configure a real ModelClient"}
 
 
async def main() -> None:
    logging.basicConfig(level=logging.INFO)
 
    skills = SkillRegistry("./skills")
    mcp = MCPClient([
        MCPServer(name="internal-crm", url="https://mcp.internal/crm"),
        MCPServer(name="code-executor", url="https://mcp.internal/sandbox"),
    ])
    spawner = SubagentSpawner()
 
    harness = Harness(
        model=FakeModel(),
        memory=HybridMemory(max_working_tokens=8000),
        sandbox=Sandbox(timeout_s=20.0),
    )
 
    # Wire MCP — every tool from every server is now sandbox-protected.
    try:
        await wire_mcp(harness, mcp)
    except Exception as e:
        logging.warning("MCP discovery failed (offline?): %s", e)
 
    # Optional: register a "spawn_subagent" tool that the model can call.
    async def _spawn(args):
        return await spawner.run_task(
            SubagentSpec(role=args.get("role", "generic")),
            goal=args["goal"],
        )
 
    harness.register_tool(
        "spawn_subagent",
        _spawn,
        {
            "name": "spawn_subagent",
            "description": "Spawn an isolated subagent for a focused subtask.",
            "parameters": {
                "type": "object",
                "properties": {
                    "role": {"type": "string"},
                    "goal": {"type": "string"},
                },
                "required": ["goal"],
            },
        },
    )
 
    # Inject any matching skills BEFORE the run starts.
    goal = "Audit the latest production deploy for regressions."
    for s in skills.find(goal):
        harness.memory.append({"role": "system",
                               "content": f"[skill:{s.name}] {s.instructions}"})
 
    answer = await harness.run(goal)
    print("ANSWER:", answer)
 
    await mcp.aclose()
 
 
if __name__ == "__main__":
    asyncio.run(main())