cleveragents-core/.opencode/agents/shared/session_state.md

Session State Persistence Patterns

This document defines standardized patterns for persisting agent session state to enable crash recovery and work resumption.

Core Principles

1. Persist via Forgejo - All state lives in Forgejo comments, not local files
2. Checkpoint frequently - After each major operation
3. Versioned format - Enable future migration
4. Human readable - Comments should be understandable
5. Atomic updates - Each checkpoint is a complete snapshot

Session State Issue

All agents coordinate through a single session state issue created by the product-builder:

# Session state issue has a standardized title
SESSION_STATE_TITLE_PATTERN = "[Automated] CleverAgents Build Session"

def find_session_state_issue(owner, repo):
    """Find the active session state issue."""
    issues = forgejo_list_repo_issues(
        owner, repo, 
        state="open",
        labels="Type/Automation"
    )
    
    for issue in issues:
        if SESSION_STATE_TITLE_PATTERN in issue['title']:
            return issue['number']
    
    raise ValueError("No active session state issue found")

Checkpoint Format

import json
from datetime import datetime

CHECKPOINT_VERSION = "1.0"

def create_checkpoint(agent_name, session_id, phase, data):
    """
    Create a standardized checkpoint.
    
    Args:
        agent_name: Name of the agent creating checkpoint
        session_id: Unique session identifier
        phase: Current phase/step of work
        data: Dict of state data to persist
        
    Returns:
        Checkpoint dict
    """
    checkpoint = {
        "version": CHECKPOINT_VERSION,
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "agent": agent_name,
        "session": session_id,
        "phase": phase,
        "data": data
    }
    
    return checkpoint


def format_checkpoint_comment(checkpoint):
    """Format checkpoint for Forgejo comment."""
    
    comment = f"""[CHECKPOINT] {checkpoint['agent']}
Version: {checkpoint['version']}
Phase: {checkpoint['phase']}
Session: {checkpoint['session']}
Time: {checkpoint['timestamp']}

State Data:
```json
{json.dumps(checkpoint['data'], indent=2, sort_keys=True)}

---

Automated by CleverAgents Bot Agent: {checkpoint['agent']}"""

return comment

## Saving Checkpoints

```python
def save_checkpoint(owner, repo, session_issue, checkpoint):
    """
    Save checkpoint to session state issue.
    
    Args:
        owner: Repository owner
        repo: Repository name
        session_issue: Session state issue number
        checkpoint: Checkpoint dict from create_checkpoint()
    """
    comment = format_checkpoint_comment(checkpoint)
    
    try:
        forgejo_create_issue_comment(owner, repo, session_issue, comment)
        print(f"[CHECKPOINT] Saved: {checkpoint['phase']}")
    except Exception as e:
        # Checkpoint failure shouldn't crash the agent
        print(f"[WARNING] Failed to save checkpoint: {e}")


def save_state(owner, repo, agent_name, session_id, phase, **kwargs):
    """
    Convenience function to create and save checkpoint in one call.
    
    Args:
        owner, repo: Repository location
        agent_name: Agent creating checkpoint
        session_id: Session identifier
        phase: Current work phase
        **kwargs: State data to save
    """
    session_issue = find_session_state_issue(owner, repo)
    checkpoint = create_checkpoint(agent_name, session_id, phase, kwargs)
    save_checkpoint(owner, repo, session_issue, checkpoint)

Reading Checkpoints

def parse_checkpoint_comment(comment_body):
    """Parse a checkpoint from comment body."""
    if "[CHECKPOINT]" not in comment_body:
        return None
    
    try:
        # Extract JSON block
        json_start = comment_body.find("```json\n") + 8
        json_end = comment_body.find("\n```", json_start)
        
        if json_start > 7 and json_end > json_start:
            json_str = comment_body[json_start:json_end]
            data = json.loads(json_str)
            
            # Extract metadata
            lines = comment_body.split('\n')
            checkpoint = {"data": data}
            
            for line in lines:
                if line.startswith("Version:"):
                    checkpoint["version"] = line.split(":", 1)[1].strip()
                elif line.startswith("Phase:"):
                    checkpoint["phase"] = line.split(":", 1)[1].strip()
                elif line.startswith("Session:"):
                    checkpoint["session"] = line.split(":", 1)[1].strip()
                elif line.startswith("Time:"):
                    checkpoint["timestamp"] = line.split(":", 1)[1].strip()
                elif line.startswith("[CHECKPOINT]"):
                    checkpoint["agent"] = line.split("]", 1)[1].strip()
            
            return checkpoint
    except Exception as e:
        print(f"[WARNING] Failed to parse checkpoint: {e}")
        return None


