temp/.opencode/agents/shared/coordination_protocols.md

Agent Coordination Protocols

This document defines mandatory coordination protocols to prevent conflicts when multiple agents work on the same repository.

Work Item Claiming Protocol

CRITICAL: Before starting work on ANY issue or PR, agents MUST claim it to prevent conflicts.

Constants

CLAIM_DURATION_HOURS = 2          # Claims expire after 2 hours
HEARTBEAT_INTERVAL_MINUTES = 10   # Must send heartbeat every 10 minutes
CLAIM_PREFIX = "[CLAIM:"          # Claim comment prefix
HEARTBEAT_PREFIX = "[HEARTBEAT:"  # Heartbeat comment prefix  
RELEASE_PREFIX = "[RELEASE:"      # Release comment prefix

Claiming a Work Item

def claim_work_item(owner, repo, item_type, item_number, agent_name, session_id):
    """
    Claim exclusive access to a work item (issue or PR).
    
    Args:
        owner: Repository owner
        repo: Repository name
        item_type: "issue" or "PR"
        item_number: Issue or PR number
        agent_name: Name of the claiming agent
        session_id: Unique session identifier
        
    Returns:
        (success: bool, claim_id: str, error: str)
    """
    
    # Check if already claimed
    available, current_claim = is_item_available(owner, repo, item_number)
    
    if not available:
        return (False, None, f"{item_type} #{item_number} is already claimed by {current_claim['agent']}")
    
    # Generate claim ID
    claim_id = f"{agent_name}-{session_id}-{int(time.time())}"
    
    # Post claim comment
    claim_time = datetime.utcnow()
    expire_time = claim_time + timedelta(hours=CLAIM_DURATION_HOURS)
    
    claim_comment = f"""{CLAIM_PREFIX}{claim_id}]
Agent: {agent_name}
Session: {session_id}
Started: {claim_time.isoformat()}Z
Expires: {expire_time.isoformat()}Z
Status: ACTIVE

⚠️ This {item_type} is now exclusively claimed by the above agent.
Other agents must wait for the claim to expire or be released.
Claim automatically expires after {CLAIM_DURATION_HOURS} hours without heartbeat.

---
**Automated by CleverAgents Bot**
Agent: {agent_name}"""

    try:
        forgejo_create_issue_comment(owner, repo, item_number, claim_comment)
        print(f"[CLAIMED] {item_type} #{item_number} claimed with ID: {claim_id}")
        return (True, claim_id, None)
    except Exception as e:
        return (False, None, f"Failed to post claim: {str(e)}")

Checking Availability

def is_item_available(owner, repo, item_number):
    """
    Check if a work item is available for claiming.
    
    Returns:
        (available: bool, active_claim: dict or None)
    """
    
    # Get recent comments (last 48 hours to catch expired claims)
    since = (datetime.utcnow() - timedelta(hours=48)).isoformat() + "Z"
    comments = forgejo_list_issue_comments(owner, repo, item_number, since=since)
    
    active_claims = []
    releases = []
    heartbeats = {}
    
    # Parse all coordination comments
    for comment in comments:
        body = comment.get('body', '')
        
        if CLAIM_PREFIX in body:
            claim_data = parse_claim_comment(comment)
            if claim_data:
                active_claims.append(claim_data)
                
        elif HEARTBEAT_PREFIX in body:
            hb_data = parse_heartbeat_comment(comment) 
            if hb_data:
                claim_id = hb_data['claim_id']
                # Keep only latest heartbeat per claim
                if claim_id not in heartbeats or hb_data['time'] > heartbeats[claim_id]['time']:
                    heartbeats[claim_id] = hb_data
                    
        elif RELEASE_PREFIX in body:
            rel_data = parse_release_comment(comment)
            if rel_data:
                releases.append(rel_data)
    
    # Find active claims
    now = datetime.utcnow()
    for claim in active_claims:
        claim_id = claim['claim_id']
        
        # Check if explicitly released
        if any(r['claim_id'] == claim_id for r in releases):
            continue
            
        # Check if expired
        expire_time = claim['expires']
        last_heartbeat = heartbeats.get(claim_id, {}).get('time', claim['started'])
        
        time_since_activity = now - last_heartbeat
        
        if time_since_activity < timedelta(hours=CLAIM_DURATION_HOURS):
            # Claim is still active
            return (False, claim)
    
    # No active claims
    return (True, None)

