Modpol/agentic-govbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Implement platform skill execution system and fix message formatting

Browse Source 
Major features:
- Added platform skill execution architecture allowing agent to invoke
  platform-specific actions (delete posts, update rules, announcements)
- Agent now receives available platform skills and can execute them with
  proper authorization checks via constitution

Agent improvements:
- Added platform_skills parameter to process_request()
- New action type: execute_platform_skill with skill_name and skill_parameters
- Enhanced LLM prompts to distinguish between direct execution and skill execution
- Clearer JSON format specifications for different action types

Bot orchestration:
- Bot fetches platform skills and passes to agent
- Detects execute_platform_skill actions and calls platform.execute_skill()
- Handles skill execution results with proper error reporting

Mastodon platform:
- Implemented create_announcement skill with graceful API limitation handling
- Implemented update_instance_rules skill with graceful API limitation handling
- Both skills now acknowledge Mastodon API doesn't support these operations
  programmatically and provide direct admin interface links
- Implemented delete_status skill (working)
- All admin operations use direct API calls via __api_request

Message formatting fixes:
- Fixed message chunking to preserve line breaks and paragraph structure
- Split by paragraphs first, then lines, then words as last resort
- Removed debug logging for newline tracking

Documentation:
- Updated .gitignore to exclude bot.log and nohup.out

This implements the governance bot's ability to execute platform-specific
actions while respecting constitutional authority and handling API limitations
gracefully.

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
Nathan Schneider
2026-02-09 22:45:31 -07:00
parent 422f0859f4
commit a0785f09cf
5 changed files with 261 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
src/govbot/agent.py
View File

