forked from HAL9000/cleveragents-core
e5f75c5c83
- Remove maximum cap (16) on CA_MAX_PARALLEL_WORKERS in resources.yaml - Can now be set to any positive value (32, 64, etc.) - Only minimum validation remains (must be > 0) - Remove dynamic backpressure/throttling from implementation-orchestrator - Dispatch always runs at full configured speed - Resource monitoring remains for visibility only - No automatic reduction of slots_available based on failures - Convert system-watchdog from auto-degradation to monitoring + suggestions - Renamed DEGRADATION_THRESHOLDS to HEALTH_THRESHOLDS - Removed apply_system_degradation() and check_degradation_recovery() - Changed findings to include suggestions instead of actions - Watchdog now reports issues with fix recommendations - No automatic throttling or pausing of agents The system now operates at maximum configured speed at all times, with the watchdog providing diagnostic insights when issues arise.
12 KiB
12 KiB
Session State Persistence Patterns
This document defines standardized patterns for persisting agent session state to enable crash recovery and work resumption.
Core Principles
- Persist via Forgejo - All state lives in Forgejo comments, not local files
- Checkpoint frequently - After each major operation
- Versioned format - Enable future migration
- Human readable - Comments should be understandable
- 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-orchestrator", 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-orchestrator", 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-self-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)