TDF — Task Definition Framework

# tdf_schema.py — TDF core data models
from pydantic import BaseModel, Field
from typing import Any, Literal, Optional, Union
from datetime import datetime
import uuid

class InputField(BaseModel):
    name: str
    type: str           # "string", "number", "boolean", "object", "array", "file"
    description: str
    required: bool = True
    default: Optional[Any] = None
    examples: list[Any] = []

class OutputField(BaseModel):
    name: str
    type: str
    description: str
    format: Optional[str] = None   # "markdown", "json", "csv", "html"
    min_length: Optional[int] = None
    max_length: Optional[int] = None
    schema_ref: Optional[str] = None  # JSON Schema URL for complex types

class Constraints(BaseModel):
    max_duration_seconds: Optional[int] = None
    max_cost_usd: Optional[float] = None
    max_tokens: Optional[int] = None
    allowed_tools: Optional[list[str]] = None   # tool allowlist
    forbidden_domains: Optional[list[str]] = None  # web restrictions
    requires_human_review: bool = False
    retry_policy: dict = Field(default_factory=lambda: {
        "max_attempts": 3,
        "backoff": "exponential"
    })

class VerificationRule(BaseModel):
    type: Literal["contains", "regex", "json_schema", "llm_judge", "human", "code_test"]
    target_field: str  # which output field to verify
    rule: str          # the check expression
    threshold: Optional[float] = None  # for llm_judge: min score 0-1

class VerificationSpec(BaseModel):
    strategy: Literal["all", "any", "majority"] = "all"
    rules: list[VerificationRule]
    on_failure: Literal["retry", "escalate", "fail"] = "retry"

class TDFTask(BaseModel):
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    version: str = "1.0"
    name: str
    description: str
    category: str   # "research", "coding", "analysis", "writing", "data"
    input: list[InputField]
    output: list[OutputField]
    constraints: Constraints = Field(default_factory=Constraints)
    verification: VerificationSpec
    metadata: dict = Field(default_factory=dict)
    created_at: datetime = Field(default_factory=datetime.utcnow)

class TaskRun(BaseModel):
    """A specific execution of a TDF task with bound input values."""
    id: str = Field(default_factory=lambda: str(uuid.uuid4()))
    task_id: str
    agent_id: str
    input_values: dict[str, Any]
    status: Literal["pending", "running", "completed", "failed", "verified"] = "pending"
    output_values: Optional[dict[str, Any]] = None
    verification_result: Optional[dict] = None
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    cost_usd: Optional[float] = None
    tokens_used: Optional[int] = None

# tdf_tasks.py — Real task definitions for common agentic workflows
from tdf_schema import TDFTask, InputField, OutputField, Constraints, VerificationSpec, VerificationRule

# -------------------------------------------------------
# TASK 1: Research Report
# -------------------------------------------------------
research_task = TDFTask(
    name="Research Report",
    description="Search the web and write a structured research report on a given topic",
    category="research",
    input=[
        InputField(
            name="topic",
            type="string",
            description="The topic to research",
            required=True,
            examples=["quantum computing", "LangGraph vs CrewAI comparison"]
        ),
        InputField(
            name="depth",
            type="string",
            description="Research depth: 'quick' (3 sources) or 'deep' (10+ sources)",
            required=False,
            default="quick"
        ),
        InputField(
            name="output_format",
            type="string",
            description="Output format: 'markdown' or 'json'",
            required=False,
            default="markdown"
        )
    ],
    output=[
        OutputField(
            name="report",
            type="string",
            description="The research report",
            format="markdown",
            min_length=500,
            max_length=5000
        ),
        OutputField(
            name="sources",
            type="array",
            description="List of cited sources",
            min_length=3
        ),
        OutputField(
            name="summary",
            type="string",
            description="2-3 sentence executive summary",
            max_length=300
        )
    ],
    constraints=Constraints(
        max_duration_seconds=120,
        max_cost_usd=0.50,
        max_tokens=4096,
        allowed_tools=["web_search", "fetch_webpage"],
        requires_human_review=False
    ),
    verification=VerificationSpec(
        strategy="all",
        rules=[
            VerificationRule(
                type="contains",
                target_field="sources",
                rule="len(value) >= 3",
            ),
            VerificationRule(
                type="regex",
                target_field="report",
                rule=r"#{1,3} ",  # must have at least one heading
            ),
            VerificationRule(
                type="llm_judge",
                target_field="report",
                rule="Is this report comprehensive, well-structured, and accurate?",
                threshold=0.7
            )
        ],
        on_failure="retry"
    ),
    metadata={"tags": ["research", "web"], "priority": "normal"}
)

# -------------------------------------------------------
# TASK 2: Code Review
# -------------------------------------------------------
code_review_task = TDFTask(
    name="Code Review",
    description="Review a code snippet for bugs, security issues, and style",
    category="coding",
    input=[
        InputField(name="code", type="string", description="The code to review", required=True),
        InputField(name="language", type="string", description="Programming language", required=True),
        InputField(name="focus", type="array",
                   description="Areas to focus on: ['security', 'performance', 'style', 'bugs']",
                   required=False, default=["bugs", "security"])
    ],
    output=[
        OutputField(name="issues", type="array",
                    description="List of issues found, each with: line, severity, description, suggestion"),
        OutputField(name="score", type="number",
                    description="Overall code quality score 0-10"),
        OutputField(name="summary", type="string",
                    description="Brief summary of overall code quality", max_length=200)
    ],
    constraints=Constraints(
        max_duration_seconds=60,
        max_cost_usd=0.25,
        requires_human_review=False
    ),
    verification=VerificationSpec(
        strategy="all",
        rules=[
            VerificationRule(type="json_schema", target_field="issues",
                             rule='{"type":"array","items":{"required":["severity","description"]}}'),
            VerificationRule(type="contains", target_field="score",
                             rule="0 <= value <= 10")
        ]
    ),
    metadata={"tags": ["code", "review", "security"]}
)

