Spaces:
Running
Running
File size: 4,701 Bytes
6918275 | 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 | from __future__ import annotations
from helpers.api import ApiHandler, Request, Response
from plugins._goal.tools import goal
class Goal(ApiHandler):
async def process(self, input: dict, request: Request) -> dict | Response:
action = str(input.get("action", "") or "").strip().lower()
context_id = str(input.get("context_id", "") or "").strip()
try:
if action == "get":
return {"ok": True, "goal": goal.public_goal(goal.get_goal(context_id))}
if action in {"set", "create"}:
return self._set(context_id, input)
if action == "update":
return self._update(context_id, input)
if action == "revise":
current_goal = 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"),
)
return {"ok": True, "goal": goal.public_goal(current_goal)}
if action == "checkpoint":
current_goal = 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"),
)
return {"ok": True, "goal": goal.public_goal(current_goal)}
if action == "pause":
return self._status(context_id, "paused")
if action == "resume":
current_goal = goal.resume_goal(context_id)
return {"ok": True, "goal": goal.public_goal(current_goal)}
if action == "complete":
current_goal = goal.complete_goal(
context_id, note=input.get("note"), result=input.get("result")
)
return {"ok": True, "goal": goal.public_goal(current_goal)}
if action == "cancel":
return self._status(context_id, "cancelled")
if action == "delete":
goal.delete_goal(context_id)
return {"ok": True, "goal": None}
except FileNotFoundError:
return Response(status=404, response="Goal not found")
except ValueError as error:
return Response(status=400, response=str(error))
return Response(status=400, response=f"Unknown action: {action}")
def _set(self, context_id: str, input: dict) -> dict:
current_goal = goal.create_goal(
context_id,
str(input.get("objective") or ""),
created_by=str(input.get("created_by") or "user"),
token_budget=input.get("token_budget"),
title=str(input.get("title") or ""),
owner_profile=str(input.get("owner_profile") or ""),
success_criteria=input.get("success_criteria"),
constraints=input.get("constraints"),
evidence_required=input.get("evidence_required"),
autonomy=input.get("autonomy"),
idempotency_key=str(input.get("idempotency_key") or ""),
)
return {"ok": True, "goal": goal.public_goal(current_goal)}
def _update(self, context_id: str, input: dict) -> dict:
current = goal.get_goal(context_id)
updated_goal = goal.update_goal(
context_id,
objective=input.get("objective") if "objective" in input else None,
status=input.get("status") if "status" in input else None,
note=input.get("note") if "note" in input else None,
token_budget=input.get("token_budget") if "token_budget" in input else None,
)
return {
"ok": True,
"goal": goal.public_goal(updated_goal),
"reactivated": (
current is not None
and current.get("status") in goal.FINAL_STATUSES | {"blocked"}
and updated_goal.get("status") == "active"
),
}
def _status(self, context_id: str, status: str) -> dict:
current_goal = goal.update_goal(context_id, status=status)
return {"ok": True, "goal": goal.public_goal(current_goal)}
|