def load_latest_checkpoint(owner, repo, agent_name=None, session_id=None):
    """
    Load the most recent checkpoint.
    
    Args:
        owner, repo: Repository location  
        agent_name: Filter by agent (optional)
        session_id: Filter by session (optional)
        
    Returns:
        Latest checkpoint dict or None
    """
    session_issue = find_session_state_issue(owner, repo)
    
    # Get recent comments (last 24 hours)
    comments = forgejo_list_issue_comments(
        owner, repo, session_issue,
        since=(datetime.utcnow() - timedelta(hours=24)).isoformat() + "Z"
    )
    
    checkpoints = []
    for comment in comments:
        checkpoint = parse_checkpoint_comment(comment['body'])
        if checkpoint:
            # Apply filters
            if agent_name and checkpoint.get('agent') != agent_name:
                continue
            if session_id and checkpoint.get('session') != session_id:
                continue
            
            checkpoints.append(checkpoint)
    
    if not checkpoints:
        return None
    
    # Sort by timestamp and return latest
    checkpoints.sort(key=lambda x: x.get('timestamp', ''), reverse=True)
    return checkpoints[0]

Resume Patterns

Agent-Specific Resume

def resume_agent_work(owner, repo, agent_name, session_id):
    """Resume work from last checkpoint."""
    
    checkpoint = load_latest_checkpoint(owner, repo, agent_name, session_id)
    
    if not checkpoint:
        print(f"[RESUME] No checkpoint found, starting fresh")
        return None, "fresh_start"
    
    phase = checkpoint.get('phase', 'unknown')
    data = checkpoint.get('data', {})
    
    print(f"[RESUME] Found checkpoint at phase: {phase}")
    print(f"[RESUME] Checkpoint time: {checkpoint.get('timestamp')}")
    
    return data, phase

Implementation Worker Resume

def resume_implementation(owner, repo, session_id, working_dir):
    """Resume implementation work from checkpoint."""
    
    data, phase = resume_agent_work(
        owner, repo, "implementation-worker", session_id
    )
    
    if phase == "fresh_start":
        return "start", {}
    
    # Map phases to resume points
    PHASE_RESUME_MAP = {
        "clone_complete": "phase_2_subtasks",
        "subtask_1_complete": "phase_2_continue", 
        "subtask_2_complete": "phase_2_continue",
        "all_subtasks_complete": "phase_3_commit",
        "commit_complete": "phase_4_pr",
        "pr_created": "phase_4_monitoring"
    }
    
    resume_point = PHASE_RESUME_MAP.get(phase, "start")
    
    # Restore state
    state = {
        "issue_number": data.get("issue_number"),
        "branch_name": data.get("branch_name"),
        "completed_subtasks": data.get("completed_subtasks", []),
        "pr_number": data.get("pr_number"),
        "working_dir": working_dir or data.get("working_dir")
    }
    
    return resume_point, state

Standard Checkpoints by Agent Type

Implementation Agents

# After clone
save_state(owner, repo, "implementation-worker", session_id, "clone_complete",
    issue_number=issue_number,
    branch_name=branch_name,
    working_dir=working_dir
)

# After each subtask
save_state(owner, repo, "implementation-worker", session_id, 
    f"subtask_{subtask_num}_complete",
    issue_number=issue_number,
    branch_name=branch_name,
    completed_subtasks=completed_subtasks,
    working_dir=working_dir
)

