File size: 1,396 Bytes
a952fa2 febe155 a952fa2 febe155 a952fa2 febe155 a952fa2 | 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 | """Task definition base class for CORP-ENV."""
from __future__ import annotations
from abc import ABC, abstractmethod
from typing import Any, Dict, List, Literal, Optional
from server.agents.personas import AgentSlot
Difficulty = Literal["easy", "medium", "hard"]
MasterTier = Literal["fresher", "senior", "executive"]
class CorpTask(ABC):
task_id: str
description: str
role: str
difficulty: Difficulty
master_tier: MasterTier
agent_slots: List[AgentSlot]
token_budget: int
intel_injections: Dict[str, str]
def __init__(self) -> None:
self.intel_injections = {}
@property
def available_agents(self) -> List[str]:
"""Derived slot-id list shown to the master agent."""
return [s.id for s in self.agent_slots]
def get_slot(self, slot_id: str) -> Optional[AgentSlot]:
for s in self.agent_slots:
if s.id == slot_id:
return s
return None
@abstractmethod
def initial_swd(self, episode_id: str) -> Dict[str, Any]:
raise NotImplementedError
@abstractmethod
def verifier(self, swd: Dict[str, Any]) -> Dict[str, bool]:
raise NotImplementedError
@abstractmethod
def milestone_complete(self, swd: Dict[str, Any], milestone_id: str) -> bool:
"""Return True when milestone objective is satisfied."""
raise NotImplementedError
|