@@ -60,7 +60,8 @@ class GovernanceAgent:
self,
request: str,
actor: str,
context: Optional[Dict[str, Any]] = None
context: Optional[Dict[str, Any]] = None,
platform_skills: Optional[List[Dict[str, Any]]] = None
) -> Dict[str, Any]:
"""
Process a governance request using agentic interpretation.
@@ -78,6 +79,7 @@ class GovernanceAgent:
request: Natural language request
actor: Who made the request
context: Optional context (thread ID, etc.)
platform_skills: List of available platform-specific skills
Returns:
Response dictionary with action taken and audit trail
@@ -103,7 +105,8 @@ class GovernanceAgent:
constitution=constitutional_guidance,
memory=memory_context,
actor=actor,
context=context
context=context,
platform_skills=platform_skills
)
# Step 5: Execute the decision
@@ -158,9 +161,20 @@ Return your analysis as JSON:
try:
result = self.constitution._call_llm(prompt)
# Parse JSON from response
# (In production, would use proper JSON parsing from LLM response)
return json.loads(result.get("answer", "{}"))
# result is a string, parse JSON from it
# Handle potential markdown code blocks
if "```json" in result:
json_start = result.find("```json") + 7
json_end = result.find("```", json_start)
json_str = result[json_start:json_end].strip()
elif "```" in result:
json_start = result.find("```") + 3
json_end = result.find("```", json_start)
json_str = result[json_start:json_end].strip()
else:
json_str = result.strip()
return json.loads(json_str)
except Exception as e:
return {"error": f"Failed to parse intent: {e}"}
@@ -210,7 +224,8 @@ Return your analysis as JSON:
constitution: Dict[str, Any],
memory: Dict[str, Any],
actor: str,
context: Optional[Dict[str, Any]]
context: Optional[Dict[str, Any]],
platform_skills: Optional[List[Dict[str, Any]]] = None
) -> Dict[str, Any]:
"""
Use LLM to decide what action to take.
@@ -221,13 +236,29 @@ Return your analysis as JSON:
Returns:
Decision dict with:
- action: What to do
- reasoning: Why
- reasoning: Why (formatted with line breaks for readability)
- constitution_citations: Which articles apply
- calculations: Any math needed
- state_updates: Changes to memory
"""
# Format platform skills for the prompt
platform_skills_info = ""
if platform_skills:
platform_skills_info = "\n\nAVAILABLE PLATFORM SKILLS:\n"
for skill in platform_skills:
params_str = ", ".join([f"{p['name']}: {p['type']}" for p in skill.get('parameters', [])])
platform_skills_info += f"- {skill['name']}({params_str}): {skill['description']}\n"
if skill.get('constitutional_authorization'):
platform_skills_info += f" Authorization: {skill['constitutional_authorization']}\n"
prompt = f"""You are a governance bot interpreting a community constitution.
IMPORTANT: When generating the "response" field, use newline characters (\\n) for line breaks:
- Put each key point on its own line using \\n
- Use double newlines (\\n\\n) between paragraphs
- Format like this: "First point\\n\\nSecond point\\n\\nThird point"
- Make responses readable and well-formatted with actual newline characters
INTENT:
{json.dumps(intent, indent=2)}
@@ -238,13 +269,28 @@ CURRENT MEMORY STATE:
{json.dumps(memory, indent=2)}
ACTOR: {actor}
{platform_skills_info}
Based on the constitution and current state, decide what action to take.
For proposals:
- What type of proposal is this? (standard, urgent, constitutional, etc.)
- What discussion period does the constitution specify?
- What voting threshold is required?
IMPORTANT AUTHORITY CHECK:
- Does the constitution give the ACTOR direct authority to make this decision?
- If YES and no platform skill needed: Use action "direct_execution"
- If YES and platform skill needed: Use action "execute_platform_skill"
- If NO: Create a process for community decision-making
CRITICAL - For platform-specific actions:
- When the user REQUESTS an action (like "create an announcement", "delete this post", "update rules"),
check if a platform skill exists for that action
- If a skill exists AND the actor has authority: EXECUTE the skill using "execute_platform_skill" action
- Do NOT just tell the user to use the skill - the user is asking YOU to do it for them
- Include skill_name and skill_parameters in your response
- The bot will execute the skill on the platform and report the result
- Only use "direct_execution" if there is NO platform skill for the requested action
For proposals/processes (only if actor lacks authority):
- What type of process is this? (proposal, consensus, discussion, etc.)
- What decision period does the constitution specify?
- What threshold or mechanism is required?
- Are there any special requirements?
For votes:
@@ -266,16 +312,30 @@ Available tools for calculations:
- random_select(items, count): Random selection
Return your decision as JSON:
For execute_platform_skill action:
{{
"action": "create_process|record_vote|complete_process|query_response",
"reasoning": "explain your interpretation",
"action": "execute_platform_skill",
"skill_name": "create_announcement", // REQUIRED - exact name of the skill to execute
"skill_parameters": {{ // REQUIRED - parameters for the skill
"text": "The announcement text here"
}},
"reasoning": "internal reasoning for audit log",
"response": "Confirmation message to user",
"constitution_citations": ["Article X, Section Y", ...]
}}
For other actions:
{{
"action": "direct_execution|create_process|record_vote|complete_process|query_response",
"reasoning": "internal reasoning for audit log (not shown to user)",
"response": "clean, concise public response to the user (use line breaks for readability)",
"constitution_citations": ["Article X, Section Y", ...],
"parameters": {{
// Action-specific parameters
"process_type": "...",
"deadline_days": X,
"threshold_expression": "agree > disagree",
// etc.
"threshold_expression": "agree > disagree"
}},
"calculations": [
{{
@@ -289,7 +349,20 @@ Return your decision as JSON:
try:
result = self.constitution._call_llm(prompt)
decision = json.loads(result.get("answer", "{}"))
# result is a string, parse JSON from it
# Handle potential markdown code blocks
if "```json" in result:
json_start = result.find("```json") + 7
json_end = result.find("```", json_start)
json_str = result[json_start:json_end].strip()
elif "```" in result:
json_start = result.find("```") + 3
json_end = result.find("```", json_start)
json_str = result[json_start:json_end].strip()
else:
json_str = result.strip()
decision = json.loads(json_str)
# Execute any calculations using tools
if "calculations" in decision:
@@ -326,7 +399,28 @@ Return your decision as JSON:
params = decision.get("parameters", {})
try:
if action == "create_process":
if action == "direct_execution":
# Actor has authority - acknowledge and execute
return {
"response": decision.get("response", decision.get("reasoning")),
"reasoning": decision.get("reasoning"), # For audit log
"constitution_citations": decision.get("constitution_citations", []),
"success": True
}
elif action == "execute_platform_skill":
# Actor has authority and needs to execute a platform-specific skill
return {
"action": "execute_platform_skill",
"skill_name": decision.get("skill_name"),
"skill_parameters": decision.get("skill_parameters", {}),
"response": decision.get("response", "Executing platform action..."),
"reasoning": decision.get("reasoning"), # For audit log
"constitution_citations": decision.get("constitution_citations", []),
"success": True
}
elif action == "create_process":
return self._create_process_from_decision(decision, actor, context)
elif action == "record_vote":
@@ -337,7 +431,8 @@ Return your decision as JSON:
elif action == "query_response":
return {
"response": decision.get("reasoning"),
"response": decision.get("response", decision.get("reasoning")),
"reasoning": decision.get("reasoning"), # For audit log
"constitution_citations": decision.get("constitution_citations", []),
"success": True
}

112
src/govbot/bot.py
View File

@@ -197,14 +197,62 @@ class Govbot:
"platform_message": message,
}
# Get available platform skills
platform_skills_list = []
try:
platform_skills = self.platform.get_skills()
# Convert PlatformSkill objects to dicts for the agent
for skill in platform_skills:
skill_dict = {
"name": skill.name,
"description": skill.description,
"category": skill.category,
"parameters": [
{"name": p.name, "type": p.type, "description": p.description}
for p in skill.parameters
],
"constitutional_authorization": skill.constitutional_authorization,
}
platform_skills_list.append(skill_dict)
except Exception as e:
logger.warning(f"Could not get platform skills: {e}")
result = self.agent.process_request(
request=message.text,
actor=f"@{message.author_handle}",
context=context,
platform_skills=platform_skills_list if platform_skills_list else None,
)
# Post response
response = result.get("response", "Sorry, I couldn't process that request.")
# Check if we need to execute a platform skill
if result.get("action") == "execute_platform_skill":
skill_name = result.get("skill_name")
skill_params = result.get("skill_parameters", {})
logger.info(f"Executing platform skill: {skill_name}")
try:
skill_result = self.platform.execute_skill(
skill_name=skill_name,
parameters=skill_params,
actor=f"@{message.author_handle}",
)
if skill_result.get("success"):
# Skill executed successfully
response = result.get("response", "Action completed successfully.")
logger.info(f"Platform skill executed successfully: {skill_result.get('message')}")
else:
# Skill execution failed
response = f"Failed to execute action: {skill_result.get('message', 'Unknown error')}"
logger.error(f"Platform skill execution failed: {skill_result.get('message')}")
except Exception as e:
logger.error(f"Error executing platform skill: {e}", exc_info=True)
response = f"Error executing action: {str(e)}"
else:
# Normal response without skill execution
response = result.get("response", "Sorry, I couldn't process that request.")
# Handle long responses by splitting into thread
try:
@@ -307,24 +355,62 @@ Process ID: {process_id}
)
return
# Split into chunks
words = response.split()
# Split into chunks while preserving newlines
# Split by paragraph breaks first to keep formatting intact
paragraphs = response.split('\n\n')
chunks = []
current_chunk = []
current_length = 0
for word in words:
word_len = len(word) + 1 # +1 for space
if current_length + word_len > max_length and current_chunk:
chunks.append(" ".join(current_chunk))
current_chunk = [word]
current_length = word_len
for para in paragraphs:
para_length = len(para) + 2 # +2 for paragraph break
# If single paragraph is too long, split it by sentences or words
if para_length > max_length:
# Try to split by newlines within paragraph
lines = para.split('\n')
for line in lines:
line_length = len(line) + 1 # +1 for newline
if current_length + line_length > max_length and current_chunk:
# Flush current chunk
chunks.append('\n\n'.join(current_chunk))
current_chunk = [line]
current_length = line_length
elif line_length > max_length:
# Single line too long, split by words
if current_chunk:
chunks.append('\n\n'.join(current_chunk))
current_chunk = []
current_length = 0
words = line.split()
word_chunk = []
word_length = 0
for word in words:
word_len = len(word) + 1
if word_length + word_len > max_length and word_chunk:
chunks.append(' '.join(word_chunk))
word_chunk = [word]
word_length = word_len
else:
word_chunk.append(word)
word_length += word_len
if word_chunk:
current_chunk.append(' '.join(word_chunk))
current_length = len(current_chunk[-1])
else:
current_chunk.append(line)
current_length += line_length
elif current_length + para_length > max_length and current_chunk:
# Flush current chunk
chunks.append('\n\n'.join(current_chunk))
current_chunk = [para]
current_length = para_length
else:
current_chunk.append(word)
current_length += word_len
current_chunk.append(para)
current_length += para_length
if current_chunk:
chunks.append(" ".join(current_chunk))
chunks.append('\n\n'.join(current_chunk))
# Post chunks as a thread
last_id = reply_to_id

61
src/govbot/platforms/mastodon.py
View File

@@ -587,20 +587,27 @@ class MastodonAdapter(PlatformAdapter):
}
def _update_instance_rules(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Update instance rules"""
# Note: This requires admin API access
# Implementation depends on Mastodon version and API availability
rules = params["rules"]
"""Update instance rules - web admin only"""
rules = params.get("rules", [])
# This would use admin API to update instance rules
# Exact implementation varies by Mastodon version
# Note: Mastodon's API does not provide endpoints for creating/updating rules.
# Rules must be managed through the web admin interface.
# See: https://docs.joinmastodon.org/methods/instance/
rules_text = "\n".join([f"- {rule}" for rule in rules])
return {
"success": True,
"message": f"Updated instance rules ({len(rules)} rules)",
"success": False,
"message": (
f"Proposed server rules:\n{rules_text}\n\n"
"Note: Mastodon's API does not support managing server rules programmatically. "
f"To update the server rules, please visit:\n"
f"{self.instance_url}/admin/server_settings/rules\n\n"
"You can add, edit, or remove rules through the admin interface."
),
"data": {"rules": rules},
"reversible": True,
"reverse_params": {"rules": "previous_rules"}, # Would need to store previous
"reversible": False,
"requires_manual_action": True,
}
def _update_instance_description(self, params: Dict[str, Any]) -> Dict[str, Any]:
@@ -649,23 +656,35 @@ class MastodonAdapter(PlatformAdapter):
}
def _create_announcement(self, params: Dict[str, Any]) -> Dict[str, Any]:
"""Create instance announcement"""
"""Create instance announcement - web admin only"""
text = params["text"]
starts_at = params.get("starts_at")
ends_at = params.get("ends_at")
announcement = self.client.admin_announcement_create(
text=text,
starts_at=starts_at,
ends_at=ends_at,
)
# Note: Mastodon's API does not provide endpoints for creating announcements.
# Announcements must be created through the web admin interface.
# See: https://docs.joinmastodon.org/methods/announcements/
timing_info = ""
if starts_at or ends_at:
timing_info = "\n\nTiming:"
if starts_at:
timing_info += f"\n- Start: {starts_at}"
if ends_at:
timing_info += f"\n- End: {ends_at}"
return {
"success": True,
"message": "Created announcement",
"data": {"announcement_id": announcement["id"], "text": text},
"reversible": True,
"reverse_params": {"announcement_id": announcement["id"]},
"success": False,
"message": (
f"Announcement text:\n{text}{timing_info}\n\n"
"Note: Mastodon's API does not support creating announcements programmatically. "
f"To post this announcement, please visit:\n"
f"{self.instance_url}/admin/announcements\n\n"
"Click 'New announcement', paste the text above, and publish it."
),
"data": {"text": text, "starts_at": starts_at, "ends_at": ends_at},
"reversible": False,
"requires_manual_action": True,
}
def _strip_markdown(self, text: str) -> str:

13
src/govbot/scheduler.py
View File

@@ -100,12 +100,13 @@ class GovernanceScheduler:
)
# TODO: Post result to Mastodon
# Check for pending reminders
reminders = self.agent.primitives.get_pending_reminders()
for reminder in reminders:
logger.info(f"Sending reminder: {reminder['data']['message']}")
# TODO: Post reminder to Mastodon
self.agent.primitives.mark_reminder_sent(reminder["id"])
# Check for pending reminders (if agent has primitives - old architecture)
if hasattr(self.agent, 'primitives'):
reminders = self.agent.primitives.get_pending_reminders()
for reminder in reminders:
logger.info(f"Sending reminder: {reminder['data']['message']}")
# TODO: Post reminder to Mastodon
self.agent.primitives.mark_reminder_sent(reminder["id"])
# Check for veto votes (every cycle)
self._check_veto_votes()