# After commit
save_state(owner, repo, "implementation-worker", session_id, "commit_complete",
    issue_number=issue_number,
    branch_name=branch_name,
    commit_sha=commit_sha,
    working_dir=working_dir
)

# After PR creation
save_state(owner, repo, "implementation-worker", session_id, "pr_created",
    issue_number=issue_number,
    pr_number=pr_number,
    branch_name=branch_name
)

Pool Supervisors

# Periodic health checkpoint
save_state(owner, repo, "implementation-pool-supervisor", session_id, "health_check",
    cycle_count=cycle_count,
    active_workers=active_workers,
    completed_issues=completed_issues,
    failed_issues=failed_issues,
    queue_depth=queue_depth
)

# After dispatching worker
save_state(owner, repo, "implementation-pool-supervisor", session_id, "worker_dispatched",
    worker_session_id=worker_session_id,
    issue_number=issue_number,
    dispatch_time=dispatch_time,
    active_workers=active_workers
)

Review Agents

# After posting review
save_state(owner, repo, "pr-reviewer", session_id, "review_posted",
    pr_number=pr_number,
    review_state=review_state,  # APPROVE, REQUEST_CHANGES, COMMENT
    findings_count=len(findings)
)

Health Monitoring Pattern

def post_health_signal(owner, repo, agent_name, session_id, metrics):
    """Post a health signal to session state."""
    
    comment = f"""[HEALTH] {agent_name}
Session: {session_id}
Time: {datetime.utcnow().isoformat()}Z
Status: Active

Metrics:
```json
{json.dumps(metrics, indent=2)}

Next check: in 10 iterations

---

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

session_issue = find_session_state_issue(owner, repo)
forgejo_create_issue_comment(owner, repo, session_issue, comment)

## Best Practices

1. **Checkpoint after expensive operations** - Don't redo costly work
2. **Include enough state to resume** - But not excessive detail
3. **Use consistent phase names** - Makes resume logic clearer
4. **Version your checkpoints** - Enables format evolution
5. **Handle missing checkpoints gracefully** - Always have a fresh start path

## Example: Complete Implementation with Checkpoints

```python
def implementation_with_checkpoints():
    """Example showing checkpoint integration."""
    
    # Check for resume
    data, phase = resume_agent_work(owner, repo, agent_name, session_id)
    
    if phase == "clone_complete":
        # Skip clone, restore state
        working_dir = data["working_dir"]
        branch_name = data["branch_name"]
    else:
        # Fresh start - do clone
        working_dir = clone_repository()
        branch_name = create_branch()
        
        # Checkpoint
        save_state(owner, repo, agent_name, session_id, "clone_complete",
            working_dir=working_dir,
            branch_name=branch_name,
            issue_number=issue_number
        )
    
    # Continue with subtasks...
    if phase in ["fresh_start", "clone_complete"]:
        completed = data.get("completed_subtasks", [])
        
        for i, subtask in enumerate(subtasks):
            if i in completed:
                print(f"[SKIP] Subtask {i} already completed")
                continue
                
            implement_subtask(subtask)
            completed.append(i)
            
            # Checkpoint after each
            save_state(owner, repo, agent_name, session_id, 
                f"subtask_{i}_complete",
                working_dir=working_dir,
                branch_name=branch_name,
                completed_subtasks=completed,
                issue_number=issue_number
            )
    
    # And so on...

Anti-Patterns to Avoid

# ❌ WRONG - Local file checkpoints
with open("/tmp/checkpoint.json", "w") as f:
    json.dump(state, f)

# ✓ CORRECT - Forgejo comment checkpoints
save_checkpoint(owner, repo, session_issue, checkpoint)

# ❌ WRONG - Huge checkpoints
save_state(..., entire_file_contents=file_contents)  # Too big

# ✓ CORRECT - Just enough to resume
save_state(..., files_modified=["file1.py", "file2.py"])

# ❌ WRONG - No version in checkpoint
data = {"stuff": 123}

# ✓ CORRECT - Versioned format
checkpoint = create_checkpoint(agent, session, phase, data)