feat(gns2): mass-update all 30 agents with GNS-2 protocol
- 29 agents updated with GNS-2 checkpoint/event protocol - 12 Tier 0 (leaf) agents: read checkpoint, write event footer, no cascade - 17 Tier 1 (task) agents: read checkpoint, recommend next agent, no direct task calls - 2 Tier 2 (meta) agents already updated: capability-analyst, agent-architect, evaluator - All agents now include GNS_EVENT footer template in comments - Frontmatter updated with '(GNS-2 Tier N)' classification Scripts added: - scripts/mass-update-gns-agents.py — idempotent mass updater - scripts/validate-gns-agents.py — protocol checker Refs: Milestone #67, Issues #99-#107
This commit is contained in:
NW
246
scripts/mass-update-gns-agents.py
Normal file
246
scripts/mass-update-gns-agents.py
Normal file
@@ -0,0 +1,246 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
GNS-2 Agent Mass Update Script
|
||||
Updates all remaining Tier 0/1 agents with GNS-2 protocol:
|
||||
- Checkpoint read requirement (read-only for Tier 0)
|
||||
- Event footer template (mandatory)
|
||||
- Tier classification (Tier 0 or 1)
|
||||
"""
|
||||
import os
|
||||
import re
|
||||
import glob
|
||||
|
||||
# Root directory of agents
|
||||
AGENTS_DIR = '.kilo/agents'
|
||||
|
||||
# Tier classification
|
||||
TIER_0_AGENTS = [
|
||||
'history-miner', 'code-skeptic', 'performance-engineer',
|
||||
'security-auditor', 'visual-tester', 'browser-automation',
|
||||
'markdown-validator', 'planner', 'reflector', 'memory-manager',
|
||||
'pipeline-judge', 'architect-indexer'
|
||||
]
|
||||
|
||||
TIER_1_AGENTS = [
|
||||
'lead-developer', 'the-fixer', 'sdet-engineer',
|
||||
'frontend-developer', 'backend-developer', 'go-developer',
|
||||
'flutter-developer', 'php-developer', 'python-developer',
|
||||
'devops-engineer', 'release-manager', 'requirement-refiner',
|
||||
'product-owner', 'prompt-optimizer', 'system-analyst',
|
||||
'workflow-architect', 'orchestrator'
|
||||
]
|
||||
|
||||
def get_tier(agent_name: str) -> int:
|
||||
if agent_name in TIER_0_AGENTS:
|
||||
return 0
|
||||
if agent_name in TIER_1_AGENTS:
|
||||
return 1
|
||||
return -1 # Unknown
|
||||
|
||||
def extract_frontmatter(content: str) -> tuple:
|
||||
"""Extract YAML frontmatter from markdown content."""
|
||||
if not content.startswith('---'):
|
||||
return None, content
|
||||
|
||||
parts = content.split('---', 2)
|
||||
if len(parts) < 3:
|
||||
return None, content
|
||||
|
||||
return parts[1].strip(), parts[2].strip()
|
||||
|
||||
def update_frontmatter(fm: str, tier: int) -> str:
|
||||
"""Update frontmatter with GNS-2 metadata."""
|
||||
lines = fm.split('\n')
|
||||
new_lines = []
|
||||
|
||||
# Add tier comment
|
||||
new_lines.append(f"# GNS-2 Agent (Tier {tier})")
|
||||
|
||||
for line in lines:
|
||||
# Ensure permission.task exists
|
||||
if line.strip().startswith('permission:'):
|
||||
new_lines.append(line)
|
||||
continue
|
||||
new_lines.append(line)
|
||||
|
||||
return '\n'.join(new_lines)
|
||||
|
||||
def generate_gns_protocol(tier: int) -> str:
|
||||
"""Generate GNS-2 protocol section for an agent."""
|
||||
|
||||
if tier == 0:
|
||||
return """## GNS-2 Protocol
|
||||
|
||||
### Tier
|
||||
Tier 0 (Leaf Agent / No Cascade)
|
||||
- `max_cascade_depth: 0` (no subagent calls)
|
||||
- Read checkpoint only (do not modify)
|
||||
- Write event footer on completion
|
||||
|
||||
### On Entry (MANDATORY)
|
||||
1. Read issue body from Gitea API
|
||||
2. Parse `## GNS Checkpoint` YAML block
|
||||
3. Extract task from checkpoint or last event
|
||||
|
||||
### During Work
|
||||
- Execute atomic task as specified in checkpoint
|
||||
- Follow existing behavior guidelines
|
||||
- Do NOT spawn subagents
|
||||
|
||||
### On Exit (MANDATORY)
|
||||
1. Post comment with result + GNS_EVENT footer
|
||||
2. Do NOT modify checkpoint (read-only)
|
||||
3. Set `next_agent` recommendation in event footer
|
||||
|
||||
### Next Recommendation
|
||||
After completion, recommend next agent in event footer:
|
||||
- `code-skeptic`: after code written
|
||||
- `performance-engineer`: after code tested
|
||||
- `security-auditor`: after performance reviewed
|
||||
"""
|
||||
|
||||
elif tier == 1:
|
||||
return """## GNS-2 Protocol
|
||||
|
||||
### Tier
|
||||
Tier 1 (Task Agent / Orchestrator-Mediated Cascade)
|
||||
- `max_cascade_depth: 1` (request orchestrator to spawn, do not spawn directly)
|
||||
- Can read checkpoint and recommend next agent
|
||||
- Event footer triggers orchestrator polling
|
||||
|
||||
### On Entry (MANDATORY)
|
||||
1. Read issue body from Gitea API
|
||||
2. Parse `## GNS Checkpoint` YAML block
|
||||
3. Verify `checkpoint.budget.remaining > estimated_cost`
|
||||
|
||||
### During Work
|
||||
- Execute task as specified
|
||||
- If subagent needed, write recommendation in event footer
|
||||
- Do NOT call `task` tool directly (Tier 1)
|
||||
|
||||
### On Exit (MANDATORY)
|
||||
1. Update labels if needed (quality::*, phase::*)
|
||||
2. Post comment with result + GNS_EVENT footer
|
||||
3. Include `next_agent` recommendation
|
||||
|
||||
### GNS Event Footer Template
|
||||
```markdown
|
||||
---
|
||||
<!-- GNS_EVENT: {
|
||||
"type": "subagent_result",
|
||||
"agent": "AGENT_NAME",
|
||||
"invocation_id": "AGENT-{issue}-{seq}",
|
||||
"parent_id": "{parent_invocation}",
|
||||
"depth": 1,
|
||||
"budget": {"remaining": {remaining}},
|
||||
"state_changes": {
|
||||
"labels_add": ["phase::{phase}"],
|
||||
"labels_remove": ["phase::{old_phase}"],
|
||||
"assignee": "{next_agent}",
|
||||
"is_locked": false
|
||||
},
|
||||
"next_agent": "{next_agent}",
|
||||
"estimated_next_tokens": {estimate},
|
||||
"timestamp": "{iso8601}"
|
||||
} -->
|
||||
```
|
||||
"""
|
||||
|
||||
return ""
|
||||
|
||||
def update_agent_file(filepath: str) -> bool:
|
||||
"""Update a single agent file with GNS-2 protocol."""
|
||||
|
||||
agent_name = os.path.basename(filepath).replace('.md', '')
|
||||
tier = get_tier(agent_name)
|
||||
|
||||
if tier < 0:
|
||||
print(f"⚠️ Unknown agent: {agent_name}, skipping")
|
||||
return False
|
||||
|
||||
with open(filepath, 'r') as f:
|
||||
content = f.read()
|
||||
|
||||
# Check if already updated
|
||||
if 'GNS-2 Protocol' in content:
|
||||
print(f"⏭️ {agent_name} already has GNS-2 protocol")
|
||||
return False
|
||||
|
||||
fm_raw, body = extract_frontmatter(content)
|
||||
|
||||
if fm_raw is None:
|
||||
print(f"❌ {agent_name}: no frontmatter found")
|
||||
return False
|
||||
|
||||
# Update description to mention GNS-2
|
||||
fm_lines = fm_raw.split('\n')
|
||||
new_fm_lines = []
|
||||
for line in fm_lines:
|
||||
if line.startswith('description:'):
|
||||
desc = line.replace('description:', '').strip()
|
||||
new_fm_lines.append(f'description: {desc} (GNS-2 Tier {tier})')
|
||||
else:
|
||||
new_fm_lines.append(line)
|
||||
|
||||
new_fm = '---\n' + '\n'.join(new_fm_lines) + '\n---'
|
||||
|
||||
# Generate GNS-2 section
|
||||
gns_section = generate_gns_protocol(tier)
|
||||
|
||||
# Combine: frontmatter + original body + GNS section
|
||||
# Insert GNS section before <!-- gitea-commenting -->
|
||||
gitea_pattern = r'<gitea-commenting[^/]*/>'
|
||||
|
||||
if re.search(gitea_pattern, body):
|
||||
# Insert before gitea-commenting tag
|
||||
new_body = re.sub(
|
||||
gitea_pattern,
|
||||
f"{gns_section}\n\n\\g<0>",
|
||||
body
|
||||
)
|
||||
else:
|
||||
# Append at end
|
||||
new_body = body + '\n\n' + gns_section
|
||||
|
||||
new_content = new_fm + '\n' + new_body
|
||||
|
||||
with open(filepath, 'w') as f:
|
||||
f.write(new_content)
|
||||
|
||||
print(f"✅ {agent_name} (Tier {tier})")
|
||||
return True
|
||||
|
||||
def main():
|
||||
print("GNS-2 Agent Mass Update")
|
||||
print(f"Target: {AGENTS_DIR}")
|
||||
print(f"Tier 0 (Leaf): {len(TIER_0_AGENTS)}")
|
||||
print(f"Tier 1 (Task): {len(TIER_1_AGENTS)}")
|
||||
print()
|
||||
|
||||
updated = 0
|
||||
skipped = 0
|
||||
failed = 0
|
||||
|
||||
for filepath in sorted(glob.glob(os.path.join(AGENTS_DIR, '*.md'))):
|
||||
agent_name = os.path.basename(filepath).replace('.md', '')
|
||||
|
||||
# Skip already updated agents
|
||||
if agent_name in ['capability-analyst', 'agent-architect', 'evaluator']:
|
||||
print(f"⏭️ {agent_name} (already GNS-2)")
|
||||
skipped += 1
|
||||
continue
|
||||
|
||||
result = update_agent_file(filepath)
|
||||
if result:
|
||||
updated += 1
|
||||
elif 'already' in f'{result}':
|
||||
skipped += 1
|
||||
else:
|
||||
failed += 1
|
||||
|
||||
print()
|
||||
print(f"Done: {updated} updated, {skipped} skipped, {failed} failed")
|
||||
print(f"Total: {updated + skipped + failed} agents processed")
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
Reference in New Issue
Block a user