Lab Objective

Practice the integrity controls behind a local agent relay without using a live browser conversation or a real coding worker.

The lab focuses on three questions:

  • Is the next instruction genuinely newer than the last one consumed?
  • Does one active loop own the worker?
  • Does a failed handover repeat work or leave a recoverable state?

Lab Environment

  • Language: Python
  • State: local JSON or SQLite fixture
  • External accounts: none
  • Worker: stubbed function
  • Goal: validate ordering and state transitions, not model quality

Scenario

A local relay reads a conversation, hands the most recent instruction to a worker, and later posts the worker’s result. If it reads its own old post as a new instruction, or two relays run concurrently, the system can loop or duplicate work.

Step 1 - Represent the Consumed State

state = {
    "conversation_id": "chat-1",
    "last_consumed_hash": "abc123",
    "round": 4,
}

The state records what the relay has already acted on. It does not rely only on a visible message count, because counts can change when the UI rerenders or when the relay posts its own handover.

Step 2 - Reject a Stale Instruction

import hashlib

def digest(text: str) -> str:
    return hashlib.sha256(text.encode()).hexdigest()

def is_new_instruction(text: str, state: dict) -> bool:
    return digest(text) != state["last_consumed_hash"]

assert is_new_instruction("new task", state)
assert not is_new_instruction("old task", {**state, "last_consumed_hash": digest("old task")})

This is a minimal freshness gate. In a production relay, identity and document position should also be verified.

Step 3 - Enforce One Active Loop

from pathlib import Path

lock = Path("/tmp/agent-bridge-lab.lock")

def acquire_lock() -> bool:
    if lock.exists():
        return False
    lock.write_text("active")
    return True

assert acquire_lock() is True
assert acquire_lock() is False
lock.unlink()

The exact implementation can vary, but the property matters: a second local loop must refuse before it competes for the same model and task state.

Step 4 - Model the Handover Transition

record = {"instruction": "new task", "state": "received"}

record["state"] = "consumed"
try:
    # worker_result = run_stubbed_worker(record["instruction"])
    worker_result = "completed safely"
    record["handover"] = worker_result
    record["state"] = "posted"
except Exception:
    record["state"] = "needs_recovery"

Writing state around the transition gives a later recovery routine something concrete to inspect instead of rerunning blindly.

Security Takeaways

  • A relay needs a durable definition of “already consumed.”
  • One task loop should own one worker context at a time.
  • Browser-visible order is not a safe database.
  • Output still needs redaction before it crosses into another tool or chat.
  • Recovery must distinguish “work never started” from “work finished but handover failed.”

Reflection

The model is not the only moving part in an agent loop. Identity, freshness, locking, and durable transitions decide whether the loop does the intended work once or repeats it until someone notices.

TL;DR

Use explicit consumed-state, a single-loop lock, and recoverable handover states before trusting an agent relay to run repeatedly.