Sending Heartbeats

def send_heartbeat(owner, repo, item_number, claim_id, status_update=None):
    """
    Send heartbeat to maintain claim on a work item.
    Should be called every HEARTBEAT_INTERVAL_MINUTES.
    
    Args:
        status_update: Optional status message (e.g., "Running tests...")
    """
    
    heartbeat_comment = f"""{HEARTBEAT_PREFIX}{claim_id}]
Time: {datetime.utcnow().isoformat()}Z  
Status: {status_update or "Working..."}

---
**Automated by CleverAgents Bot**"""
    
    try:
        forgejo_create_issue_comment(owner, repo, item_number, heartbeat_comment)
        return True
    except:
        print(f"[WARNING] Failed to send heartbeat for claim {claim_id}")
        return False

Releasing a Claim

def release_claim(owner, repo, item_number, claim_id, reason="completed"):
    """
    Explicitly release a claim on a work item.
    
    Args:
        reason: Why the claim is being released
                "completed" - Work finished successfully
                "failed" - Could not complete the work
                "timeout" - Taking too long
                "error" - Unexpected error occurred
    """
    
    release_comment = f"""{RELEASE_PREFIX}{claim_id}]  
Time: {datetime.utcnow().isoformat()}Z
Reason: {reason}

✓ This work item is now available for other agents.

---
**Automated by CleverAgents Bot**"""
    
    try:
        forgejo_create_issue_comment(owner, repo, item_number, release_comment)
        print(f"[RELEASED] Claim {claim_id} released: {reason}")
        return True
    except:
        print(f"[ERROR] Failed to release claim {claim_id}")
        return False

Helper Parsers

def parse_claim_comment(comment):
    """Parse a claim comment into structured data."""
    body = comment.get('body', '')
    if CLAIM_PREFIX not in body:
        return None
        
    try:
        # Extract claim ID
        claim_id = body[body.find(CLAIM_PREFIX) + len(CLAIM_PREFIX):body.find(']')]
        
        # Extract fields
        lines = body.split('\n')
        data = {'claim_id': claim_id}
        
        for line in lines:
            if line.startswith('Agent:'):
                data['agent'] = line.split(':', 1)[1].strip()
            elif line.startswith('Session:'):
                data['session'] = line.split(':', 1)[1].strip()
            elif line.startswith('Started:'):
                data['started'] = datetime.fromisoformat(line.split(':', 1)[1].strip().rstrip('Z'))
            elif line.startswith('Expires:'):
                data['expires'] = datetime.fromisoformat(line.split(':', 1)[1].strip().rstrip('Z'))
                
        return data
    except:
        return None


def parse_heartbeat_comment(comment):
    """Parse a heartbeat comment."""
    body = comment.get('body', '')
    if HEARTBEAT_PREFIX not in body:
        return None
        
    try:
        claim_id = body[body.find(HEARTBEAT_PREFIX) + len(HEARTBEAT_PREFIX):body.find(']')]
        
        lines = body.split('\n')
        for line in lines:
            if line.startswith('Time:'):
                time_str = line.split(':', 1)[1].strip().rstrip('Z')
                return {
                    'claim_id': claim_id,
                    'time': datetime.fromisoformat(time_str)
                }
    except:
        return None


def parse_release_comment(comment):
    """Parse a release comment."""
    body = comment.get('body', '')
    if RELEASE_PREFIX not in body:
        return None
        
    try:
        claim_id = body[body.find(RELEASE_PREFIX) + len(RELEASE_PREFIX):body.find(']')]
        return {'claim_id': claim_id}
    except:
        return None

Concurrent Work Detection

def detect_concurrent_work(owner, repo, pr_number):
    """
    Detect if other agents are actively working on the same PR.
    Used to coordinate work that might conflict.
    """
    
    # Check claims in last 30 minutes
    since = (datetime.utcnow() - timedelta(minutes=30)).isoformat() + "Z"
    comments = forgejo_list_issue_comments(owner, repo, pr_number, since=since)
    
    active_agents = []
    
    for comment in comments:
        if "[WORKING]" in comment.get('body', ''):
            # Extract agent info
            agent_match = re.search(r'Agent: (\S+)', comment['body'])
            work_match = re.search(r'Work type: (\S+)', comment['body'])
            
            if agent_match:
                active_agents.append({
                    'agent': agent_match.group(1),
                    'work_type': work_match.group(1) if work_match else 'unknown',
                    'time': comment['created_at']
                })
    
    return active_agents


