"""High-reliability issue drafting for Jira-style requests.""" from typing import Any, Dict, Optional from request_intelligence import ( analyze_request, build_issue_fallback, build_request_contract, validate_issue_response, ) class IssueWrapper: """Draft, validate, and if necessary deterministically recover an issue.""" def __init__(self, brain=None): self.brain = brain def _messages(self, query: str, context: Dict[str, Any]) -> list[dict[str, str]]: analysis = analyze_request(query) profile = (context or {}).get("profile", {}) user_model = (context or {}).get("user_model") history = (context or {}).get("history", []) profile_context = self.brain._format_profile(profile) if self.brain else "" system = self.brain._build_system( profile_context, [], user_model, request_contract=build_request_contract(analysis), ) if self.brain else build_request_contract(analysis) system = ( "You are InvictaTill's senior product triage and issue-writing agent.\n" "Create a Jira-ready issue that faithfully represents the user's actual request.\n" f"Use issue type: {analysis.issue_type}.\n" "Required sections: Summary, Issue type, Priority, Component, Goal/Description, " "Workflow or Steps (when supplied), Acceptance criteria, Edge cases, and Details to confirm.\n" "For bugs, separate observed behavior from expected behavior, but never invent reproduction steps.\n" "For incidents, include impact and recovery/verification fields, marking unknowns To be confirmed.\n" "For stories/features, convert every supplied workflow stage into testable acceptance criteria.\n" "Attachment errors and unsupported-file notices are processing metadata, not product defects, " "unless the user explicitly asks to report them.\n" "Return the completed issue only.\n\n" + system ) messages = [{"role": "system", "content": system}] if history and self.brain: messages.extend(self.brain._format_history(history)) messages.append({"role": "user", "content": str(query)}) return messages def run(self, query: str, context: Dict[str, Any] = None) -> Optional[str]: fallback = build_issue_fallback(query) if not self.brain: return fallback messages = self._messages(query, context or {}) try: response = self.brain._call_llm( messages, stream=False, temperature=0.15, max_tokens=4096, enable_thinking=True, reasoning_budget=2048, timeout_seconds=50, ) draft = str(response.choices[0].message.content or "").strip() valid, failures = validate_issue_response(draft, query) if valid: return draft repair_messages = messages + [ {"role": "assistant", "content": draft}, { "role": "user", "content": ( "The draft failed these checks: " + ", ".join(failures) + ". " "Regenerate it using only facts in my request. Return the complete corrected issue." ), }, ] repaired = self.brain._call_llm( repair_messages, stream=False, temperature=0.1, max_tokens=4096, enable_thinking=False, timeout_seconds=25, ) repaired_text = str(repaired.choices[0].message.content or "").strip() repaired_valid, _ = validate_issue_response(repaired_text, query) return repaired_text if repaired_valid else fallback except Exception as exc: print(f"Issue agent fallback activated: {exc}") return fallback