{"id":"agent-loop-prevention","version":"1.0.0","primitive":"code_execution","description":"Detect and break infinite agent loops using step counters, state hashing, and hard ceilings","registry_refs":["none (stdlib only)"],"tags":[],"solves":[],"auth_required":false,"verified":false,"last_verified":"null","next_check":"2026-07-19","eval_result":"null","eval_env":"null","mast":[],"ref":"https://arxiv.org/abs/2503.13657","inputs":[],"executable":"# ============================================\n# checklist:     agent-loop-prevention\n# version:       1.0.0\n# primitive:     code_execution\n# description:   Detect and break infinite agent loops using step counters, state hashing, and hard ceilings\n# registry_refs: none (stdlib only)\n# auth_required: false\n# verified:      false\n# last_verified: null\n# next_check:    2026-07-19\n# eval_result:   null\n# eval_env:      null\n#\n# INPUTS:\n#   - MAX_STEPS: hard ceiling on agent steps (default: 10)\n#   - MAX_REPEATED_STATES: consecutive identical states before abort (default: 3)\n#   - TASK: string description of agent task (default: demo task)\n#\n# OUTPUTS:\n#   - steps_executed: total steps run before termination\n#   - termination_reason: \"task_complete\" | \"max_steps\" | \"repeated_state\"\n#   - state_history: list of state hashes per step\n#   - result: structured dict\n#\n# MAST FAILURE MODES ADDRESSED:\n# FM-1.3 Step Repetition        — state hash detects identical states, aborts on repeat\n# FM-1.5 Unaware of Termination — MAX_STEPS hard ceiling, explicit termination conditions\n# FM-1.1 Disobey Task Spec      — task completion check on every step, not just at end\n# FM-3.2 No or Incomplete Verification — state logged every step, not just final\n# FM-3.3 Incorrect Verification — exact hash comparison, not fuzzy similarity\n#\n# ref: https://arxiv.org/abs/2503.13657\n# ============================================\n\nimport sys, hashlib, json, time\n\n# ── PRE_EXECUTION ─────────────────────────────────────────────\n# No external registry needed — stdlib only\n# Validate config before execution\nMAX_STEPS = 10\nMAX_REPEATED_STATES = 3\nTASK = \"find the value 42 in a list\"\n\nassert MAX_STEPS > 0, \"ABORT: MAX_STEPS must be > 0\"\nassert MAX_REPEATED_STATES > 0, \"ABORT: MAX_REPEATED_STATES must be > 0\"\nassert isinstance(TASK, str) and TASK.strip(), \"ABORT: TASK must be a non-empty string\"\n\nprint(f\"[CONFIG] MAX_STEPS={MAX_STEPS}, MAX_REPEATED_STATES={MAX_REPEATED_STATES}\")\nprint(f\"[TASK] {TASK}\")\n\n# ── EXECUTION ──────────────────────────────────────────────────\n\ndef hash_state(state: dict) -> str:\n    \"\"\"Deterministic hash of agent state — FM-1.3 loop detection.\"\"\"\n    serialized = json.dumps(state, sort_keys=True)\n    return hashlib.sha256(serialized.encode()).hexdigest()[:16]\n\ndef is_task_complete(state: dict) -> bool:\n    \"\"\"Binary completion check — FM-1.5 termination condition.\"\"\"\n    return state.get(\"found\") is True\n\ndef agent_step(state: dict, step: int) -> dict:\n    \"\"\"\n    Simulated agent step. In production replace with real LLM tool call.\n    Deliberately loops for first 4 steps to demonstrate detection,\n    then makes progress.\n    \"\"\"\n    new_state = dict(state)\n\n    if step < 4:\n        # Simulate agent stuck in loop — same action repeated\n        new_state[\"action\"] = \"search_list\"\n        new_state[\"position\"] = 0  # never advances — loop detected\n    else:\n        # Simulate agent making progress\n        new_state[\"action\"] = \"search_list\"\n        new_state[\"position\"] = step - 3\n        if new_state[\"position\"] >= 3:\n            new_state[\"found\"] = True\n            new_state[\"value\"] = 42\n\n    return new_state\n\n# Agent execution loop with loop prevention\nstate = {\"action\": None, \"position\": 0, \"found\": False, \"value\": None}\nstate_history = []\nrepeat_counter = 0\nsteps_executed = 0\ntermination_reason = None\nlast_hash = None\n\nt0 = time.time()\n\nfor step in range(1, MAX_STEPS + 1):\n    steps_executed = step\n\n    # Execute step\n    state = agent_step(state, step)\n    current_hash = hash_state(state)\n    state_history.append(current_hash)\n\n    print(f\"[STEP {step:02d}] hash={current_hash} action={state['action']} position={state['position']} found={state['found']}\")\n\n    # FM-1.3 — detect repeated state\n    if current_hash == last_hash:\n        repeat_counter += 1\n        print(f\"[LOOP-DETECT] Repeated state {repeat_counter}/{MAX_REPEATED_STATES}\")\n        if repeat_counter >= MAX_REPEATED_STATES:\n            termination_reason = \"repeated_state\"\n            print(f\"[ABORT] Repeated state ceiling hit at step {step} — breaking loop\")\n            break\n    else:\n        repeat_counter = 0\n\n    last_hash = current_hash\n\n    # FM-1.5 — check task completion every step\n    if is_task_complete(state):\n        termination_reason = \"task_complete\"\n        print(f\"[COMPLETE] Task finished at step {step}\")\n        break\n\nelse:\n    # FM-1.5 — hard ceiling reached\n    termination_reason = \"max_steps\"\n    print(f\"[ABORT] MAX_STEPS ceiling ({MAX_STEPS}) reached — hard stop\")\n\nelapsed_ms = int((time.time() - t0) * 1000)\n\n# ── POST_EXECUTION ─────────────────────────────────────────────\n# FM-3.2 — verify we have a recorded termination reason\nassert termination_reason is not None, \"FAIL: no termination reason recorded\"\n\n# FM-3.3 — exact termination condition check\nassert termination_reason in (\"task_complete\", \"max_steps\", \"repeated_state\"), (\n    f\"FAIL: unknown termination_reason '{termination_reason}'\"\n)\n\n# Verify step count is within bounds\nassert 1 <= steps_executed <= MAX_STEPS, (\n    f\"FAIL: steps_executed={steps_executed} out of bounds [1, {MAX_STEPS}]\"\n)\n\n# Verify state history was recorded\nassert len(state_history) == steps_executed, (\n    f\"FAIL: state_history length {len(state_history)} != steps_executed {steps_executed}\"\n)\n\nresult = {\n    \"steps_executed\": steps_executed,\n    \"termination_reason\": termination_reason,\n    \"max_steps\": MAX_STEPS,\n    \"state_history\": state_history,\n    \"elapsed_ms\": elapsed_ms,\n    \"task\": TASK,\n}\n\nprint(json.dumps(result, indent=2))\nprint(\"PASS\")"}