# tdf_executor.py — Execute and verify TDF tasks
from anthropic import Anthropic
from tdf_schema import TDFTask, TaskRun, VerificationRule
from datetime import datetime
import json, re

client = Anthropic()

class TDFExecutor:
    def __init__(self, tools: dict = None):
        self.tools = tools or {}

    def build_system_prompt(self, task: TDFTask) -> str:
        """Convert a TDF task definition into a system prompt."""
        output_specs = "
".join(
            f"- {o.name} ({o.type}): {o.description}"
            + (f" [format: {o.format}]" if o.format else "")
            + (f" [max {o.max_length} chars]" if o.max_length else "")
            for o in task.output
        )

        return f"""You are executing the following task: {task.name}

Task description: {task.description}

You MUST produce ALL of these outputs:
{output_specs}

Constraints:
- Max tokens: {task.constraints.max_tokens or 'no limit'}
- Allowed tools: {', '.join(task.constraints.allowed_tools or ['any'])}

Return your response as a JSON object with keys matching the output field names exactly."""

    def build_user_prompt(self, task: TDFTask, input_values: dict) -> str:
        """Build the user prompt with bound input values."""
        lines = ["Execute this task with the following inputs:
"]
        for field in task.input:
            value = input_values.get(field.name, field.default)
            lines.append(f"{field.name}: {json.dumps(value)}")
        return "
".join(lines)

    async def execute(self, task: TDFTask, input_values: dict,
                      agent_id: str = "default") -> TaskRun:
        """Execute a TDF task and return a TaskRun with results."""
        run = TaskRun(
            task_id=task.id,
            agent_id=agent_id,
            input_values=input_values,
            status="running",
            started_at=datetime.utcnow()
        )

        try:
            response = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=task.constraints.max_tokens or 4096,
                system=self.build_system_prompt(task),
                messages=[{
                    "role": "user",
                    "content": self.build_user_prompt(task, input_values)
                }]
            )

            # Parse JSON output
            text = response.content[0].text
            # Extract JSON from markdown code block if present
            json_match = re.search('```json\s*([\s\S]+?)\s*```', text)
            if json_match:
                text = json_match.group(1)

            run.output_values = json.loads(text)
            run.tokens_used = response.usage.input_tokens + response.usage.output_tokens
            run.completed_at = datetime.utcnow()
            run.status = "completed"

        except Exception as e:
            run.status = "failed"
            run.output_values = {"error": str(e)}

        return run

    def verify(self, task: TDFTask, run: TaskRun) -> dict:
        """Run verification rules against the task output."""
        if run.status != "completed" or not run.output_values:
            return {"passed": False, "reason": "Task did not complete"}

        results = []
        for rule in task.verification.rules:
            field_value = run.output_values.get(rule.target_field)
            passed, reason = self._check_rule(rule, field_value)
            results.append({"rule": rule.type, "field": rule.target_field,
                            "passed": passed, "reason": reason})

        strategy = task.verification.strategy
        all_passed = all(r["passed"] for r in results)
        any_passed = any(r["passed"] for r in results)
        majority_passed = sum(1 for r in results if r["passed"]) > len(results) / 2

        final = {"all": all_passed, "any": any_passed, "majority": majority_passed}[strategy]

        return {"passed": final, "strategy": strategy, "rule_results": results}

    def _check_rule(self, rule: VerificationRule, value) -> tuple[bool, str]:
        if rule.type == "contains":
            try:
                result = eval(rule.rule, {"value": value, "len": len})
                return bool(result), "eval passed" if result else "eval failed"
            except Exception as e:
                return False, str(e)
        elif rule.type == "regex":
            if not isinstance(value, str):
                return False, "value is not a string"
            match = re.search(rule.rule, value)
            return bool(match), "pattern found" if match else "pattern not found"
        elif rule.type == "llm_judge":
            # Use Claude to judge quality
            resp = client.messages.create(
                model="claude-haiku-4-5-20251001",
                max_tokens=100,
                messages=[{
                    "role": "user",
                    "content": f"{rule.rule}

Content to evaluate:
{str(value)[:2000]}

Respond with a score from 0.0 to 1.0 only."
                }]
            )
            try:
                score = float(resp.content[0].text.strip())
                threshold = rule.threshold or 0.7
                return score >= threshold, f"LLM score: {score:.2f} (threshold: {threshold})"
            except ValueError:
                return False, "Could not parse LLM judge score"
        return True, "rule type not implemented — defaulting to pass"

# Usage example
import asyncio

async def main():
    executor = TDFExecutor()

    # Execute the research task
    run = await executor.execute(
        task=research_task,
        input_values={"topic": "Task Definition Framework for AI agents", "depth": "quick"},
        agent_id="claude-agent-1"
    )

    print(f"Task status: {run.status}")
    print(f"Tokens used: {run.tokens_used}")

    # Verify the output
    verification = executor.verify(research_task, run)
    print(f"Verification passed: {verification['passed']}")
    for result in verification["rule_results"]:
        icon = "PASS" if result["passed"] else "FAIL"
        print(f"  [{icon}] {result['rule']} on {result['field']}: {result['reason']}")

asyncio.run(main())