alvis
305eeae38b
Each agent now exports a module-level MANIFEST declaring id, version, pref_schema, required_consents, ttl_sec, and silenced_in_contexts. The registry surfaces both the agent and its manifest, and rejects on mismatch so the two cannot drift. ml/serving exposes GET /agents/registry; services/api proxies it as GET /api/agents/registry with a 60s in-process cache so admin pageviews don't hammer upstream. Failures aren't cached. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
71 lines
2.6 KiB
Python
71 lines
2.6 KiB
Python
from __future__ import annotations
|
|
from collections import defaultdict
|
|
from typing import ClassVar
|
|
from .base import BaseAgent, AgentInput, AgentOutput
|
|
from .manifest import AgentManifest
|
|
|
|
|
|
MANIFEST = AgentManifest(
|
|
id="focus-area",
|
|
version="1.0.0",
|
|
description="Identifies the most congested project/area in the user's task list.",
|
|
pref_schema={
|
|
"type": "object",
|
|
"additionalProperties": False,
|
|
"properties": {
|
|
"preferred_areas": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"default": [],
|
|
"description": "Project / label names to prioritise when multiple areas tie.",
|
|
},
|
|
},
|
|
},
|
|
context_schema=["todoist.tasks"],
|
|
required_consents=["data:core", "data:todoist", "agent:focus-area"],
|
|
output_contract={"type": "snippet", "format": "free_text"},
|
|
ttl_sec=43_200,
|
|
)
|
|
|
|
|
|
class FocusAreaAgent(BaseAgent):
|
|
"""Identifies the most congested project/area in the user's task list."""
|
|
agent_id: ClassVar[str] = MANIFEST.id
|
|
ttl_seconds: ClassVar[int] = MANIFEST.ttl_sec
|
|
version: ClassVar[str] = MANIFEST.version
|
|
|
|
def compute(self, inp: AgentInput) -> AgentOutput:
|
|
by_project: dict[str, list[dict]] = defaultdict(list)
|
|
for task in inp.tasks:
|
|
project = task.get("project_id") or task.get("project") or "default"
|
|
by_project[project].append(task)
|
|
|
|
if not by_project:
|
|
prompt = "No tasks available to identify a focus area."
|
|
return self._make_output(inp, prompt, {"project_count": 0})
|
|
|
|
# Score each project: overdue tasks count double
|
|
def score(tasks: list[dict]) -> float:
|
|
return sum(2.0 if t.get("is_overdue") else 1.0 for t in tasks)
|
|
|
|
top_project, top_tasks = max(by_project.items(), key=lambda kv: score(kv[1]))
|
|
overdue_in_top = sum(1 for t in top_tasks if t.get("is_overdue"))
|
|
label = "the default project" if top_project == "default" else f'"{top_project}"'
|
|
n = len(top_tasks)
|
|
|
|
parts = [
|
|
f"The user's most congested area is {label} "
|
|
f"({n} task{'s' if n != 1 else ''}, {overdue_in_top} overdue)."
|
|
]
|
|
if overdue_in_top >= 3:
|
|
parts.append("Consider surfacing an action from this area.")
|
|
|
|
prompt = " ".join(parts)
|
|
snapshot = {
|
|
"top_project": top_project,
|
|
"top_task_count": n,
|
|
"top_overdue_count": overdue_in_top,
|
|
"project_count": len(by_project),
|
|
}
|
|
return self._make_output(inp, prompt, snapshot)
|