Spaces:
Running
Running
File size: 6,752 Bytes
9af40c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | from __future__ import annotations
import uuid
from agent import AgentContext, AgentContextType, UserMessage
from helpers import subagents
from helpers.api import ApiHandler, Request, Response
from initialize import initialize_agent
from plugins._goal.tools import goal
class Delegate(ApiHandler):
"""API-key protected structured goal delegation for Spynel."""
@classmethod
def requires_auth(cls) -> bool:
return False
@classmethod
def requires_csrf(cls) -> bool:
return False
@classmethod
def requires_api_key(cls) -> bool:
return True
async def process(self, input: dict, request: Request) -> dict | Response:
action = str(input.get("action") or "create").strip().lower()
if action != "create":
return self._operate(action, input)
profile = str(input.get("agent_profile") or "").strip()
objective = str(input.get("objective") or input.get("message") or "").strip()
idempotency_key = str(input.get("idempotency_key") or "").strip()
idempotency_scope = str(input.get("idempotency_scope") or "").strip()
project_name = input.get("project_name")
if not profile or profile not in subagents.get_available_agents_dict(
project_name
):
return Response(
'{"error":"Agent profile not found"}',
status=404,
mimetype="application/json",
)
if not objective:
return Response(
'{"error":"Goal objective is required"}',
status=400,
mimetype="application/json",
)
if idempotency_key:
existing = goal._find_idempotent(idempotency_key, idempotency_scope)
if existing:
return {
"ok": True,
"accepted": False,
"idempotent_replay": True,
"context_id": existing["context_id"],
"goal_id": existing["goal_id"],
"goal": goal.public_goal(existing),
}
context = AgentContext(
config=initialize_agent(override_settings={"agent_profile": profile}),
type=AgentContextType.USER,
)
default_level = (
"bounded"
if profile in {"reviewer", "security", "docs"}
else "operational"
if profile == "shipper"
else "implementation"
)
current_goal = goal.create_goal(
context.id,
objective,
created_by="spynel",
owner_profile=profile,
title=str(input.get("title") or ""),
parent_goal_id=input.get("parent_goal_id"),
success_criteria=input.get("success_criteria"),
constraints=input.get("constraints"),
evidence_required=input.get("evidence_required"),
autonomy=input.get("autonomy") or {"level": default_level},
depends_on=input.get("depends_on"),
idempotency_key=idempotency_key,
idempotency_scope=idempotency_scope,
)
should_start = goal.dependencies_ready(current_goal)
if not should_start:
current_goal = goal.update_goal(context.id, status="pending")
else:
message_id = str(uuid.uuid4())
context.log.log(
type="user", heading="", content=objective, kvps={}, id=message_id
)
context.communicate(
UserMessage(message=objective, attachments=[], id=message_id)
)
return {
"ok": True,
"accepted": True,
"context_id": context.id,
"goal_id": current_goal["goal_id"],
"state": current_goal["status"],
"goal": goal.public_goal(current_goal),
}
def _operate(self, action: str, input: dict) -> dict | Response:
context_id = str(input.get("context_id") or "").strip()
current = goal.get_goal(context_id) if context_id else None
if not current or (
input.get("goal_id") and input.get("goal_id") != current["goal_id"]
):
return Response(
'{"error":"Goal not found"}', status=404, mimetype="application/json"
)
if action in {"get", "status"}:
updated = current
elif action == "checkpoint":
updated = goal.checkpoint_goal(
context_id,
milestone=input.get("milestone"),
note=str(input.get("note") or ""),
evidence=input.get("evidence"),
criterion_updates=input.get("criterion_updates"),
requires_attention=input.get("requires_attention"),
attention_reason=input.get("attention_reason"),
attention_message=input.get("attention_message"),
)
elif action == "revise":
updated = goal.revise_goal(
context_id,
objective=input.get("objective"),
title=input.get("title"),
success_criteria=input.get("success_criteria"),
constraints=input.get("constraints"),
evidence_required=input.get("evidence_required"),
autonomy=input.get("autonomy"),
)
elif action in {"pause", "resume", "cancel"}:
updated = (
goal.resume_goal(context_id)
if action == "resume"
else goal.update_goal(
context_id,
status={"pause": "paused", "cancel": "cancelled"}[action],
)
)
context = AgentContext.get(context_id)
if context is not None and action != "resume":
context.paused = action != "resume"
elif action == "complete":
updated = goal.complete_goal(
context_id, note=input.get("note"), result=input.get("result")
)
elif action == "children":
return {
"ok": True,
"goal_id": current["goal_id"],
"children": [
goal.public_goal(item)
for item in goal.list_child_goals(current["goal_id"])
],
}
else:
return Response(
'{"error":"Unknown goal action"}',
status=400,
mimetype="application/json",
)
return {
"ok": True,
"context_id": updated["context_id"],
"goal_id": updated["goal_id"],
"goal": goal.public_goal(updated),
}
|