za-ek/agents
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

IssueTriage can now interact with task in YouTrack and it is idempotent

Browse Source
This commit is contained in:
za-ek
2026-09-20 20:56:36 +03:00
parent 74577cb50e
commit f25d9a5fab
34 changed files with 1834 additions and 201 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.idea/Agents.iml
Generated
+2
View File
@@ -2,6 +2,8 @@
<module type="PYTHON_MODULE" version="4">
<component name="NewModuleRootManager">
<content url="file://$MODULE_DIR$">
<sourceFolder url="file://$MODULE_DIR$" isTestSource="false" />
<sourceFolder url="file://$MODULE_DIR$/agents/issue_triage/tests" isTestSource="true" />
<excludeFolder url="file://$MODULE_DIR$/.venv" />
</content>
<orderEntry type="jdk" jdkName="Python 3.14 (Agents)" jdkType="Python SDK" />
CHANGELOG.md
+10 -1
View File
@@ -1,3 +1,12 @@
### 2026-09-20
1. Each agent has a *Model* for both input and output
2. Implemented base class for Model that provide abstract render_dbg function (allows print debug info in console) and to_prompt_text that allows to summarize a model for prompting
3. Model was renamed to Contract as it best represent the future usage
4. IssueReader renamed to IssueTriage since it create a step sequence for solution
5. IssueTriage can now post comments to clarify requirements and ask for missing agents skills
### 2026-09-19
1. Created an agent that can accept YouTrack™ task ID, read an issue, create a context for a task (body, comments, image attachments) and decompose it for implementing
2. **[missing]** extract a decomposing because it's manner depends on the task context (i.e. backend-dev, frontend-dev, seller, marketing, ad, design, ...) so an issue_reader agent should listen for not delivered tasks, create a context and delegate a decomposition for specific agent. If there is no agent can create this work - add comment that it cannot be resolved until appropriate agent will be created, assign a task to admin and change status to "To Be Discussed". There should be an agent that has information about all the rest of agents and can provide the information, who can do that task and how (in case of confusing - ask another agents if they can do the task). As this architecture is a very future automatization, it is enough to have just a registry of agents and omit the conversation of issue reader with another agents.
2. **[missing]** extract a decomposing because it's manner depends on the task context (i.e. backend-dev, frontend-dev, seller, marketing, ad, design, ...) so an issue_reader agent should listen for not delivered tasks, create a context and delegate a decomposition for specific agent. If there is no agent can create this work - add comment that it cannot be resolved until appropriate agent will be created, assign a task to admin and change status to "To Be Discussed". There should be an agent that has information about all the rest of agents and can provide the information, who can do that task and how (in case of confusing - ask another agents if they can do the task). As this architecture is a very future automatization, it is enough to have just a registry of agents and omit the conversation of issue reader with another agents. **[adopted restrictions]** agents work in context of single project which is zArch (issues ARCH-*) so:
1. project is defined, .yaml exists
2. agent list is known agents can ask each other
analyzer.py → agents/architect_ts/analyzer.py
View File
agents/base_agent.py
+6
View File
@@ -0,0 +1,6 @@
from abc import ABC, abstractmethod
class BaseAgent(ABC):
@abstractmethod
def run(self, issue_id: str):
pass # To be implemented by subclasses
start.py → agents/codebase_analyst/start.py
+2 -5
View File
@@ -7,17 +7,14 @@ Multi-Agent Development Workflow
Поддерживает несколько проектов, состояние хранится в папке проекта.
"""
import json
import tiktoken
import os
import pathlib
import re
import sys
import time
from pathlib import Path
from typing import List, Dict, Any, Optional
from typing import List, Dict, Any
import requests
from playwright.sync_api import sync_playwright
from analyzer import CodeAnalyzer
from agents.architect_ts.analyzer import CodeAnalyzer
# ---------- Конфигурация ----------
CONFIG_FILE = "config.json"
agents/CoderAgent.py → agents/coder/CoderAgent.py
View File
agents/issue_reader/IssueReaderAgent.py
-44
View File
@@ -1,44 +0,0 @@
"""
Issue Decomposing Agent
This agent specializes in development tasks including:
- Web systems architecture
"""
from typing import Dict, Any
import json
from agents.issue_reader.models import TaskPlan
from agents.issue_reader.decomposer import IssueDecomposer
from agents.issue_reader.validator import enforce
from common.youtrack_mcp_client import IssueNotFound
class IssueReaderAgent:
def __init__(self, context_builder, llm):
self.name = "Issue decomposing agent"
self.ctx_builder = context_builder
self.decomposer = IssueDecomposer(llm)
async def plan_issue(self, issue_id: str) -> TaskPlan:
ctx = await self.ctx_builder.build(issue_id)
if ctx is None:
raise IssueNotFound(f"Issue {issue_id} was not found in YouTrack")
plan = self.decomposer.decompose(ctx)
plan.issue_id = ctx.issue_id
return enforce(plan)
def render(self, plan) -> str:
lines = [f"# Plan for {plan.issue_id}", plan.summary, ""]
for t in plan.tasks:
deps = f" (after {', '.join(t.depends_on)})" if t.depends_on else ""
lines.append(f"- [{t.kind.value}] {t.id}: {t.title}{deps}")
for ac in t.acceptance_criteria:
lines.append(f" ✓ {ac}")
for q in t.open_questions:
lines.append(f" ? {q}")
if plan.unknowns:
lines.append("\nUnknowns:")
lines += [f" - {u}" for u in plan.unknowns]
return "\n".join(lines)
agents/issue_reader/models.py
-128
View File
@@ -1,128 +0,0 @@
# models.py
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Optional
import uuid
class TaskKind(str, Enum):
INVESTIGATE = "investigate" # read code, reproduce, research
DESIGN = "design" # API/schema decisions
IMPLEMENT = "implement" # write code
MIGRATE = "migrate" # DB / data changes
TEST = "test" # unit/integration/e2e
DOCS = "docs" # docs, changelog
REVIEW = "review" # code review, verification
OPS = "ops" # deploy, config, feature flag
class Atomicity(str, Enum):
ATOMIC = "atomic" # single clear action, one owner, < 1 day
NEEDS_SPLIT = "needs_split" # still too big, recurse
@dataclass
class IssueContext:
issue_id: str
title: str
body: str
comments: list[dict] # [{"author": ..., "body": ..., "created_at": ...}]
labels: list[str] = field(default_factory=list)
repo: Optional[str] = None
metadata: dict = field(default_factory=dict)
images: list[str] = field(default_factory=list)
project: ProjectContext | None = None
def to_prompt_text(self) -> str:
parts = [f"# Issue {self.issue_id}: {self.title}", "", self.body]
if self.project:
parts.append("\n## Project info")
parts.append(self.project.to_prompt_text())
meta_lines = []
for key in ("state", "url"):
if self.metadata.get(key):
meta_lines.append(f"- {key.capitalize()}: {self.metadata[key]}")
if meta_lines:
parts.append("\n## Issue Metadata")
parts.extend(meta_lines)
if self.comments:
parts.append("\n## Discussion / Comments")
for c in self.comments:
parts.append(f"- @{c['author']} ({c['created_at']}): {c['body']}")
return "\n".join(parts)
@dataclass
class Task:
id: str
title: str
kind: TaskKind
description: str
acceptance_criteria: list[str] = field(default_factory=list)
depends_on: list[str] = field(default_factory=list)
atomicity: Atomicity = Atomicity.ATOMIC
estimate_hours: Optional[float] = None
files_hint: list[str] = field(default_factory=list) # probable touch points
open_questions: list[str] = field(default_factory=list)
@staticmethod
def new(**kw) -> "Task":
return Task(id=str(uuid.uuid4())[:8], **kw)
@dataclass
class TaskPlan:
issue_id: str
summary: str
tasks: list[Task]
unknowns: list[str] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
@dataclass
class ProjectContext:
"""Static, per-project knowledge injected into every decomposition."""
project_key: str # e.g. "ARCH"
language: str = ""
backend_framework: str = "" # "FastAPI", "Django", "Spring Boot"
frontend_framework: str = "" # "React + Vite", "Vue 3"
ui_library: str = "" # "shadcn/ui", "MUI", "Ant Design"
database: str = "" # "PostgreSQL 16 via SQLAlchemy 2"
orm: str = ""
test_framework: str = "" # "pytest + httpx.AsyncClient"
package_manager: str = "" # "uv", "poetry", "npm"
auth: str = "" # "JWT via fastapi-users"
deployment: str = "" # "Docker Compose on Hetzner"
conventions: list[str] = field(default_factory=list) # free-form notes
extra: dict[str, str] = field(default_factory=dict) # anything else
def to_prompt_text(self) -> str:
lines = [f"## Project Context ({self.project_key})"]
fields = [
("Language", self.language),
("Backend framework", self.backend_framework),
("Frontend framework", self.frontend_framework),
("UI library", self.ui_library),
("Database", self.database),
("ORM", self.orm),
("Test framework", self.test_framework),
("Package manager", self.package_manager),
("Auth", self.auth),
("Deployment", self.deployment),
]
for label, value in fields:
if value:
lines.append(f"- {label}: {value}")
for k, v in self.extra.items():
if v:
lines.append(f"- {k}: {v}")
if self.conventions:
lines.append("")
lines.append("Conventions:")
lines.extend(f"- {c}" for c in self.conventions)
return "\n".join(lines)
agents/issue_triage/IssueTriageAgent.py
+627
View File
@@ -0,0 +1,627 @@
import json
import logging
from dataclasses import dataclass
from agents.issue_triage.context_builder import YouTrackContextBuilder
from agents.issue_triage.decomposer import IssueDecomposer
from agents.issue_triage.marker import TriageMarker, find_prior_marker, has_reporter_reply_since
from agents.issue_triage.merge import Action, merge
from agents.issue_triage.models import CapabilityResult, AmbiguityResult, parse_capability_result, \
parse_ambiguity_result, AmbiguityFinding, Plan, parse_task_plan
from agents.issue_triage.triage import render_validation_failure, render_plan_comment, validate_plan, \
render_questions_comment, render_blocker_comment
from agents.registry import agent, AgentRegistry, Capability
from common.llm_client import LLMClient
from common.youtrack_mcp_client import IssueNotFound
from contracts.IssueContext import IssueContext
from contracts.TaskPlan import TaskPlan
logger = logging.getLogger("triage_agent")
logger.setLevel(logging.DEBUG)
@dataclass
class TriageContext:
issue: IssueContext
registry: AgentRegistry
triage_agent: IssueTriageAgent
async def run_triage(ctx: TriageContext) -> TriageMarker:
# --- Step 1: cheap deterministic gate ---
action, prior = precheck(ctx.issue, ctx.registry)
if action == Action.SKIP:
assert prior is not None
return prior
# --- Step 2: run both Phase A passes ---
# in production: concurrent; here sequential for clarity
cap = ctx.triage_agent.audit_capability(ctx.issue)
amb = ctx.triage_agent.audit_ambiguity(ctx.issue)
# --- Step 3: merge ---
decision = merge(cap, amb)
# --- Step 4: act ---
if decision.action == Action.POST_CAPABILITY_BLOCKER:
marker = TriageMarker(
issue_hash=ctx.issue.body_hash(),
capability=cap.status,
clarity=amb.status,
decision="BLOCKED_CAPABILITY",
missing_caps=decision.missing_caps,
)
logger.debug("add_issue_comment" + json.dumps({ "issueId": ctx.issue.issue_id, "text": render_blocker_comment(marker, cap, amb, decision), })),
await ctx.triage_agent.youtrack_mcp.call_tool(
"add_issue_comment",
{
"issueId": ctx.issue.issue_id,
"text": render_blocker_comment(marker, cap, amb, decision),
},
)
return marker
if decision.action == Action.POST_QUESTIONS:
marker = TriageMarker(
issue_hash=ctx.issue.body_hash(),
capability=cap.status,
clarity="AWAITING_REPLY",
decision="BLOCKED_CLARITY",
missing_caps=cap.missing_capabilities,
)
await ctx.triage_agent.youtrack_mcp.call_tool(
"add_issue_comment",
{
"issueId": ctx.issue.issue_id,
"text": render_questions_comment(marker, cap, amb, decision),
},
)
return marker
if decision.action == Action.PROCEED_TO_SEQUENCE:
plan = ctx.triage_agent.sequence(
ctx.issue, ctx.registry,
assumptions=decision.assumptions,
)
# validate before posting — see previous message
errors = validate_plan(plan, ctx.registry)
if errors:
plan = ctx.triage_agent.sequence( # one retry with errors fed back
ctx.issue, ctx.registry,
assumptions=decision.assumptions + [f"FIX: {e}" for e in errors],
)
errors = validate_plan(plan, ctx.registry)
if errors:
marker = TriageMarker(
issue_hash=ctx.issue.body_hash(),
capability="COVERED",
clarity=amb.status,
decision="BLOCKED_CAPABILITY",
missing_caps=["plan-validation-failed"],
)
await ctx.triage_agent.youtrack_mcp.call_tool(
"add_issue_comment",
{
"issueId": ctx.issue.issue_id,
"text": render_validation_failure(marker, plan, errors),
},
)
return marker
marker = TriageMarker(
issue_hash=ctx.issue.body_hash(),
capability="COVERED",
clarity=amb.status,
decision="PLANNED",
)
await ctx.triage_agent.youtrack_mcp.call_tool(
"add_issue_comment",
{
"issueId": ctx.issue.issue_id,
"text": render_plan_comment(marker, plan, decision.assumptions, decision.questions),
},
)
# handoff to executor — enqueue, don't block
# ctx.tracker.enqueue_execution(ctx.issue.id, plan)
return marker
raise AssertionError(f"unhandled action: {decision.action}")
CAPABILITY_PROMPT = """
You are an adversarial capability auditor for an autonomous coding system.
Your job is NOT to plan the task. Your job is to find reasons the task
CANNOT be completed using ONLY the agents listed below.
Be specific. A valid objection must:
- Name a concrete capability the issue requires.
- Point to the part of the issue that requires it.
- Show that no listed agent provides that capability.
- Explain why no combination of listed agents can substitute.
Invalid objections (do NOT produce these):
- Vague worries ("this might be complex").
- Missing information ("I don't know the repo"). That's a separate
category — see below.
- Capabilities that ARE covered but that you'd prefer were covered
by a different agent.
If you can find no valid objection, return status "COVERED" with an
empty objections list. Do NOT invent objections to seem useful.
Do NOT produce a plan. Do NOT suggest new agents in this step.
## ISSUE
{{ISSUE}}
## AVAILABLE AGENTS
{{AGENTS}}
## OUTPUT SCHEMA
{
"status": "COVERED" | "GAP" | "INSUFFICIENT_INFO",
"objections": [
{
"required_capability": "<short tag>",
"needed_for": "<quote or paraphrase from the issue>",
"why_no_agent_covers_it": "<which agents you considered and why they fail>",
"substitution_attempt": "<could two agents combine? why not?>"
}
],
"missing_capabilities": ["<deduped tags from objections>"],
"questions": ["..."] // only if status = INSUFFICIENT_INFO
}
"""
AMBIGUITY_PROMPT = """
You are an ambiguity auditor for an autonomous coding system.
A downstream planner will convert this issue into a concrete execution
plan. Your job is to find everything that would force that planner to
GUESS.
A valid ambiguity finding satisfies ALL of:
1. A reasonable engineer, given only the issue text, could interpret it
two or more materially different ways.
2. The interpretations lead to different plans (different steps,
different agents, different outputs).
3. The issue text does not resolve which interpretation is correct.
Invalid findings (do NOT produce these):
- "The issue is vague" without naming the two interpretations.
- Missing info that ANY competent engineer would infer from context
(repo conventions, standard tooling, obvious defaults).
- Missing info that the PLANNER can decide without risk. If a choice
is low-stakes and reversible, the planner may pick a default.
- Requests for information that exists elsewhere in the issue or in
the provided context.
If you find no valid ambiguity, return status "CLEAR" with an empty
list. Do NOT invent questions to seem thorough. An empty list is a
valid, common, and correct answer.
## ISSUE
{{ISSUE}}
## CONTEXT THE PLANNER WILL HAVE
- Available agents:
{{AGENTS}}
## OUTPUT SCHEMA
{
"status": "CLEAR" | "AMBIGUOUS",
"findings": [
{
"aspect": "<short label, e.g. 'target environment'>",
"interpretation_a": "<one concrete reading>",
"interpretation_b": "<the other concrete reading>",
"why_it_matters": "<what the plan does differently under each>",
"suggested_question": "<one sentence to post to the reporter>",
"blocking": true | false
}
],
"assumptions_planner_may_make": [
"<low-stakes defaults the planner is safe to pick>"
],
"questions": ["<deduped suggested_question for blocking findings>"]
}
"""
SEQUENCE_PROMPT = """\
You are a task sequencing planner for an autonomous coding system.
You receive an issue and a list of AVAILABLE AGENTS. Your job is to
produce a dependency-ordered plan — a DAG of steps — that, if executed
in the correct order, resolves the issue.
You are NOT deciding whether the task is possible. That decision has
already been made: the required capabilities are covered by the agents
listed below. Your job is to sequence and estimate.
## HARD CONSTRAINTS
Violating any of these makes your output invalid:
1. You may ONLY use agents whose `id` appears in AVAILABLE AGENTS.
Never invent agents. Never use placeholder ids like "TODO".
2. Every step must have a unique `step_id` (s01, s02, ...).
3. Every step must have `depends_on` — an array of step_ids, possibly
empty. Declare every dependency, including transitive ones you rely
on for data. Missing a dependency is worse than an unnecessary one.
4. The dependency graph must be acyclic.
5. Every step must have a `verification` field: how we confirm the step
succeeded. If you cannot state a verification, the step is malformed.
6. Every step must have `inputs` and `outputs` naming concrete artifacts.
Every `input` must either be produced by an ancestor step or be one
of the GLOBAL inputs listed below.
7. Estimates are in minutes. `p90` must be >= `p50`. Prefer wide ranges
over false precision. Do not estimate optimistically to look good.
8. Each step's `goal` is imperative and one sentence. Not a category.
Bad: "handle testing". Good: "run the pytest suite and report failures".
9. If you cannot produce a valid plan, return
status "INSUFFICIENT_INFO" with `open_questions` explaining what is
missing. Do NOT fabricate steps to fill the plan.
You will be called once. Return ONLY the JSON object described below.
No prose. No markdown fences. No commentary before or after.
## ISSUE
{{ISSUE}}
## AVAILABLE AGENTS
{{AGENTS}}
## ASSUMPTIONS ALREADY MADE
The triage agent has already decided the following. Treat them as given;
do not re-litigate them and do not add steps to verify them.
{{ASSUMPTIONS}}
## GLOBAL INPUTS
These artifacts exist before the plan runs and may be consumed by any
step without a producing ancestor:
- issue_body
- repo
- repo:files
- acceptance_criteria
- env:credentials
## OUTPUT SCHEMA
{
"status": "PLANNED" | "INSUFFICIENT_CAPABILITY" | "INSUFFICIENT_INFO",
"summary": "<one sentence describing the approach>",
"steps": [
{
"step_id": "s01",
"agent_id": "<must be in AVAILABLE AGENTS>",
"goal": "<imperative, one sentence>",
"inputs": ["<artifact>", ...],
"outputs": ["<artifact>", ...],
"depends_on": ["s00", ...],
"verification": "<how we confirm success>",
"estimate_minutes": {"p50": <int>, "p90": <int>},
"risk": "low" | "medium" | "high",
"notes": "<optional, one line>"
}
],
"parallel_groups": [["s01","s02"], ["s03"]],
"critical_path": ["s01","s03"],
"total_estimate_minutes": {"p50": <int>, "p90": <int>},
"assumptions": ["..."],
"open_questions": ["..."],
"missing_capabilities": ["..."]
}
Field notes:
- `parallel_groups` lists steps that can run concurrently. Every step in
a group must have identical `depends_on` sets and no inter-dependencies.
- `critical_path` is the longest dependency chain by `estimate_minutes.p50`.
- `total_estimate_minutes.p50` is the sum of p50 along the critical path
— NOT the sum of all steps. Parallel work does not add.
Compute it; do not guess.
- `total_estimate_minutes.p90` is the sum of p90 along the p90-weighted
critical path. This path may differ from the p50 one.
- `missing_capabilities` is populated ONLY when
status = "INSUFFICIENT_CAPABILITY". Otherwise it is an empty array.
- `open_questions` is populated when status = "INSUFFICIENT_INFO".
## RULES FOR SEQUENCING
- Use the minimum number of steps that fully resolves the issue.
Do not add ceremony steps (planning, review, cleanup) unless the
issue explicitly asks for them or an agent is dedicated to them.
- Prefer narrow, verifiable steps over one large step.
- If two steps touch the same artifact, they are NOT parallel — one
must depend on the other.
- A "review" or "gate" step is only valid if an agent whose job is
review/gating is listed in AVAILABLE AGENTS. Otherwise do not add one.
- If the issue is trivially resolvable by a single agent in a single
step, return a one-step plan. Do not inflate it.
- If any part of the issue is unresolvable with the given agents, return
status "INSUFFICIENT_CAPABILITY" with the missing tags. Do not
substitute a different capability and hope it works.
## EXAMPLES
### Example 1 — single-step plan
Issue: "Add a `--verbose` flag to the CLI that prints extra diagnostics."
Available agents:
- id: code-writer, capabilities: [codegen:python, edit_repo]
- id: test-runner, capabilities: [run_tests]
Correct output:
{
"status": "PLANNED",
"summary": "Add a --verbose flag to the CLI argument parser and thread it through the logger.",
"steps": [
{
"step_id": "s01",
"agent_id": "code-writer",
"goal": "Add a --verbose flag to the CLI parser and enable debug logging when set.",
"inputs": ["repo:files"],
"outputs": ["repo:files"],
"depends_on": [],
"verification": "Running `cli --verbose` emits DEBUG-level log lines; without it, none appear.",
"estimate_minutes": {"p50": 15, "p90": 40},
"risk": "low"
}
],
"parallel_groups": [["s01"]],
"critical_path": ["s01"],
"total_estimate_minutes": {"p50": 15, "p90": 40},
"assumptions": [],
"open_questions": [],
"missing_capabilities": []
}
### Example 2 — two-step plan with a dependency
Issue: "Fix the off-by-one in pagination and add a regression test."
Available agents:
- id: code-writer, capabilities: [codegen:python, edit_repo]
- id: test-runner, capabilities: [run_tests]
Correct output:
{
"status": "PLANNED",
"summary": "Fix the pagination off-by-one, then add a regression test that fails on the old behavior.",
"steps": [
{
"step_id": "s01",
"agent_id": "code-writer",
"goal": "Fix the off-by-one in the pagination offset calculation.",
"inputs": ["repo:files", "issue_body"],
"outputs": ["repo:files"],
"depends_on": [],
"verification": "The existing pagination unit test passes, and manual check with page=1 returns items 0-9.",
"estimate_minutes": {"p50": 20, "p90": 60},
"risk": "medium"
},
{
"step_id": "s02",
"agent_id": "code-writer",
"goal": "Add a regression test covering page boundaries.",
"inputs": ["repo:files"],
"outputs": ["repo:files"],
"depends_on": ["s01"],
"verification": "The new test fails if s01 is reverted, passes with s01 applied.",
"estimate_minutes": {"p50": 15, "p90": 40},
"risk": "low"
}
],
"parallel_groups": [["s01"], ["s02"]],
"critical_path": ["s01", "s02"],
"total_estimate_minutes": {"p50": 35, "p90": 100},
"assumptions": [],
"open_questions": [],
"missing_capabilities": []
}
Note in Example 2: `test-runner` is available but NOT used — the issue
asks to *add* a test, which is code-writing work. Do not force agents
into plans just because they exist.
### Example 3 — insufficient capability (escape hatch)
Issue: "Migrate the users table to add a `last_login` timestamp column."
Available agents:
- id: code-writer, capabilities: [codegen:python, edit_repo]
- id: test-runner, capabilities: [run_tests]
Correct output:
{
"status": "INSUFFICIENT_CAPABILITY",
"summary": "",
"steps": [],
"parallel_groups": [],
"critical_path": [],
"total_estimate_minutes": {"p50": 0, "p90": 0},
"assumptions": [],
"open_questions": [],
"missing_capabilities": ["db:migration"]
}
Now produce the JSON for the issue above.
"""
def prior_comment_id(issue: IssueContext) -> str:
for c in reversed(issue.comments):
if TriageMarker.parse(c['body']):
return c['id']
return ""
def _capability_now_available(cap_str: str, registry: AgentRegistry) -> bool:
try:
cap = Capability(cap_str)
except ValueError:
return False
return bool(registry.by_capability(cap))
def precheck(issue: IssueContext, registry: AgentRegistry) -> tuple[Action, TriageMarker | None]:
"""
Returns the action to take before doing any LLM work.
This is the cheap, deterministic gate.
"""
prior = find_prior_marker(issue)
if prior is None:
return Action.RERUN, None
# issue body changed since last triage → re-run everything
if prior.issue_hash != issue.body_hash():
return Action.RERUN, prior
# PLANNED: never re-plan. Execution is downstream's problem.
if prior.decision == "PLANNED":
return Action.SKIP, prior
# BLOCKED_CLARITY: wait for reporter reply
if prior.decision == "BLOCKED_CLARITY":
if has_reporter_reply_since(issue, prior_comment_id(issue)):
return Action.RERUN, prior
else:
logger.debug("NO REPLY")
return Action.SKIP, prior
# BLOCKED_CAPABILITY: check if new agents close the gap
if prior.decision == "BLOCKED_CAPABILITY":
still_missing = [
c for c in prior.missing_caps
if not _capability_now_available(c, registry)
]
if not still_missing:
return Action.RERUN, prior
# even if still missing, if reporter added info, re-run
# to potentially discover the issue changed shape
if has_reporter_reply_since(issue, prior_comment_id(issue)):
return Action.RERUN, prior
return Action.SKIP, prior
return Action.RERUN, prior
@agent(
name="IssueTriageAgent",
description="Dispatcher Agent with idempotency guarantees",
version="1.0.0",
capabilities={"text", "llm"},
tags={"development"},
input_schema={"type": "object", "properties": {"text": {"type": "string"}}},
priority=10,
)
class IssueTriageAgent:
def __init__(self, context_builder: YouTrackContextBuilder, llm: LLMClient, agent_registry: AgentRegistry, youtrack_mcp):
self.name = "IssueTriageAgent"
self.ctx_builder = context_builder
self.decomposer = IssueDecomposer(llm)
self.llm = llm
self.agent_registry = agent_registry
self.youtrack_mcp = youtrack_mcp
async def plan_issue(self, issue_id: str) -> TriageMarker:
issue_ctx = await self.ctx_builder.build(issue_id)
if issue_ctx is None:
raise IssueNotFound(f"Issue {issue_id} was not found in YouTrack")
return await run_triage(TriageContext(
issue_ctx,
self.agent_registry,
self,
))
def audit_capability(
self, issue: IssueContext
) -> CapabilityResult:
raw = self.llm.chat_with_schema(
(CAPABILITY_PROMPT
.replace("{{AGENTS}}", self.agent_registry.to_prompt_text())
.replace("{{ISSUE}}", issue.to_prompt_text())),
{
"type": "json_schema",
"json_schema": {
"name": "CapabilityResult",
"schema": CapabilityResult.model_json_schema(),
"strict": True,
},
},
)
return parse_capability_result(raw)
def audit_ambiguity(
self, issue: IssueContext
) -> AmbiguityResult:
prompt = (
AMBIGUITY_PROMPT
.replace("{{ISSUE}}", issue.to_prompt_text())
.replace("{{AGENTS}}", self.agent_registry.to_prompt_text())
.replace(
"{{PROJECT}}",
issue.project.to_prompt_text() if issue.project else "",
)
)
raw = self.llm.chat_with_schema(
prompt,
{
"type": "json_schema",
"json_schema": {
"name": "AmbiguityResult",
"schema": AmbiguityResult.model_json_schema(),
"strict": True,
},
},
)
try:
return parse_ambiguity_result(raw)
except ValueError as exc:
logger.warning("ambiguity parse failed: %s — raw=%r", exc, raw)
return AmbiguityResult(
status="AMBIGUOUS",
findings=[
AmbiguityFinding(
aspect="triage-internal",
interpretation_a="(internal parser error)",
interpretation_b="(internal parser error)",
why_it_matters="triage agent could not interpret its own analysis",
suggested_question="triage agent failed to analyze this issue; please review manually",
blocking=True,
)
],
questions=["triage agent failed to analyze this issue; please review manually"],
)
def sequence(
self, issue: IssueContext, registry: AgentRegistry,
assumptions: list[str],
) -> Plan:
prompt = (
SEQUENCE_PROMPT
.replace("{{ISSUE}}", issue.to_prompt_text())
.replace("{{AGENTS}}", registry.to_prompt_text())
.replace("{{ASSUMPTIONS}}", "\n".join(f"- {a}" for a in assumptions) if assumptions else "(none — you are free to make low-stakes defaults)")
.replace("{{PROJECT}}", issue.project.to_prompt_text() if issue.project else "")
)
raw = self.llm.client.chat(
prompt,
response_format={
"type": "json_schema",
"json_schema": {
"name": "TaskPlan",
"schema": Plan.model_json_schema(),
"strict": True,
},
},
)
return parse_task_plan(raw)
agents/issue_reader/README.md → agents/issue_triage/README.md
View File
agents/issue_reader/__init__.py → agents/issue_triage/__init__.py
View File
agents/issue_reader/context_builder.py → agents/issue_triage/context_builder.py
+8 -4
View File
@@ -1,14 +1,15 @@
# context_builder.py
import base64
import json
import logging
import re
import httpx
from agents.issue_reader.models import IssueContext
from agents.issue_reader.project_loader import load_project
from agents.issue_triage.project_loader import load_project
from common.as_type import _as_dict, _as_list, _first, _extract_base64
from common.youtrack_mcp_client import IssueNotFound
from contracts.IssueContext import IssueContext
logger = logging.getLogger("context_builder")
logger.setLevel(logging.DEBUG)
@@ -47,6 +48,7 @@ class YouTrackContextBuilder:
)
logger.debug(f"Found {len(comments_raw)} comments")
logger.debug(json.dumps(comments_raw))
comments = _as_list(comments_raw)
except IssueNotFound:
comments = []
@@ -60,7 +62,8 @@ class YouTrackContextBuilder:
body=_first(issue, "description", "body", default=""),
comments=[
{
"author": (c.get("author") or {}).get("login", "unknown"),
"id": c.get("url"),
"author": c.get("author") or "unknown",
"body": c.get("text") or c.get("body") or "",
"created_at": str(c.get("created") or c.get("created_at") or ""),
}
@@ -81,7 +84,8 @@ class YouTrackContextBuilder:
"state": self._extract_field(issue, "State"),
},
images=images,
project=project
project=project,
acceptance_criteria=""
)
async def _fetch_image_base64(self, issue: dict) -> list[str]:
agents/issue_reader/decomposer.py → agents/issue_triage/decomposer.py
+6 -5
View File
@@ -3,8 +3,9 @@ import json
from openai import images
from agents.issue_reader.models import IssueContext, Task, TaskPlan, TaskKind, Atomicity
from contracts.IssueContext import IssueContext
from contracts.Task import Task, TaskKind, Atomicity
from contracts.TaskPlan import TaskPlan
SYSTEM_PROMPT = """You are a senior engineer.
Your job: given a YouTrack issue, its comments, and the project context,
@@ -37,7 +38,7 @@ Schema:
"acceptance_criteria": ["..."],
"depends_on": ["t0"],
"atomicity": "atomic|needs_split",
"estimate_hours": 4,
"estimate_minutes": 240,
"files_hint": ["src/foo.py"],
"open_questions": ["..."]
}
@@ -70,7 +71,7 @@ class IssueDecomposer:
acceptance_criteria=t.get("acceptance_criteria", []),
depends_on=t.get("depends_on", []),
atomicity=Atomicity(t.get("atomicity", "atomic")),
estimate_hours=t.get("estimate_hours"),
estimate_minutes=t.get("estimate_minutes"),
files_hint=t.get("files_hint", []),
open_questions=t.get("open_questions", []),
)
@@ -121,7 +122,7 @@ class IssueDecomposer:
acceptance_criteria=t.get("acceptance_criteria", []),
depends_on=t.get("depends_on", []),
atomicity=Atomicity(t.get("atomicity", "atomic")),
estimate_hours=t.get("estimate_hours"),
estimate_minutes=t.get("estimate_minutes"),
files_hint=t.get("files_hint", []),
open_questions=t.get("open_questions", []),
)
agents/issue_triage/helpers.py
+153
View File
@@ -0,0 +1,153 @@
from collections.abc import Iterator
from agents.issue_triage.models import Step
from contracts.Task import Task
from contracts.TaskPlan import TaskPlan
def has_cycle(steps: list[Task]) -> bool:
"""
Return True if the dependency graph contains a cycle.
Ignores depends_on entries that reference unknown step_ids
(those are caught by a separate validator).
"""
by_id = {s.id: s for s in steps}
WHITE, GRAY, BLACK = 0, 1, 2
color: dict[str, int] = {sid: WHITE for sid in by_id}
for start in by_id:
if color[start] != WHITE:
continue
# iterative DFS: stack of (node, iterator over deps)
stack: list[tuple[str, Iterator[str]]] = [
(start, iter(by_id[start].depends_on))
]
color[start] = GRAY
while stack:
node, it = stack[-1]
try:
dep = next(it)
except StopIteration:
color[node] = BLACK
stack.pop()
continue
if dep not in by_id:
continue # unknown dep — handled elsewhere
if color[dep] == GRAY:
return True # back edge → cycle
if color[dep] == WHITE:
color[dep] = GRAY
stack.append((dep, iter(by_id[dep].depends_on)))
return False
def transitive_ancestors(step: Task, all_steps: list[Task]) -> set[str]:
"""
All step_ids that `step` (transitively) depends on.
Assumes the graph is acyclic — call has_cycle first, or this may loop.
"""
by_id = {s.id: s for s in all_steps}
seen: set[str] = set()
stack = list(step.depends_on)
while stack:
sid = stack.pop()
if sid in seen:
continue
seen.add(sid)
node = by_id.get(sid)
if node is None:
continue
stack.extend(node.depends_on)
return seen
def transitive_ancestor_steps(step: Task, all_steps: list[Task]) -> list[Task]:
ids = transitive_ancestors(step, all_steps)
return [s for s in all_steps if s.id in ids]
GLOBAL_INPUTS: frozenset[str] = frozenset({
"issue_body",
"repo",
"repo:files",
"acceptance_criteria",
"env:credentials",
# add as your executor's pre-existing inputs grow
})
def longest_path(steps: list[Task]) -> list[str]:
"""
Return the step_ids on the critical path (by estimate_p50).
Assumes acyclic. Returns the empty list for empty input.
"""
if not steps:
return []
by_id = {s.id: s for s in steps}
# topological order (Kahn's algorithm)
indeg: dict[str, int] = {sid: 0 for sid in by_id}
children: dict[str, list[str]] = {sid: [] for sid in by_id}
for s in steps:
for dep in s.depends_on:
if dep not in by_id:
continue
indeg[s.id] += 1
children[dep].append(s.id)
queue = [sid for sid, d in indeg.items() if d == 0]
topo: list[str] = []
while queue:
sid = queue.pop()
topo.append(sid)
for child in children[sid]:
indeg[child] -= 1
if indeg[child] == 0:
queue.append(child)
# if topo doesn't cover all steps, there's a cycle
# (defensive; validator should have caught it)
if len(topo) != len(by_id):
raise ValueError("longest_path: cycle detected in plan")
# DP over topo order:
# dist[sid] = longest weighted path ending at sid
# prev[sid] = predecessor on that path
dist: dict[str, int] = {}
prev: dict[str, str | None] = {}
for sid in topo:
step = by_id[sid]
best = step.estimate_p50
best_pred: str | None = None
for dep in step.depends_on:
if dep not in by_id:
continue
candidate = dist[dep] + step.estimate_p50
if candidate > best:
best = candidate
best_pred = dep
dist[sid] = best
prev[sid] = best_pred
# find sink with max dist, walk back
end = max(dist, key=lambda sid: dist[sid])
path: list[str] = []
cur: str | None = end
while cur is not None:
path.append(cur)
cur = prev[cur]
path.reverse()
return path
def path_steps(path: list[str], plan: TaskPlan) -> list[Task]:
"""Resolve a list of step_ids to their Step objects, in path order."""
by_id = {s.id: s for s in plan.tasks}
return [by_id[sid] for sid in path if sid in by_id]
agents/issue_triage/marker.py
+69
View File
@@ -0,0 +1,69 @@
import json
import logging
import os
from dataclasses import dataclass, field
from typing import Literal
from contracts.IssueContext import IssueContext
logger = logging.getLogger("marker_agent")
logger.setLevel(logging.DEBUG)
MARKER_PREFIX = "<!-- triage-agent:v1"
@dataclass
class TriageMarker:
issue_hash: str
capability: Literal["COVERED", "GAP", "INSUFFICIENT_INFO"]
clarity: Literal["CLEAR", "AMBIGUOUS", "AWAITING_REPLY"]
decision: Literal["PLANNED", "BLOCKED_CAPABILITY", "BLOCKED_CLARITY"]
missing_caps: list[str] = field(default_factory=list)
def render(self) -> str:
caps = ",".join(self.missing_caps)
return (
f"{MARKER_PREFIX} issue-hash:{self.issue_hash} "
f"capability:{self.capability} clarity:{self.clarity} "
f"decision:{self.decision} missing-caps:{caps} -->"
)
@classmethod
def parse(cls, text: str) -> "TriageMarker | None":
import re
m = re.search(
r"<!-- triage-agent:v1 issue-hash:(\w+) capability:(\w+) "
r"clarity:(\w+) decision:(\w+) missing-caps:([\w:,\-/\s]*)\s*-->",
text,
)
if not m:
return None
caps = [c for c in m.group(5).split(",") if c]
return cls(
issue_hash=m.group(1),
capability=m.group(2), # type: ignore
clarity=m.group(3), # type: ignore
decision=m.group(4), # type: ignore
missing_caps=caps,
)
def find_prior_marker(issue: IssueContext) -> TriageMarker | None:
logger.debug(json.dumps(issue.comments, indent=4))
for c in reversed(issue.comments):
m = TriageMarker.parse(c['body'])
if m:
return m
logger.debug("No prior marker found")
return None
def has_reporter_reply_since(issue: IssueContext, marker_comment_id: str) -> bool:
found = False
for c in issue.comments:
if c['id'] == marker_comment_id:
found = True
continue
if found and c['author'] != os.getenv('YOUTRACK_TRIAGE_AUTHOR'):
return True
return False
agents/issue_triage/merge.py
+63
View File
@@ -0,0 +1,63 @@
from dataclasses import field, dataclass
from enum import Enum, auto
from agents.issue_triage.models import CapabilityResult, AmbiguityResult
class Action(Enum):
PROCEED_TO_SEQUENCE = auto()
POST_CAPABILITY_BLOCKER = auto()
POST_QUESTIONS = auto()
SKIP = auto() # idempotent no-op
RERUN = auto() # re-trigger fired, redo passes
@dataclass
class MergeDecision:
action: Action
reason: str
# for POST_* actions, what to include
missing_caps: list[str] = field(default_factory=list)
questions: list[str] = field(default_factory=list)
assumptions: list[str] = field(default_factory=list)
def merge(
cap: CapabilityResult,
amb: AmbiguityResult,
) -> MergeDecision:
# Insufficient info from either pass dominates — we don't have enough
# to even classify. Ask questions, don't post a blocker.
if cap.status == "INSUFFICIENT_INFO":
return MergeDecision(
Action.POST_QUESTIONS,
reason="capability pass needs more info",
questions=cap.questions,
)
if amb.status == "AMBIGUOUS" and amb.blocking_findings():
# blocking ambiguity masks capability — we may be asking about
# capabilities the issue didn't actually require.
return MergeDecision(
Action.POST_QUESTIONS,
reason="blocking ambiguity",
questions=[f.suggested_question for f in amb.blocking_findings()],
assumptions=amb.assumptions_planner_may_make,
)
if cap.status == "GAP":
return MergeDecision(
Action.POST_CAPABILITY_BLOCKER,
reason="capability gap",
missing_caps=cap.missing_capabilities,
# non-blocking ambiguity still goes into the plan later, but
# we can't plan yet, so surface it in the blocker comment too
questions=[f.suggested_question for f in amb.findings],
)
# cap == COVERED, clarity == CLEAR (or non-blocking ambiguous)
return MergeDecision(
Action.PROCEED_TO_SEQUENCE,
reason="covered and clear",
assumptions=amb.assumptions_planner_may_make,
questions=[f.suggested_question for f in amb.findings if not f.blocking],
)
agents/issue_triage/models.py
+231
View File
@@ -0,0 +1,231 @@
import json
from dataclasses import dataclass, field
from typing import Literal
from pydantic import BaseModel, Field, ValidationError, ConfigDict
class Objection(BaseModel):
model_config = ConfigDict(extra="forbid")
required_capability: str
needed_for: str
why_no_agent_covers_it: str
substitution_attempt: str
class CapabilityResult(BaseModel):
model_config = ConfigDict(extra="forbid")
status: Literal["COVERED", "GAP", "INSUFFICIENT_INFO"]
objections: list[Objection]
missing_capabilities: list[str]
questions: list[str]
# --- Phase A.2: ambiguity ---
class AmbiguityFinding(BaseModel):
model_config = ConfigDict(extra="forbid")
aspect: str
interpretation_a: str
interpretation_b: str
why_it_matters: str
suggested_question: str
blocking: bool
class AmbiguityResult(BaseModel):
model_config = ConfigDict(extra="forbid")
status: Literal["CLEAR", "AMBIGUOUS"]
findings: list[AmbiguityFinding]
assumptions_planner_may_make: list[str]
questions: list[str]
def blocking_findings(self) -> list[AmbiguityFinding]:
return [f for f in self.findings if f.blocking]
class EstimateModel(BaseModel):
model_config = ConfigDict(extra="forbid")
p50: int
p90: int
@dataclass
class Step:
model_config = ConfigDict(extra="forbid")
step_id: str
agent_id: str
goal: str
inputs: list[str]
outputs: list[str]
depends_on: list[str]
verification: str
estimate_p50: int
estimate_p90: int
risk: Literal["low", "medium", "high"]
class Plan(BaseModel):
model_config = ConfigDict(extra="forbid")
status: Literal["PLANNED", "INSUFFICIENT_CAPABILITY", "INSUFFICIENT_INFO"]
summary: str
steps: list[Step]
parallel_groups: list[list[str]]
critical_path: list[str]
total_p50: int
total_p90: int
assumptions: list[str]
open_questions: list[str]
missing_capabilities: list[str] = Field(default_factory=list)
def parse_capability_result(raw: str | dict) -> CapabilityResult:
"""
Convert an LLM response (JSON string or already-parsed dict) into a
CapabilityResult. Raises ValueError on malformed input.
"""
if isinstance(raw, str):
try:
data = json.loads(_strip_code_fences(raw))
except json.JSONDecodeError as exc:
raise ValueError(f"capability response not valid JSON: {exc}") from exc
else:
data = raw
try:
model = CapabilityResult.model_validate(data)
except ValidationError as exc:
raise ValueError(f"capability response failed schema: {exc}") from exc
# cross-field consistency the schema can't express
if model.status == "GAP" and not model.missing_capabilities:
raise ValueError("status=GAP but missing_capabilities is empty")
if model.status == "COVERED" and model.objections:
raise ValueError("status=COVERED but objections is non-empty")
if model.status == "INSUFFICIENT_INFO" and not model.questions:
raise ValueError("status=INSUFFICIENT_INFO but questions is empty")
return CapabilityResult(
status=model.status,
objections=[
Objection(
required_capability=o.required_capability,
needed_for=o.needed_for,
why_no_agent_covers_it=o.why_no_agent_covers_it,
substitution_attempt=o.substitution_attempt,
)
for o in model.objections
],
missing_capabilities=list(model.missing_capabilities),
questions=list(model.questions),
)
def _strip_code_fences(s: str) -> str:
"""Some models wrap JSON in ```json ... ``` despite instructions."""
s = s.strip()
if s.startswith("```"):
# drop first line (``` or ```json) and trailing ```
s = s.split("\n", 1)[1] if "\n" in s else s
if s.endswith("```"):
s = s[: -3]
return s.strip()
def parse_ambiguity_result(raw: str | dict) -> AmbiguityResult:
if isinstance(raw, str):
try:
data = json.loads(_strip_code_fences(raw))
except json.JSONDecodeError as exc:
raise ValueError(f"ambiguity response not valid JSON: {exc}") from exc
else:
data = raw
try:
model = AmbiguityResult.model_validate(data)
except ValidationError as exc:
raise ValueError(f"ambiguity response failed schema: {exc}") from exc
# cross-field consistency the schema can't express
if model.status == "CLEAR" and model.findings:
raise ValueError("status=CLEAR but findings is non-empty")
if model.status == "AMBIGUOUS" and not model.findings:
raise ValueError("status=AMBIGUOUS but findings is empty")
return AmbiguityResult(
status=model.status,
findings=[
AmbiguityFinding(
aspect=f.aspect,
interpretation_a=f.interpretation_a,
interpretation_b=f.interpretation_b,
why_it_matters=f.why_it_matters,
suggested_question=f.suggested_question,
blocking=f.blocking,
)
for f in model.findings
],
assumptions_planner_may_make=list(model.assumptions_planner_may_make),
questions=list(model.questions),
)
def parse_task_plan(raw: str | dict) -> Plan:
if isinstance(raw, str):
try:
data = json.loads(_strip_code_fences(raw))
except json.JSONDecodeError as exc:
raise ValueError(f"sequence response not valid JSON: {exc}") from exc
else:
data = raw
try:
model = Plan.model_validate(data)
except ValidationError as exc:
raise ValueError(f"sequence response failed schema: {exc}") from exc
_check_task_plan_semantics(model)
return Plan(
status=model.status,
summary=model.summary,
steps=[
Step(
step_id=s.step_id,
agent_id=s.agent_id,
goal=s.goal,
inputs=list(s.inputs),
outputs=list(s.outputs),
depends_on=list(s.depends_on),
verification=s.verification,
estimate_p50=s.estimate_p50,
estimate_p90=s.estimate_p90,
risk=s.risk,
)
for s in model.steps
],
parallel_groups=[list(g) for g in model.parallel_groups],
critical_path=list(model.critical_path),
total_p50=model.total_estimate_minutes.p50,
total_p90=model.total_estimate_minutes.p90,
assumptions=list(model.assumptions),
open_questions=list(model.open_questions),
missing_capabilities=list(model.missing_capabilities),
)
def _check_task_plan_semantics(model: Plan) -> None:
if model.status == "PLANNED":
if not model.steps:
raise ValueError("status=PLANNED but steps is empty")
elif model.status == "INSUFFICIENT_CAPABILITY":
if not model.missing_capabilities:
raise ValueError("status=INSUFFICIENT_CAPABILITY but missing_capabilities is empty")
if model.steps:
raise ValueError("status=INSUFFICIENT_CAPABILITY but steps is non-empty")
elif model.status == "INSUFFICIENT_INFO":
if not model.open_questions:
raise ValueError("status=INSUFFICIENT_INFO but open_questions is empty")
if model.steps:
raise ValueError("status=INSUFFICIENT_INFO but steps is non-empty")
agents/issue_reader/project_loader.py → agents/issue_triage/project_loader.py
+1 -1
View File
@@ -1,7 +1,7 @@
from pathlib import Path
import yaml
from agents.issue_reader.models import ProjectContext
from contracts.ProjectContext import ProjectContext
PROJECTS_DIR = Path(__file__).parent.parent.parent / "projects"
agents/issue_triage/tests/tests.py
+25
View File
@@ -0,0 +1,25 @@
def test_skip_when_planned_and_unchanged():
issue = make_issue(comments=[marker_comment(decision="PLANNED", hash="abc")])
issue.body_hash = lambda: "abc"
action, _ = precheck(issue, empty_registry())
assert action == Action.SKIP
def test_rerun_when_new_capability_closes_gap():
issue = make_issue(comments=[marker_comment(
decision="BLOCKED_CAPABILITY",
hash="abc",
missing_caps=["db:migration"],
)])
issue.body_hash = lambda: "abc"
registry = registry_with(Agent("m", "...", frozenset({Capability.DB_MIGRATION})))
action, _ = precheck(issue, registry)
assert action == Action.RERUN
def test_blocking_ambiguity_dominates_capability_gap():
cap = CapabilityResult(status="GAP", missing_capabilities=["x"])
amb = AmbiguityResult(
status="AMBIGUOUS",
findings=[AmbiguityFinding(..., blocking=True)],
)
d = merge(cap, amb)
assert d.action == Action.POST_QUESTIONS # not POST_CAPABILITY_BLOCKER
agents/issue_triage/triage.py
+170
View File
@@ -0,0 +1,170 @@
import json
from dataclasses import dataclass, asdict
from agents.issue_triage.helpers import has_cycle, longest_path, path_steps
from agents.issue_triage.marker import TriageMarker
from agents.issue_triage.merge import Action, merge, MergeDecision
from agents.issue_triage.models import CapabilityResult, AmbiguityResult
from agents.registry import AgentRegistry
from contracts.IssueContext import IssueContext
from contracts.TaskPlan import TaskPlan
def validate_plan(plan: TaskPlan, registry):
errors = []
# 1. All agent_ids exist
for step in plan.tasks:
if step.agent not in registry and step.agent != "human":
errors.append(f"{step.id}: unknown agent {step.agent}")
# 2. depends_on references valid, earlier-declared step_ids
seen = set()
for step in plan.tasks:
for dep in step.depends_on:
if dep not in seen:
errors.append(f"{step.id}: forward or unknown dep {dep}")
seen.add(step.id)
# 3. Acyclicity (topological sort)
if has_cycle(plan.tasks):
errors.append("dependency cycle detected")
# 4. Dataflow: inputs must be produced by an ancestor
# for step in plan.tasks:
# ancestors = transitive_ancestors(step, plan.tasks)
# produced = {o for a in ancestors for o in a.get("outputs", [])}
# for inp in step.get("inputs", []):
# if inp not in produced and inp not in GLOBAL_INPUTS:
# errors.append(f"{step['step_id']}: input '{inp}' has no producer")
# 5. Estimates sane
for step in plan.tasks:
e = step.estimate_minutes
if e["p90"] < e["p50"]:
errors.append(f"{step.id}: p90 < p50")
# 6. Critical path matches depends_on
computed = longest_path(plan.tasks)
if set(computed) != set(plan.critical_path):
errors.append("critical_path mismatch")
# 7. Totals recompute, don't trust
recomputed_p50 = sum(s.estimate_minutes["p50"] for s in path_steps(computed, plan))
if recomputed_p50 != plan.total_estimate_minutes["p50"]:
errors.append("total p50 mismatch")
return errors
def render_blocker_comment(marker, cap, amb, decision) -> str:
lines = [
marker.render(),
"**Triage: BLOCKED — missing capabilities**",
"",
"This issue cannot be resolved with the currently registered agents.",
"",
]
for o in cap.objections:
lines += [
f"- **`{o.required_capability}`** needed for: {o.needed_for}",
f" - Why no agent covers it: {o.why_no_agent_covers_it}",
f" - Substitution attempt: {o.substitution_attempt}",
"",
]
if decision.missing_caps:
lines.append(f"**Missing capabilities:** {', '.join(decision.missing_caps)}")
return "\n".join(lines)
def render_questions_comment(
marker: TriageMarker,
cap: CapabilityResult,
amb: AmbiguityResult,
decision: MergeDecision,
) -> str:
lines = [
marker.render(),
"**Triage: BLOCKED — awaiting clarification**",
"",
"Please answer the following before this task can be planned:",
"",
]
for q in decision.questions:
lines.append(f"- {q}")
# if the capability pass also flagged something, surface it — the
# question may resolve it (e.g. "which DB?" could unblock db:migration)
if cap.status == "GAP" and cap.missing_capabilities:
lines += [
"",
"_Note: this issue also appears to require capabilities that "
f"no agent currently provides: `{', '.join(cap.missing_capabilities)}`. "
"The answer may change this assessment._",
]
lines += [
"",
f"_Marker (do not edit):_ `{marker.issue_hash}`",
]
return "\n".join(lines)
def plan_to_dict(plan: TaskPlan) -> dict:
return asdict(plan)
def render_validation_failure(
marker: TriageMarker,
plan: TaskPlan,
errors: list[str],
) -> str:
lines = [
marker.render(),
"**Triage: INTERNAL ERROR — plan validation failed**",
"",
"The sequencing pass produced a plan that did not pass validation "
"after one retry. This is a bug in the triage agent, not a problem "
"with the issue. A maintainer should investigate.",
"",
"### Validation errors",
"",
]
for e in errors:
lines.append(f"- `{e}`")
lines += [
"",
"### Rejected plan (raw)",
"",
"```json",
json.dumps(plan_to_dict(plan), indent=2),
"```",
"",
"_This comment was posted automatically. Do not edit the marker "
"line above._",
]
return "\n".join(lines)
def render_plan_comment(marker, plan: TaskPlan, assumptions, questions) -> str:
lines = [
marker.render(),
f"**Triage: PLANNED** — {plan.summary}",
"",
f"**Estimate:** {plan.total_p50}–{plan.total_p90} min (p50–p90)",
"",
"### Steps",
"",
]
for s in plan.tasks:
deps = f" (after {', '.join(s.depends_on)})" if s.depends_on else ""
lines.append(
f"- `{s.id}` **{s.agent}**{deps}: {s.acceptance_criteria} "
f"[{s.estimate_p50}–{s.estimate_p90}m, {s.risk}]"
)
if assumptions:
lines += ["", "### Assumptions", ""]
lines += [f"- {a}" for a in assumptions]
if questions:
lines += ["", "### Non-blocking questions", ""]
lines += [f"- {q}" for q in questions]
return "\n".join(lines)
agents/issue_reader/validator.py → agents/issue_triage/validator.py
+4 -3
View File
@@ -1,5 +1,6 @@
# validator.py
from agents.issue_reader.models import Task, TaskPlan, Atomicity
from contracts.Task import Task, Atomicity
from contracts.TaskPlan import TaskPlan
MAX_HOURS = 8
VAGUE_VERBS = ("handle", "support", "improve", "refactor", "manage", "deal with")
@@ -7,8 +8,8 @@ VAGUE_VERBS = ("handle", "support", "improve", "refactor", "manage", "deal with"
def validate(task: Task) -> list[str]:
issues = []
if task.estimate_hours and task.estimate_hours > MAX_HOURS:
issues.append(f"estimate {task.estimate_hours}h > {MAX_HOURS}h")
if task.estimate_minutes and task.estimate_minutes > MAX_HOURS:
issues.append(f"estimate {task.estimate_minutes}h > {MAX_HOURS}h")
if " and " in task.title.lower():
issues.append("title contains 'and' — likely two tasks")
if any(v in task.title.lower() for v in VAGUE_VERBS):
agents/registry.py
+156
View File
@@ -0,0 +1,156 @@
from __future__ import annotations
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Iterable, Type, TypeVar
T = TypeVar("T", bound=type)
class Capability(str, Enum):
CODEGEN_PY = "codegen:python"
CODEGEN_TS = "codegen:ts"
EDIT_REPO = "edit_repo"
RUN_TESTS = "run_tests"
OPEN_PR = "open_pr"
DB_MIGRATION = "db:migration"
INFRA_APPLY = "infra:apply"
@dataclass(frozen=True)
class AgentMeta:
name: str
description: str = ""
version: str = "0.0.0"
capabilities: frozenset[Capability] = field(default_factory=frozenset)
constraints: tuple[str, ...] = ()
input_schema: dict[str, Any] | None = None
output_schema: dict[str, Any] | None = None
priority: int = 0
tags: frozenset[str] = field(default_factory=frozenset)
requires: tuple[str, ...] = () # имена других агентов/сервисов
enabled: bool = True
extra: dict[str, Any] = field(default_factory=dict)
def has(self, capability: str) -> bool:
return capability in self.capabilities
class AgentRegistry:
def __init__(self) -> None:
self._agents: dict[str, Type] = {}
self._meta: dict[str, AgentMeta] = {}
def register(self, meta: AgentMeta) -> Callable[[T], T]:
def deco(cls: T) -> T:
if meta.name in self._agents:
existing = self._agents[meta.name]
raise ValueError(
f"Agent '{meta.name}' already registered "
f"({existing.__module__}.{existing.__qualname__})"
)
self._agents[meta.name] = cls
self._meta[meta.name] = meta
cls.agent_meta = meta # type: ignore[attr-defined]
return cls
return deco
# ---- выборки ----
def get(self, name: str) -> Type:
return self._agents[name]
def meta(self, name: str) -> AgentMeta:
return self._meta[name]
def all(self) -> dict[str, Type]:
return dict(self._agents)
def all_meta(self) -> dict[str, AgentMeta]:
return dict(self._meta)
def by_capability(self, *caps: str) -> list[AgentMeta]:
need = set(caps)
return [
m for m in self._meta.values()
if m.enabled and need.issubset(m.capabilities)
]
def by_tag(self, *tags: str) -> list[AgentMeta]:
need = set(tags)
return [m for m in self._meta.values() if need.issubset(m.tags)]
def search(self, text: str) -> list[AgentMeta]:
text = text.lower()
return [
m for m in self._meta.values()
if text in m.name.lower() or text in m.description.lower()
]
def sorted_by_priority(self) -> list[AgentMeta]:
return sorted(self._meta.values(), key=lambda m: -m.priority)
def resolve_dependencies(self, name: str) -> list[str]:
"""Топологический порядок зависимостей агента."""
order: list[str] = []
seen: set[str] = set()
temp: set[str] = set()
def visit(n: str) -> None:
if n in seen:
return
if n in temp:
raise ValueError(f"Circular dependency at '{n}'")
temp.add(n)
for dep in self._meta[n].requires:
if dep not in self._meta:
raise KeyError(f"Agent '{n}' requires unknown '{dep}'")
visit(dep)
temp.discard(n)
seen.add(n)
order.append(n)
visit(name)
return order
def to_prompt_text(self):
return "\n".join(
[
"- id:" + _agent.__qualname__
+ "\ncapabilities: ["
+ ", ".join(self.meta(_agent.__qualname__).capabilities)
+ "]" for _agent in self._agents.values()
]
)
# глобальный реестр
registry = AgentRegistry()
def agent(
name: str,
*,
description: str = "",
version: str = "0.0.0",
capabilities: Iterable[str] = (),
tags: Iterable[str] = (),
requires: Iterable[str] = (),
input_schema: dict[str, Any] | None = None,
output_schema: dict[str, Any] | None = None,
priority: int = 0,
enabled: bool = True,
**extra: Any,
):
"""Декоратор для регистрации агента с метаданными."""
meta = AgentMeta(
name=name,
description=description,
version=version,
capabilities=frozenset(capabilities),
tags=frozenset(tags),
requires=tuple(requires),
input_schema=input_schema,
output_schema=output_schema,
priority=priority,
enabled=enabled,
extra=dict(extra),
)
return registry.register(meta)
agents/PlannerAgent.py → agents/solution_architect/PlannerAgent.py
View File
agents/TesterAgent.py → agents/tester_web/TesterAgent.py
+1 -1
View File
@@ -1,4 +1,4 @@
from start import Project
from agents.codebase_analyst.start import Project
class TesterAgent:
common/as_type.py
+5
View File
@@ -14,6 +14,9 @@ def _as_dict(result: Any) -> dict:
if isinstance(result, dict):
return result
if isinstance(result, str):
return json.loads(result)
if not isinstance(result, list):
raise ValueError(f"Unexpected MCP response shape: {type(result)}")
@@ -57,6 +60,8 @@ def _as_list(result: Any) -> list[dict]:
text = result[0].get("text", "")
elif isinstance(result, list):
return result # already a list of dicts
elif isinstance(result, str):
return json.loads(result)
else:
raise ValueError(f"Unexpected MCP response shape: {type(result)}")
common/llm_client.py
+40
View File
@@ -6,6 +6,18 @@ class LLMClient:
self.model = model
self.client = OpenAI(base_url=base_url, api_key=api_key or "ollama")
def chat_with_schema(self, prompt: str, schema: dict):
assert_strict_mode_clean(schema)
kwargs = {
"model": self.model,
"messages": [
{"role": "user", "content": prompt},
],
"response_format": schema,
}
resp = self.client.chat.completions.create(**kwargs)
return resp.choices[0].message.content
def chat(self, system: str, user: str, json_mode: bool = False, images: list[str] | None = None) -> str | None:
images = images or []
if images:
@@ -29,3 +41,31 @@ class LLMClient:
kwargs["response_format"] = {"type": "json_object"}
resp = self.client.chat.completions.create(**kwargs)
return resp.choices[0].message.content
def assert_strict_mode_clean(schema: dict, path: str = "$") -> None:
"""Recursively verify a JSON Schema obeys OpenAI strict mode."""
if schema.get("type") == "object":
props = schema.get("properties", {})
required = set(schema.get("required", []))
if schema.get("additionalProperties") is not False:
raise ValueError(f"{path}: additionalProperties must be false")
for name in props:
if name not in required:
raise ValueError(f"{path}.{name}: not in required")
for name, sub in props.items():
assert_strict_mode_clean(sub, f"{path}.{name}")
if schema.get("type") == "array":
assert_strict_mode_clean(schema["items"], f"{path}[]")
for name, sub in schema.get("$defs", {}).items():
assert_strict_mode_clean(sub, f"$defs.{name}")
# OpenAI rejects these keywords in strict mode
for banned in ("default", "oneOf"):
if banned in schema:
raise ValueError(f"{path}: '{banned}' not allowed in strict mode")
common/youtrack_mcp_client.py
+23 -1
View File
@@ -63,7 +63,25 @@ class YouTrackMCPClient:
logger.debug("MCP CALL: %s with %s", name, arguments)
result = await self._call_tool_or_raise(name, arguments)
logger.debug("MCP RAW RESULT: %s", repr(result)[:500])
return result.structured_content or result.content
if result.structured_content is not None:
return result.structured_content
# Convert content blocks to JSON-serializable form
if result.content:
out = []
for block in result.content:
if getattr(block, "type", None) == "text":
out.append({"type": "text", "text": block.text})
else:
# fallback for other block types
out.append({"type": getattr(block, "type", "unknown"),
"data": str(block)})
# If it's a single text block, unwrap to plain string
if len(out) == 1 and out[0]["type"] == "text":
return out[0]["text"]
return out
return None
async def _call_tool_or_raise(self, name: str, arguments: dict):
try:
@@ -88,3 +106,7 @@ class YouTrackMCPClient:
raise RuntimeError(f"MCP tool {name} failed: {text}")
return result
async def list_tools(self):
"""List all available tools from the MCP server."""
return await self._client.list_tools()
contracts/IssueContext.py
+49
View File
@@ -0,0 +1,49 @@
import hashlib
from dataclasses import dataclass, field
from typing import Optional
from contracts.ProjectContext import ProjectContext
@dataclass
class IssueContext:
issue_id: str
title: str
body: str
acceptance_criteria: str | None
comments: list[dict]
labels: list[str] = field(default_factory=list)
repo: Optional[str] = None
metadata: dict = field(default_factory=dict)
images: list[str] = field(default_factory=list)
project: ProjectContext | None = None
def body_hash(self) -> str:
h = hashlib.sha256()
h.update(self.title.encode())
h.update(self.body.encode())
h.update((self.acceptance_criteria or "").encode())
return h.hexdigest()[:8]
def to_prompt_text(self) -> str:
parts = [f"# Issue {self.issue_id}: {self.title}", "", self.body]
if self.project:
parts.append("\n## Project info")
parts.append(self.project.to_prompt_text())
meta_lines = []
for key in ("state", "url"):
if self.metadata.get(key):
meta_lines.append(f"- {key.capitalize()}: {self.metadata[key]}")
if meta_lines:
parts.append("\n## Issue Metadata")
parts.extend(meta_lines)
if self.comments:
parts.append("\n## Discussion / Comments")
for c in self.comments:
parts.append(f"- @{c['author']} ({c['created_at']}): {c['body']}")
return "\n".join(parts)
contracts/ProjectContext.py
+45
View File
@@ -0,0 +1,45 @@
from dataclasses import field, dataclass
@dataclass
class ProjectContext:
"""Static, per-project knowledge injected into every decomposition."""
project_key: str # e.g. "ARCH"
language: str = ""
backend_framework: str = "" # "FastAPI", "Django", "Spring Boot"
frontend_framework: str = "" # "React + Vite", "Vue 3"
ui_library: str = "" # "shadcn/ui", "MUI", "Ant Design"
database: str = "" # "PostgreSQL 16 via SQLAlchemy 2"
orm: str = ""
test_framework: str = "" # "pytest + httpx.AsyncClient"
package_manager: str = "" # "uv", "poetry", "npm"
auth: str = "" # "JWT via fastapi-users"
deployment: str = "" # "Docker Compose on Hetzner"
conventions: list[str] = field(default_factory=list) # free-form notes
extra: dict[str, str] = field(default_factory=dict) # anything else
def to_prompt_text(self) -> str:
lines = [f"## Project Context ({self.project_key})"]
fields = [
("Language", self.language),
("Backend framework", self.backend_framework),
("Frontend framework", self.frontend_framework),
("UI library", self.ui_library),
("Database", self.database),
("ORM", self.orm),
("Test framework", self.test_framework),
("Package manager", self.package_manager),
("Auth", self.auth),
("Deployment", self.deployment),
]
for label, value in fields:
if value:
lines.append(f"- {label}: {value}")
for k, v in self.extra.items():
if v:
lines.append(f"- {k}: {v}")
if self.conventions:
lines.append("")
lines.append("Conventions:")
lines.extend(f"- {c}" for c in self.conventions)
return "\n".join(lines)
contracts/Task.py
+67
View File
@@ -0,0 +1,67 @@
import uuid
from dataclasses import dataclass, field
from enum import Enum
from typing import Optional
from contracts.base import BaseContract
class Atomicity(str, Enum):
ATOMIC = "atomic" # single clear action, one owner, < 1 day
NEEDS_SPLIT = "needs_split" # still too big, recurse
class TaskKind(str, Enum):
INVESTIGATE = "investigate" # read code, reproduce, research
DESIGN = "design" # API/schema decisions
IMPLEMENT = "implement" # write code
MIGRATE = "migrate" # DB / data changes
TEST = "test" # unit/integration/e2e
DOCS = "docs" # docs, changelog
REVIEW = "review" # code review, verification
OPS = "ops" # deploy, config, feature flag
@dataclass
class Task(BaseContract):
id: str
title: str
kind: TaskKind
description: str
agent: str
estimate_p50: int
estimate_p90: int
risk: Literal["low", "medium", "high"]
acceptance_criteria: list[str] = field(default_factory=list)
depends_on: list[str] = field(default_factory=list)
atomicity: Atomicity = Atomicity.ATOMIC
estimate_minutes: Optional[float] = None
files_hint: list[str] = field(default_factory=list) # probable touch points
open_questions: list[str] = field(default_factory=list)
@staticmethod
def new(**kw) -> "Task":
return Task(id=str(uuid.uuid4())[:8], **kw)
def render_dbg(self) -> str:
lines = [f"# Task: {self.description}"]
deps = f" (after {', '.join(self.depends_on)})" if self.depends_on else ""
lines.append(f"- [{self.kind.value}] {self.id}: {self.title}{deps}")
for ac in self.acceptance_criteria:
lines.append(f" ✓ {ac}")
for q in self.open_questions:
lines.append(f" ? {q}")
return "\n".join(lines)
def to_prompt_text(self) -> str:
lines = [
f"# Task: {self.description}",
f"# Before start you must ensure that another tasks have been done: {', '.join(self.depends_on)}" if self.depends_on else "",
f"# Acceptance criteria: \n{"\n - ".join(self.acceptance_criteria)}",
]
return "\n".join(lines)
def is_filled(self) -> bool:
return len(self.open_questions) == 0 and self.atomicity == Atomicity.ATOMIC
contracts/TaskPlan.py
+37
View File
@@ -0,0 +1,37 @@
from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional
from contracts.Task import Task
from contracts.base import BaseContract
class TaskPlan(BaseContract):
issue_id: str
summary: str
tasks: list[Task]
parallel_groups: list[list[str]]
critical_path: list[str]
total_p50: int
total_p90: int
assumptions: list[str]
total_estimate_minutes: Optional[float] = None
unknowns: list[str] = field(default_factory=list)
created_at: datetime = field(default_factory=datetime.utcnow)
def render_dbg(self) -> str:
lines = [f"# Plan for {self.issue_id}", self.summary, ""]
for t in self.tasks:
lines.extend(t.render_dbg())
if self.unknowns:
lines.append("\nUnknowns:")
lines += [f" - {u}" for u in self.unknowns]
return "\n".join(lines)
def to_prompt_text(self) -> str:
return self.render_dbg()
def is_filled(self) -> bool:
return self.unknowns == []
contracts/__init__.py
View File
contracts/base.py
+16
View File
@@ -0,0 +1,16 @@
from abc import abstractmethod, ABC
from pydantic import BaseModel
'''
A contract between agents
'''
class BaseContract(BaseModel):
def to_prompt_text(self) -> str:
pass
def render_dbg(self) -> str:
pass
def is_filled(self) -> bool:
pass
main.py
+16 -6
View File
@@ -5,8 +5,9 @@ from contextlib import asynccontextmanager
import httpx
from dotenv import load_dotenv
from fastapi import FastAPI, HTTPException
from agents.issue_reader.IssueReaderAgent import IssueReaderAgent
from agents.issue_reader.context_builder import YouTrackContextBuilder
from agents.issue_triage.IssueTriageAgent import IssueTriageAgent
from agents.issue_triage.context_builder import YouTrackContextBuilder
from agents.registry import AgentRegistry
from common.llm_client import LLMClient
from common.youtrack_mcp_client import YouTrackMCPClient, IssueNotFound
@@ -26,6 +27,13 @@ async def lifespan(app: FastAPI):
str(os.getenv('YOUTRACK_MCP_SERVER')),
str(os.getenv('YOUTRACK_MCP_TOKEN')),
).connect()
#
#
# res = await mcp.call_tool(
# "add_issue_comment",
# {"issueId": "ARCH-229", "text": "test"}
# )
# logger.info(res)
http_for_attachments = httpx.AsyncClient(
headers={"Authorization": f"Bearer {env('YOUTRACK_MCP_TOKEN')}"},
@@ -34,13 +42,15 @@ async def lifespan(app: FastAPI):
follow_redirects=True,
)
app.state.issue_reader_agent = IssueReaderAgent(
app.state.issue_reader_agent = IssueTriageAgent(
YouTrackContextBuilder(mcp, http_for_attachments),
LLMClient(
base_url=env("LLM_ADDRESS"),
api_key=env("LLM_API_KEY"),
model=env("LLM_MODEL"),
),
AgentRegistry(),
mcp
)
yield
@@ -59,14 +69,14 @@ async def say_hello(name: str):
return {"message": f"Hello {name}"}
@app.get("/decomposing")
@app.get("/consume_task")
async def decomposing_issue(issue: str):
agent: IssueReaderAgent = app.state.issue_reader_agent
agent: IssueTriageAgent = app.state.issue_reader_agent
try:
logger.info(f"Received an issue: {issue}")
plan = await agent.plan_issue(issue)
logger.info("Task planning successful.")
print(agent.render(plan))
logger.info(plan.render())
return {"plan": plan}
except IssueNotFound:
logger.error(f"Issue {issue} not found.")