Compositional execution-state management for long-horizon agentic code generation.
Agent Harness provides a robust framework for managing execution state in long-horizon agentic code generation loops. It tracks data dependencies across code steps, intelligently detects which steps need re-execution after feedback, generates targeted patches, and replays only the affected subgraph—enabling efficient iterative refinement of generated code.
Built for AI agent systems that must handle stateful, multi-turn code synthesis with automatic error recovery and deterministic replay semantics. Integrates seamlessly with LLM-based feedback loops, AST analysis, and execution sandboxes.
pip install agent-harnessfromagent_harness.esgimportExecutionStateGraphfromagent_harness.executorimportExecutorfromagent_harness.feedback_interpreterimportdiagnosefromagent_harness.patch_generatorimportgenerate_patch# Create an execution state graphesg=ExecutionStateGraph()
# Add and execute stepsstep_id=esg.add_step("step_0", "x = 5\ny = x + 3")
executor=Executor(namespace={})
result=executor.execute(step_id, "x = 5\ny = x + 3")
# Record outputs and track dependenciesesg.record_output(step_id, result.namespace)
# Add dependent steps with automatic edge detectionstep_id_2=esg.add_step("step_1", "z = y * 2")
esg.add_edge("step_0", "step_1")
# Diagnose failures and generate patchesdiagnosis=diagnose(result, esg)
patch=generate_patch(diagnosis, esg)
# Replay affected subgraphstale_steps=esg.mark_stale("step_0")
replay_order=esg.replay_from("step_0")Automatically infer read/write dependencies across code steps using AST analysis. Track which variables each step consumes and produces.
fromagent_harness.dependency_analyzerimportget_reads, get_writes, infer_edgesreads=get_reads("z = y * 2") # {'y'}writes=get_writes("x = 5; y = x + 3") # {'x', 'y'}steps= [("step_0", "x = 5"), ("step_1", "y = x + 3")]
edges=infer_edges(steps) # [('step_0', 'step_1')]Query ancestor steps needed to recompute a variable, mark stale steps, and replay only affected subgraphs.
ancestors=esg.get_ancestors("step_2", {"x"})
stale=esg.mark_stale("step_0")
subgraph=esg.get_subgraph_from("step_1")Diagnose execution failures and generate targeted code patches from error feedback.
diagnosis=diagnose(step_result, esg)
patch=generate_patch(diagnosis, esg)Agent Harness organizes execution state as a directed acyclic graph (DAG) where each node represents a code step and edges represent data dependencies.
ExecutionStateGraph (esg.py) manages the DAG, tracking step metadata, output namespaces, and staleness. DependencyAnalyzer (dependency_analyzer.py) uses AST introspection to infer variable read/write patterns and automatically detect dependencies. Executor (executor.py) runs steps in isolated namespaces and captures results. FeedbackInterpreter (feedback_interpreter.py) analyzes execution failures to produce diagnostic reports. PatchGenerator (patch_generator.py) synthesizes code patches from diagnoses.
The flow: steps are added → dependencies inferred → steps executed → outputs recorded → feedback triggers diagnosis → patches generated → staleness propagated → subgraph replayed.
esg=ExecutionStateGraph()
step_id=esg.add_step(step_id: str, source: str) ->stresg.record_output(step_id: str, namespace: dict)
esg.add_edge(src_step_id: str, dst_step_id: str) ->Noneancestors=esg.get_ancestors(step_id: str, variable_names: set[str] |None=None) ->list[str]
subgraph=esg.get_subgraph_from(step_id: str) ->list[str]
esg.mark_stale(step_id: str) ->Nonereplay_order=esg.replay_from(step_id: str) ->list[str]
node=esg.get_node(step_id: str) ->dict[str, Any]
graph=esg.graph() ->nx.DiGraphexecutor=Executor(namespace={})
result=executor.execute(step_id: str, source: str) ->StepResultns=result.namespace# dict[str, Any]diagnosis=diagnose(step_result: StepResult, esg: ExecutionStateGraph) ->DiagnosisReportpatch=generate_patch(diagnosis: DiagnosisReport, esg: ExecutionStateGraph) ->PatchResultfromagent_harness.dependency_analyzerimportget_reads, get_writes, infer_edgesreads=get_reads(source: str) ->set[str]
writes=get_writes(source: str) ->set[str]
edges=infer_edges(steps: list[tuple[str, str]]) ->list[tuple[str, str]]Agent Harness includes 268 comprehensive tests. Run the full suite with:
pytest tests/Contributions welcome! Please open issues and pull requests on GitHub. See docs at lumi-node.github.io/agent-harness.
MIT License — see LICENSE for details.