def coordinate_pr_work(owner, repo, pr_number, agent_name, work_type):
    """
    Coordinate work on a PR with other agents.
    
    Work types that conflict:
    - "code-change" conflicts with "code-change"
    - "merge-attempt" conflicts with everything
    - "review" does not conflict with anything
    
    Returns: "proceed", "wait", or "abort"
    """
    
    active_agents = detect_concurrent_work(owner, repo, pr_number)
    
    if not active_agents:
        return "proceed"
    
    # Define conflict matrix
    CONFLICTS = {
        'code-change': ['code-change', 'merge-attempt'],
        'merge-attempt': ['code-change', 'merge-attempt', 'test-fix'],
        'test-fix': ['code-change', 'merge-attempt', 'test-fix'],
        'review': []  # Reviews don't conflict
    }
    
    conflicting_work = CONFLICTS.get(work_type, [])
    
    for agent in active_agents:
        if agent['work_type'] in conflicting_work:
            # Check how long they've been working
            work_duration = datetime.utcnow() - datetime.fromisoformat(agent['time'])
            
            if work_duration < timedelta(minutes=15):
                print(f"[COORDINATION] Waiting for {agent['agent']} to complete {agent['work_type']}")
                return "wait"
            else:
                # They've been working too long, might be stuck
                print(f"[COORDINATION] {agent['agent']} has been working for {work_duration}, proceeding anyway")
    
    return "proceed"

Usage in Agents

Example: Implementation Worker

# At the start of work
def begin_work_on_issue(owner, repo, issue_number, agent_name, session_id):
    # Claim the issue
    success, claim_id, error = claim_work_item(
        owner, repo, "issue", issue_number, agent_name, session_id
    )
    
    if not success:
        print(f"[ABORT] Could not claim issue: {error}")
        return False
        
    # Set up heartbeat schedule
    last_heartbeat = time.time()
    
    try:
        # Do the work...
        while not work_complete:
            # Send heartbeat if needed
            if time.time() - last_heartbeat > HEARTBEAT_INTERVAL_MINUTES * 60:
                send_heartbeat(owner, repo, issue_number, claim_id, 
                             f"Implementing subtask {current_subtask}")
                last_heartbeat = time.time()
            
            # Continue work...
            
    finally:
        # Always release claim
        release_claim(owner, repo, issue_number, claim_id, 
                     "completed" if work_complete else "failed")

Example: PR Work Coordination

# Before making changes to a PR
def work_on_pr(owner, repo, pr_number, agent_name, work_type):
    # Check coordination
    action = coordinate_pr_work(owner, repo, pr_number, agent_name, work_type)
    
    if action == "wait":
        # Wait and retry
        print("[WAITING] Another agent is working on this PR")
        time.sleep(300)  # 5 minutes
        return work_on_pr(owner, repo, pr_number, agent_name, work_type)
        
    elif action == "abort":
        print("[ABORT] Cannot work on PR due to conflicts")
        return False
        
    # Proceed with work
    # Post working indicator
    forgejo_create_issue_comment(owner, repo, pr_number,
        f"[WORKING] Starting {work_type}\n" +
        f"Agent: {agent_name}\n" + 
        f"Work type: {work_type}\n\n" +
        "---\n**Automated by CleverAgents Bot**")
    
    # Do the work...

Enforcement Rules

1. No work without claims - Agents MUST NOT start work without successfully claiming
2. Respect existing claims - Agents MUST check availability before claiming
3. Send regular heartbeats - Claims without heartbeats expire and work may be lost
4. Always release claims - Use try/finally to ensure claims are released
5. Coordinate conflicting work - Check for concurrent agents on PRs

System Watchdog Monitoring

The system-watchdog will monitor for:

• Work started without claims
• Expired claims with agents still working
• Missing heartbeats on active work
• Unreleased claims after work completion
• Agents ignoring existing claims