"""Single-Agent reference runtime: deterministic test model + optional real OpenAI Responses adapter. Python 3.10+; stdlib only. Real mode needs OPENAI_API_KEY and network. This is a research reference, not a production sandbox. """ from __future__ import annotations from dataclasses import dataclass, field, asdict from typing import Protocol import argparse, json, os, time, uuid, urllib.request, urllib.error, unittest class ContractError(Exception): pass class BudgetExceeded(Exception): pass class AuthorizationDenied(Exception): pass class ProviderError(Exception): pass @dataclass(frozen=True) class Config: max_steps:int=8 max_tool_calls:int=4 deadline_seconds:float=30 allowed_tools:tuple[str,...]=("search_evidence",) model:str="gpt-4.1-mini" def validate(self): if min(self.max_steps,self.max_tool_calls)<1 or self.deadline_seconds<=0: raise ContractError("invalid limits") if any(t not in ("search_evidence",) for t in self.allowed_tools): raise ContractError("unknown tool") @dataclass class Event: seq:int kind:str payload:dict @dataclass class State: run_id:str question:str status:str="RUNNING" steps:int=0 tool_calls:int=0 evidence:list[dict]=field(default_factory=list) answer:str="" events:list[Event]=field(default_factory=list) stop_reason:str="" def emit(self,kind,**payload): self.events.append(Event(len(self.events)+1,kind,payload)) class Model(Protocol): def decide(self,question:str,evidence:list[dict])->dict: ... class DeterministicModel: def decide(self,question,evidence): if not evidence: return {"kind":"tool","name":"search_evidence","arguments":{"query":question}} return {"kind":"finish","answer":"Evidence found: "+evidence[0]["text"],"citations":[evidence[0]["id"]]} class ResponsesModel: """Minimal OpenAI Responses HTTP adapter; only text JSON decisions, no provider-native tools. API compatibility must be rechecked for the model/account before real use. """ def __init__(self,model,timeout=20): self.model=model; self.timeout=timeout self.key=os.getenv("OPENAI_API_KEY") if not self.key: raise ProviderError("OPENAI_API_KEY is required") def decide(self,question,evidence): instruction=("Return only a JSON object. Allowed shapes: " '{"kind":"tool","name":"search_evidence","arguments":{"query":"..."}} or ' '{"kind":"finish","answer":"...","citations":["source-id"]} or ' '{"kind":"stop","reason":"..."}. ' "Use tool if evidence empty. Only cite IDs supplied in evidence. Never invent sources.") payload={"model":self.model,"instructions":instruction, "input":json.dumps({"question":question,"evidence":evidence},ensure_ascii=False), "max_output_tokens":500} req=urllib.request.Request("https://api.openai.com/v1/responses", data=json.dumps(payload).encode(),method="POST", headers={"Authorization":"Bearer "+self.key,"Content-Type":"application/json"}) try: with urllib.request.urlopen(req,timeout=self.timeout) as resp: body=json.load(resp) except (urllib.error.URLError,TimeoutError,ValueError) as e: raise ProviderError(type(e).__name__) from e texts=[] for item in body.get("output",[]): for part in item.get("content",[]): if part.get("type")=="output_text": texts.append(part.get("text","")) if not texts: raise ProviderError("no output_text returned") try: return json.loads("".join(texts)) except json.JSONDecodeError as e: raise ProviderError("invalid decision JSON") from e class EvidenceTool: def __init__(self): self.docs={"agent":{"id":"doc-agent-1","text":"An agent chooses actions using observations."}} def invoke(self,query): if not isinstance(query,str) or not query.strip(): raise ContractError("query required") return self.docs.get("agent",{}).copy() if "agent" in query.lower() else {} def parse_decision(obj): if not isinstance(obj,dict): raise ContractError("decision must be object") kind=obj.get("kind") if kind=="tool": if set(obj)!={"kind","name","arguments"} or obj["name"]!="search_evidence": raise ContractError("invalid tool decision") args=obj["arguments"] if not isinstance(args,dict) or set(args)!={"query"} or not isinstance(args["query"],str): raise ContractError("invalid tool arguments") elif kind=="finish": if set(obj)!={"kind","answer","citations"} or not isinstance(obj["answer"],str) or not isinstance(obj["citations"],list) or any(not isinstance(x,str) for x in obj["citations"]): raise ContractError("invalid finish") elif kind=="stop": if set(obj)!={"kind","reason"} or not isinstance(obj["reason"],str): raise ContractError("invalid stop") else: raise ContractError("unknown decision kind") return obj class Controller: def __init__(self,model:Model,config:Config,tool=None,clock=time.monotonic): config.validate(); self.model=model;self.config=config;self.tool=tool or EvidenceTool();self.clock=clock def run(self,question): if not isinstance(question,str) or not question.strip(): raise ContractError("question required") s=State(uuid.uuid4().hex,question) start=self.clock() s.emit("START",question=question) while s.status=="RUNNING": if self.clock()-start>=self.config.deadline_seconds: s.status="STOPPED";s.stop_reason="DEADLINE";break if s.steps>=self.config.max_steps: s.status="STOPPED";s.stop_reason="STEP_BUDGET";break s.steps+=1 try: raw=self.model.decide(s.question,[x.copy() for x in s.evidence]) action=parse_decision(raw) except (ProviderError,ContractError) as e: s.status="FAILED";s.stop_reason=type(e).__name__;s.emit("MODEL_ERROR",error=type(e).__name__);break s.emit("DECISION",decision_kind=action["kind"]) if action["kind"]=="stop": s.status="STOPPED";s.stop_reason=action["reason"];break if action["kind"]=="finish": known={x["id"] for x in s.evidence} if not set(action["citations"]).issubset(known): s.status="FAILED";s.stop_reason="UNVERIFIED_CITATION";break s.answer=action["answer"];s.status="COMPLETED";break if action["name"] not in self.config.allowed_tools: s.status="FAILED";s.stop_reason="DENIED";s.emit("TOOL_DENIED",name=action["name"]);break if s.tool_calls>=self.config.max_tool_calls: s.status="STOPPED";s.stop_reason="TOOL_BUDGET";break s.tool_calls+=1 try: observation=self.tool.invoke(**action["arguments"]) if observation: if not all(k in observation for k in ("id","text")): raise ContractError("invalid observation") if observation["id"] not in {x["id"] for x in s.evidence}:s.evidence.append(observation) s.emit("TOOL_RESULT",name=action["name"],found=bool(observation)) except (ContractError,TimeoutError) as e: s.emit("TOOL_ERROR",error=type(e).__name__) s.status="FAILED";s.stop_reason=type(e).__name__ s.emit("STOP",status=s.status,reason=s.stop_reason) return s class BadCitation: def decide(self,q,e): return {"kind":"finish","answer":"unsupported","citations":["fake"]} class BadFormat: def decide(self,q,e): return {"kind":"tool","name":"search_evidence","arguments":{"query":12}} class LoopModel: def decide(self,q,e): return {"kind":"tool","name":"search_evidence","arguments":{"query":"missing"}} class ErrorTool: def invoke(self,query): raise TimeoutError("injected") class StopModel: def decide(self,q,e): return {"kind":"stop","reason":"INSUFFICIENT_EVIDENCE"} class Tests(unittest.TestCase): def test_success(self): s=Controller(DeterministicModel(),Config()).run("agent") self.assertEqual(s.status,"COMPLETED");self.assertEqual(s.tool_calls,1) def test_events_ordered(self): s=Controller(DeterministicModel(),Config()).run("agent") self.assertEqual([e.seq for e in s.events],list(range(1,len(s.events)+1))) def test_unverified_citation(self): self.assertEqual(Controller(BadCitation(),Config()).run("agent").stop_reason,"UNVERIFIED_CITATION") def test_invalid_decision(self): self.assertEqual(Controller(BadFormat(),Config()).run("agent").status,"FAILED") def test_permission(self): s=Controller(DeterministicModel(),Config(allowed_tools=())).run("agent") self.assertEqual(s.stop_reason,"DENIED");self.assertEqual(s.tool_calls,0) def test_tool_budget(self): s=Controller(LoopModel(),Config(max_tool_calls=2)).run("agent") self.assertEqual(s.stop_reason,"TOOL_BUDGET");self.assertEqual(s.tool_calls,2) def test_step_budget(self): s=Controller(LoopModel(),Config(max_steps=2)).run("agent") self.assertEqual(s.stop_reason,"STEP_BUDGET") def test_tool_timeout(self): s=Controller(DeterministicModel(),Config(),ErrorTool()).run("agent") self.assertEqual(s.status,"FAILED") def test_explicit_stop(self): self.assertEqual(Controller(StopModel(),Config()).run("agent").stop_reason,"INSUFFICIENT_EVIDENCE") def test_bad_config(self): with self.assertRaises(ContractError):Config(max_steps=0).validate() def test_bad_input(self): with self.assertRaises(ContractError):Controller(DeterministicModel(),Config()).run("") def test_deadline(self): ticks=iter([0,1]) s=Controller(LoopModel(),Config(deadline_seconds=0.5),clock=lambda:next(ticks)).run("agent") self.assertEqual(s.stop_reason,"DEADLINE") if __name__=="__main__": ap=argparse.ArgumentParser() ap.add_argument("--mode",choices=["mock","real"],default="mock") ap.add_argument("--model",default="gpt-4.1-mini") ap.add_argument("--question",default="agent architecture") ap.add_argument("--test",action="store_true") ap.add_argument("--output",default="") args=ap.parse_args() if args.test: unittest.main(argv=["runtime.py"],verbosity=2) else: model=DeterministicModel() if args.mode=="mock" else ResponsesModel(args.model) state=Controller(model,Config(model=args.model)).run(args.question) result=json.dumps(asdict(state),ensure_ascii=False,indent=2) if args.output: open(args.output,"w",encoding="utf-8").write(result) print(result)