Spaces:
Running
Running
| """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 | |