import ast import gc import json import os import re import threading import time from collections import Counter from typing import Any, Literal from fastapi import FastAPI from pydantic import BaseModel, ConfigDict, Field, ValidationError AGENT_VERSION = "3.0" SERVICE_NAME = "stitch-qa-code-agent" SOURCE_AGENT_ID = "source-quality-analyst" SOURCE_DISPLAY_NAME = "Source Quality Intelligence Analyst" REPAIR_AGENT_ID = "repair-assurance-analyst" REPAIR_DISPLAY_NAME = "Repair Assurance Intelligence Analyst" LEGACY_AGENT_NAME = "code-agent" MAX_AI_SOURCE_FINDINGS = 8 MAX_REPAIR_CONTRACTS = 8 MAX_SOURCE_CONTEXT_FILES = 4 MAX_SOURCE_CONTEXT_CHARS = 12000 MAX_SOURCE_AI_CONTEXT_CHARS = 18000 MAX_SOURCE_AI_FILES = 6 MAX_REPAIR_RUNTIME_GROUPS = 6 MAX_REPAIR_SOURCE_FINDINGS = 12 MAX_REPAIR_EVIDENCE_PER_GROUP = 6 SEVERITY_ORDER = { "CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3, "INFO": 4, } RISK_ORDER = { "UNKNOWN": -1, "NONE": 0, "INFO": 1, "LOW": 2, "MEDIUM": 3, "HIGH": 4, "CRITICAL": 5, } SOURCE_AI_SCHEMA = { "type": "object", "properties": { "c": {"type": "string", "enum": ["HIGH", "MEDIUM", "LOW"]}, "x": { "type": "array", "maxItems": MAX_AI_SOURCE_FINDINGS, "items": { "type": "object", "properties": { "f": {"type": "string", "minLength": 1, "maxLength": 500}, "l": {"type": "integer", "minimum": 1}, "g": {"type": "string", "minLength": 2, "maxLength": 80}, "s": {"type": "string", "enum": ["HIGH", "MEDIUM", "LOW", "INFO"]}, "t": {"type": "string", "minLength": 6, "maxLength": 140}, "i": {"type": "string", "minLength": 8, "maxLength": 220}, "r": {"type": "string", "minLength": 8, "maxLength": 240}, "k": {"type": "boolean"}, "kr": {"type": "string", "maxLength": 160}, }, "required": ["f", "l", "g", "s", "t", "i", "r", "k", "kr"], "additionalProperties": False, }, }, }, "required": ["c", "x"], "additionalProperties": False, } REPAIR_AI_SCHEMA = { "type": "object", "properties": { "c": {"type": "string", "enum": ["HIGH", "MEDIUM", "LOW"]}, "x": { "type": "array", "minItems": 1, "maxItems": MAX_REPAIR_CONTRACTS, "items": { "type": "object", "properties": { "r": {"type": "string", "minLength": 1, "maxLength": 80}, "f": { "type": "array", "minItems": 1, "maxItems": 16, "uniqueItems": True, "items": {"type": "string", "minLength": 1, "maxLength": 80}, }, "t": { "type": "array", "maxItems": MAX_SOURCE_CONTEXT_FILES, "uniqueItems": True, "items": {"type": "string", "minLength": 1, "maxLength": 500}, }, "y": {"type": "string", "minLength": 4, "maxLength": 180}, "i": {"type": "string", "minLength": 8, "maxLength": 220}, "a": {"type": "string", "minLength": 8, "maxLength": 420}, "e": {"type": "string", "minLength": 8, "maxLength": 220}, "v": {"type": "string", "minLength": 8, "maxLength": 240}, "g": {"type": "string", "minLength": 8, "maxLength": 240}, "k": {"type": "boolean"}, "kr": {"type": "string", "maxLength": 160}, }, "required": ["r", "f", "t", "y", "i", "a", "e", "v", "g", "k", "kr"], "additionalProperties": False, }, }, }, "required": ["c", "x"], "additionalProperties": False, } def env_bool(name, default): value = os.getenv(name) if value is None: return bool(default) return value.strip().lower() in {"1", "true", "yes", "on"} def clean_text(value, limit=800): text = " ".join(str(value or "").split()).strip() if len(text) <= limit: return text return text[: limit - 3].rstrip() + "..." def normalize_list(value): if isinstance(value, list): return value if value is None: return [] return [value] def normalize_risk(value): risk = str(value or "UNKNOWN").upper() return risk if risk in RISK_ORDER else "UNKNOWN" def highest_risk(*values): normalized = [normalize_risk(value) for value in values] return max(normalized, key=lambda value: RISK_ORDER.get(value, -1), default="UNKNOWN") class SourceFileInput(BaseModel): model_config = ConfigDict(extra="ignore") path: str = Field(min_length=1, max_length=500) content: str = Field(default="", max_length=60000) truncated: bool = False original_chars: int = 0 class SourceReviewRequest(BaseModel): model_config = ConfigDict(extra="ignore") project_type: str = Field(min_length=1, max_length=200) has_tests: bool = False files: list[SourceFileInput] = Field(default_factory=list, max_length=100) discovered_files_count: int = 0 submitted_files_count: int = 0 submitted_chars: int = 0 truncated_files_count: int = 0 omitted_files_count: int = 0 read_error_files_count: int = 0 class SourceFinding(BaseModel): model_config = ConfigDict(extra="ignore") id: str file_path: str | None = None line: int | None = None category: str severity: Literal["CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"] confidence: Literal["HIGH", "MEDIUM", "LOW"] title: str evidence: str impact: str recommendation: str detector: str = "deterministic" current_knowledge_required: bool = False current_knowledge_reason: str | None = None class SourceReviewResponse(BaseModel): model_config = ConfigDict(extra="ignore") agent_id: str display_name: str agent_version: str agent: str mode: str model: str | None = None status: str confidence: str summary: str risk_level: str release_recommendation: str reviewed_files_count: int findings_count: int severity_summary: dict[str, int] category_summary: dict[str, int] findings: list[SourceFinding] = Field(default_factory=list) warnings: list[str] = Field(default_factory=list) limitations: list[str] = Field(default_factory=list) verification: str current_knowledge_required: bool = False current_knowledge_reason: str | None = None llm_metrics: dict[str, Any] | None = None llm_error: str | None = None class RepairContractInput(BaseModel): model_config = ConfigDict(extra="ignore") contract_id: str = Field(min_length=1, max_length=80) finding_refs: list[str] = Field(default_factory=list, max_length=16) title: str | None = Field(default=None, max_length=220) priority: str | None = Field(default=None, max_length=20) priority_reason: str | None = Field(default=None, max_length=500) repair_objective: str | None = Field(default=None, max_length=700) repair_strategy: str | None = Field(default=None, max_length=1000) change_boundary: str | None = Field(default=None, max_length=700) protected_behavior: str | None = Field(default=None, max_length=700) side_effect_risk: str | None = Field(default=None, max_length=40) verification: str | None = Field(default=None, max_length=700) done_condition: str | None = Field(default=None, max_length=700) status: str | None = Field(default=None, max_length=80) class RepairAssuranceRequest(BaseModel): model_config = ConfigDict(extra="ignore") project_type: str = Field(min_length=1, max_length=200) command: str = Field(default="", max_length=2000) success: bool | None = None exit_code: int | None = None failure_type: str | None = Field(default=None, max_length=200) help_message: str | None = Field(default=None, max_length=4000) runtime_analysis: dict[str, Any] | None = None source_review: dict[str, Any] | None = None repair_plan: dict[str, Any] | None = None source_files: list[SourceFileInput] = Field(default_factory=list, max_length=MAX_SOURCE_CONTEXT_FILES) file_path: str | None = Field(default=None, max_length=500) code_snippet: str | None = Field(default=None, max_length=16000) error_log: str | None = Field(default=None, max_length=16000) root_cause: str | None = Field(default=None, max_length=4000) repair_summary: str | None = Field(default=None, max_length=4000) class RepairAssuranceItem(BaseModel): model_config = ConfigDict(extra="ignore") guidance_id: str repair_contract_ref: str finding_refs: list[str] target_files: list[str] target_symbols: list[str] implementation_intent: str code_level_approach: str change_boundary: str protected_behavior: str side_effect_considerations: str targeted_verification: str regression_verification: str suggested_patch: str | None = None patch_validation_status: str current_knowledge_required: bool = False current_knowledge_reason: str | None = None status: str = "PENDING_IMPLEMENTATION" class RepairAssuranceResponse(BaseModel): model_config = ConfigDict(extra="ignore") agent_id: str display_name: str agent_version: str agent: str mode: str model: str | None = None status: str confidence: str risk_level: str auto_apply: bool = False summary: str guidance: list[RepairAssuranceItem] = Field(default_factory=list) suggested_patch: str | None = None verification: str current_knowledge_required: bool = False current_knowledge_reason: str | None = None evidence_lineage: list[dict[str, Any]] = Field(default_factory=list) shadow_validation_status: str = "NOT_RUN" warnings: list[str] = Field(default_factory=list) limitations: list[str] = Field(default_factory=list) llm_metrics: dict[str, Any] | None = None llm_error: str | None = None class ModelSourceFinding(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) file_path: str = Field(alias="f", min_length=1, max_length=500) line: int = Field(alias="l", ge=1) category: str = Field(alias="g", min_length=2, max_length=80) severity: Literal["HIGH", "MEDIUM", "LOW", "INFO"] = Field(alias="s") title: str = Field(alias="t", min_length=6, max_length=140) impact: str = Field(alias="i", min_length=8, max_length=220) recommendation: str = Field(alias="r", min_length=8, max_length=240) current_knowledge_required: bool = Field(alias="k") current_knowledge_reason: str = Field(alias="kr", max_length=160) class ModelSourcePlan(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) confidence: Literal["HIGH", "MEDIUM", "LOW"] = Field(alias="c") findings: list[ModelSourceFinding] = Field(alias="x", max_length=MAX_AI_SOURCE_FINDINGS) class ModelRepairGuidance(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) contract_ref: str = Field(alias="r", min_length=1, max_length=80) finding_refs: list[str] = Field(alias="f", min_length=1, max_length=16) target_files: list[str] = Field(alias="t", max_length=MAX_SOURCE_CONTEXT_FILES) target_symbols_text: str = Field(alias="y", min_length=4, max_length=180) implementation_intent: str = Field(alias="i", min_length=8, max_length=220) code_level_approach: str = Field(alias="a", min_length=8, max_length=420) side_effect_considerations: str = Field(alias="e", min_length=8, max_length=220) targeted_verification: str = Field(alias="v", min_length=8, max_length=240) regression_verification: str = Field(alias="g", min_length=8, max_length=240) current_knowledge_required: bool = Field(alias="k") current_knowledge_reason: str = Field(alias="kr", max_length=160) class ModelRepairPlan(BaseModel): model_config = ConfigDict(extra="forbid", populate_by_name=True) confidence: Literal["HIGH", "MEDIUM", "LOW"] = Field(alias="c") guidance: list[ModelRepairGuidance] = Field(alias="x", min_length=1, max_length=MAX_REPAIR_CONTRACTS) class ModelService: def __init__(self): self.model_repo = os.getenv( "HF_MODEL_REPO", "Qwen/Qwen2.5-Coder-1.5B-Instruct-GGUF", ).strip() self.model_file = os.getenv( "HF_MODEL_FILE", "qwen2.5-coder-1.5b-instruct-q4_k_m.gguf", ).strip() self.model_revision = os.getenv( "HF_MODEL_REVISION", "f86cb2c1fa58255f8052cc32aeede1b7482d4361", ).strip() self.quantization = os.getenv("MODEL_QUANTIZATION", "Q4_K_M").strip() self.primary_model = os.getenv( "HF_MODEL", f"{self.model_repo}:{self.quantization}", ).strip() self.cache_dir = os.path.expanduser( os.getenv("MODEL_CACHE_DIR", "~/.cache/huggingface/hub").strip() ) self.enabled = env_bool("LLM_ENABLED", True) self.local_files_only = env_bool("HF_LOCAL_FILES_ONLY", False) self.context_tokens = max(2048, int(os.getenv("MODEL_CONTEXT_TOKENS", "4096"))) self.batch_tokens = max(64, min(self.context_tokens, int(os.getenv("MODEL_BATCH_TOKENS", "128")))) self.source_new_tokens = max(128, int(os.getenv("SOURCE_MODEL_MAX_NEW_TOKENS", "320"))) self.repair_new_tokens = max(160, int(os.getenv("REPAIR_MODEL_MAX_NEW_TOKENS", "320"))) self.max_input_tokens = max(1024, int(os.getenv("MODEL_MAX_INPUT_TOKENS", "3000"))) response_budget = max(self.source_new_tokens, self.repair_new_tokens) context_safe_input = max(1024, self.context_tokens - response_budget - 128) self.max_input_tokens = min(self.max_input_tokens, context_safe_input) self.repair_max_input_tokens = max(1024, int(os.getenv("REPAIR_MODEL_MAX_INPUT_TOKENS", "2200"))) self.repair_max_input_tokens = min(self.repair_max_input_tokens, self.max_input_tokens) self.max_generation_seconds = max(20.0, float(os.getenv("MODEL_MAX_GENERATION_SECONDS", "120"))) self.threads = max(1, int(os.getenv("MODEL_THREADS", "2"))) self.threads_batch = max(1, int(os.getenv("MODEL_THREADS_BATCH", str(self.threads)))) self.temperature = max(0.0, float(os.getenv("MODEL_TEMPERATURE", "0.1"))) self.top_p = min(1.0, max(0.01, float(os.getenv("MODEL_TOP_P", "0.9")))) self.seed = int(os.getenv("MODEL_SEED", "17")) self.use_mmap = env_bool("MODEL_USE_MMAP", True) self.model = None self.model_path = None self.model_name = None self.load_error = None self.load_seconds = None self.last_generation = None self.load_lock = threading.Lock() self.generation_lock = threading.Lock() def _download_model(self): try: from huggingface_hub import hf_hub_download except ImportError as error: raise RuntimeError("huggingface_hub is required for the Agent 3 GGUF backend.") from error return hf_hub_download( repo_id=self.model_repo, filename=self.model_file, revision=self.model_revision, cache_dir=self.cache_dir, local_files_only=self.local_files_only, ) def _load_model(self): try: from llama_cpp import Llama except ImportError as error: raise RuntimeError("llama-cpp-python is required for the Agent 3 GGUF backend.") from error model_path = self._download_model() model = Llama( model_path=model_path, n_ctx=self.context_tokens, n_batch=self.batch_tokens, n_threads=self.threads, n_threads_batch=self.threads_batch, n_gpu_layers=0, seed=self.seed, use_mmap=self.use_mmap, use_mlock=False, verbose=False, ) return model_path, model def load(self): if not self.enabled: raise RuntimeError("LLM inference is disabled.") if self.model is not None: return self.model with self.load_lock: if self.model is not None: return self.model started = time.monotonic() try: model_path, model = self._load_model() self.model_path = model_path self.model = model self.model_name = self.primary_model self.load_error = None self.load_seconds = round(time.monotonic() - started, 3) return model except Exception as error: self.model = None self.model_path = None self.model_name = None self.load_error = repr(error) self.load_seconds = round(time.monotonic() - started, 3) gc.collect() raise RuntimeError(self.load_error) from error def _count_tokens(self, model, value): serialized = json.dumps(value, ensure_ascii=False, separators=(",", ":")).encode("utf-8") try: return len(model.tokenize(serialized, add_bos=False, special=True)) except TypeError: return len(model.tokenize(serialized, add_bos=False)) def _count_text_tokens(self, model, text): if not text: return 0 value = str(text).encode("utf-8") try: return len(model.tokenize(value, add_bos=False, special=True)) except TypeError: return len(model.tokenize(value, add_bos=False)) def _json_complete(self, text): candidate = str(text or "").strip() if not candidate.startswith("{") or not candidate.endswith("}"): return False try: json.loads(candidate) return True except json.JSONDecodeError: return False def _extract_json_object(self, text): candidate = str(text or "").strip() if not candidate: return None if candidate.startswith("```"): candidate = re.sub(r"^```(?:json)?\s*", "", candidate, flags=re.IGNORECASE).strip() candidate = re.sub(r"\s*```$", "", candidate).strip() try: json.loads(candidate) return candidate except json.JSONDecodeError: pass start = candidate.find("{") if start < 0: return None depth = 0 in_string = False escape = False for index in range(start, len(candidate)): char = candidate[index] if in_string: if escape: escape = False elif char == "\\": escape = True elif char == '"': in_string = False continue if char == '"': in_string = True elif char == "{": depth += 1 elif char == "}": depth -= 1 if depth == 0: extracted = candidate[start : index + 1].strip() try: json.loads(extracted) return extracted except json.JSONDecodeError: return None return None def _reset_inference_state(self, model): reset = getattr(model, "reset", None) if callable(reset): reset() def generate_json(self, messages, schema, max_new_tokens, task, input_token_limit=None, enforce_schema=True): self.last_generation = None model = self.load() with self.generation_lock: self._reset_inference_state(model) try: input_tokens = self._count_tokens(model, messages) token_limit = self.max_input_tokens if input_token_limit is None else min(self.max_input_tokens, max(1024, int(input_token_limit))) if input_tokens > token_limit: raise RuntimeError( f"Agent 3 model input contains approximately {input_tokens} tokens, exceeding the configured limit of {token_limit}." ) started = time.monotonic() parts = [] finish_reason = None timed_out = False first_content_seconds = None response_format = {"type": "json_object", "schema": schema} if enforce_schema else {"type": "json_object"} stream = model.create_chat_completion( messages=messages, response_format=response_format, max_tokens=max_new_tokens, temperature=self.temperature, top_p=self.top_p, seed=self.seed, stream=True, ) try: for chunk in stream: elapsed = time.monotonic() - started choices = chunk.get("choices") or [] if choices: choice = choices[0] delta = choice.get("delta") or {} content = delta.get("content") if content: if first_content_seconds is None: first_content_seconds = elapsed parts.append(str(content)) if self._json_complete("".join(parts)): finish_reason = "json_complete" break if choice.get("finish_reason"): finish_reason = str(choice.get("finish_reason")) if elapsed >= self.max_generation_seconds and not finish_reason: timed_out = True break finally: close = getattr(stream, "close", None) if callable(close): close() elapsed = time.monotonic() - started text = "".join(parts).strip() completion_tokens = self._count_text_tokens(model, text) tokens_per_second = ( round(completion_tokens / elapsed, 3) if elapsed > 0 and completion_tokens else 0.0 ) self.last_generation = { "backend": "llama.cpp", "task": task, "quantization": self.quantization, "input_tokens_approx": input_tokens, "completion_tokens": completion_tokens, "elapsed_seconds": round(elapsed, 3), "first_content_seconds": ( round(first_content_seconds, 3) if first_content_seconds is not None else None ), "tokens_per_second": tokens_per_second, "finish_reason": finish_reason, "timed_out": timed_out, "max_generation_seconds": self.max_generation_seconds, "max_new_tokens": max_new_tokens, } if timed_out: raise RuntimeError( "Agent 3 model exceeded the configured generation time limit " f"(generated_tokens={completion_tokens}, elapsed_seconds={elapsed:.3f}, " f"tokens_per_second={tokens_per_second:.3f})." ) if not text: raise RuntimeError("Agent 3 model returned an empty response.") json_text = self._extract_json_object(text) if not json_text: raise RuntimeError("Agent 3 JSON-constrained generation returned invalid JSON.") if json_text != text: self.last_generation["json_recovered"] = True text = json_text else: self.last_generation["json_recovered"] = False if finish_reason == "length": raise RuntimeError( "Agent 3 model reached the output token limit before completing the structured response." ) return text finally: self._reset_inference_state(model) def status(self): if not self.enabled: state = "disabled" elif self.model is not None: state = "loaded" elif self.load_error: state = "load_failed" else: state = "not_loaded" return { "enabled": self.enabled, "loaded": self.model is not None, "state": state, "backend": "llama.cpp", "configured_model": self.primary_model, "active_model": self.model_name, "model_repo": self.model_repo, "model_file": self.model_file, "model_revision": self.model_revision, "quantization": self.quantization, "load_error": self.load_error, "load_seconds": self.load_seconds, "context_tokens": self.context_tokens, "max_input_tokens": self.max_input_tokens, "source_max_new_tokens": self.source_new_tokens, "repair_max_new_tokens": self.repair_new_tokens, "repair_max_input_tokens": self.repair_max_input_tokens, "threads": self.threads, "last_generation": self.last_generation, } model_service = ModelService() app = FastAPI( title="Stitch QA Code Quality Intelligence", version=AGENT_VERSION, ) SECRET_NAME_PATTERN = re.compile( r"(?:password|passwd|secret|token|api[_-]?key|private[_-]?key|access[_-]?key)", re.IGNORECASE, ) PLACEHOLDER_SECRET_VALUES = { "", "change-me", "changeme", "example", "placeholder", "secret", "password", "token", "your-secret", "your-token", } CURRENT_KNOWLEDGE_INDICATORS = ( "version", "dependency", "deprecated", "deprecation", "compatibility", "cve", "security advisory", "vendor", "framework release", "jdk", "plugin", ) ENVIRONMENT_INDICATORS = ( "maven is not installed", "maven wrapper", "environment", "build-tool availability", "build tool availability", "path", "command timeout", "python is not installed", "pytest is not installed", ) TESTING_ONLY_INDICATORS = ( "no automated tests", "missing tests", "test files and test configuration", "add focused tests", "test suite", ) def dotted_name(node): if isinstance(node, ast.Name): return node.id if isinstance(node, ast.Attribute): parent = dotted_name(node.value) return f"{parent}.{node.attr}" if parent else node.attr return "" def line_evidence(content, line_number, fallback): lines = content.splitlines() if line_number and 0 < line_number <= len(lines): value = lines[line_number - 1].strip() if value: return value[:240] return fallback def is_placeholder_secret(value): normalized = str(value).strip().lower() if normalized in PLACEHOLDER_SECRET_VALUES: return True return any(marker in normalized for marker in ("${", "{{", "env.", "process.env", "os.getenv", "system.getenv")) def assignment_target_names(node): names = [] if isinstance(node, ast.Name): names.append(node.id) elif isinstance(node, (ast.Tuple, ast.List)): for item in node.elts: names.extend(assignment_target_names(item)) elif isinstance(node, ast.Attribute): names.append(node.attr) return names def keyword_value(call, keyword_name): for keyword in call.keywords: if keyword.arg == keyword_name: return keyword.value return None def is_true_literal(node): return isinstance(node, ast.Constant) and node.value is True def is_false_literal(node): return isinstance(node, ast.Constant) and node.value is False def node_contains_name(node, names): targets = set(names) return any(isinstance(child, ast.Name) and child.id in targets for child in ast.walk(node)) def active_guard_names(body, line_number): names = set() for statement in body: if getattr(statement, "lineno", 0) >= line_number: break if isinstance(statement, ast.If): test_text = ast.unparse(statement.test).lower() if hasattr(ast, "unparse") else "" for name_node in ast.walk(statement.test): if isinstance(name_node, ast.Name): if any(token in test_text for token in ("== 0", "!= 0", "len(", "not ", "<= 0")): names.add(name_node.id) return names def division_denominator_risk(node, function_node): denominator = node.right line = getattr(node, "lineno", None) if line is None: return None guarded = active_guard_names(function_node.body, line) if isinstance(denominator, ast.Constant): return None if isinstance(denominator, ast.Name): if denominator.id in guarded: return None return ( "validation", "MEDIUM", "Potential unguarded divisor", f"Function `{function_node.name}` divides by `{denominator.id}` without an earlier visible zero guard.", "A zero value can raise ZeroDivisionError or violate the caller's expected error contract.", "Validate the divisor before division and raise the project-specific controlled exception while preserving valid inputs.", ) if isinstance(denominator, ast.Call) and dotted_name(denominator.func) == "len" and denominator.args: arg = denominator.args[0] collection_name = arg.id if isinstance(arg, ast.Name) else "collection" if collection_name in guarded: return None return ( "validation", "MEDIUM", "Potential empty-collection divisor", f"Function `{function_node.name}` divides by len({collection_name}) without an earlier visible empty-collection guard.", "An empty collection can create a zero divisor and terminate the path unexpectedly.", "Validate that the collection is not empty before division and preserve the documented error behavior for invalid input.", ) return None def add_finding( findings, file_path, line, category, severity, confidence, title, evidence, impact, recommendation, detector="deterministic", current_knowledge_required=False, current_knowledge_reason=None, ): findings.append( { "file_path": file_path, "line": line, "category": clean_text(category, 80).lower(), "severity": severity, "confidence": confidence, "title": clean_text(title, 180), "evidence": clean_text(evidence, 300), "impact": clean_text(impact, 500), "recommendation": clean_text(recommendation, 600), "detector": detector, "current_knowledge_required": bool(current_knowledge_required), "current_knowledge_reason": clean_text(current_knowledge_reason, 240) or None, } ) def analyze_python_source(source_file, findings, warnings): path = source_file.path content = source_file.content try: tree = ast.parse(content, filename=path) except SyntaxError as error: if source_file.truncated: warnings.append(f"{path} could not be fully parsed because the submitted content was truncated.") return add_finding( findings, path, error.lineno, "correctness", "HIGH", "HIGH", "Python syntax error", error.msg, "The module cannot be imported or executed successfully.", "Correct the syntax error and rerun Stitch QA before deployment.", ) return for node in ast.walk(tree): if isinstance(node, (ast.Assign, ast.AnnAssign)): value = node.value targets = node.targets if isinstance(node, ast.Assign) else [node.target] names = [] for target in targets: names.extend(assignment_target_names(target)) if ( isinstance(value, ast.Constant) and isinstance(value.value, str) and len(value.value.strip()) >= 6 and any(SECRET_NAME_PATTERN.search(name) for name in names) and not is_placeholder_secret(value.value) ): add_finding( findings, path, getattr(node, "lineno", None), "security", "HIGH", "HIGH", "Possible hardcoded credential", "A credential-like variable is assigned a literal string value.", "Hardcoded credentials can be exposed through source control, logs, or package distribution.", "Move the value to a secret manager or environment variable and rotate any exposed credential.", ) if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): defaults = list(node.args.defaults) + [item for item in node.args.kw_defaults if item is not None] if any(isinstance(default, (ast.List, ast.Dict, ast.Set)) for default in defaults): add_finding( findings, path, getattr(node, "lineno", None), "reliability", "MEDIUM", "HIGH", "Mutable function default", f"Function `{node.name}` uses a mutable default argument.", "State can leak between calls and create difficult-to-reproduce defects.", "Use None as the default and create the mutable object inside the function.", ) for inner in ast.walk(node): if isinstance(inner, ast.BinOp) and isinstance(inner.op, ast.Div): risk = division_denominator_risk(inner, node) if risk: category, severity, title, evidence_text, impact, recommendation = risk add_finding( findings, path, getattr(inner, "lineno", None), category, severity, "MEDIUM", title, evidence_text, impact, recommendation, ) if isinstance(node, ast.ExceptHandler): if node.type is None: add_finding( findings, path, getattr(node, "lineno", None), "reliability", "MEDIUM", "HIGH", "Bare exception handler", line_evidence(content, getattr(node, "lineno", None), "A bare except handler was detected."), "SystemExit, KeyboardInterrupt, and unexpected defects can be swallowed.", "Catch only expected exception types and preserve actionable error context.", ) if node.body and all(isinstance(item, ast.Pass) for item in node.body): add_finding( findings, path, getattr(node, "lineno", None), "error-handling", "MEDIUM", "HIGH", "Exception silently ignored", line_evidence(content, getattr(node, "lineno", None), "An exception handler contains only pass."), "Failures can be hidden and leave the application in an unknown state.", "Handle the exception explicitly or log and re-raise it with safe context.", ) if not isinstance(node, ast.Call): continue function_name = dotted_name(node.func) line = getattr(node, "lineno", None) evidence = line_evidence(content, line, function_name) if function_name in {"eval", "exec", "builtins.eval", "builtins.exec"}: add_finding( findings, path, line, "security", "HIGH", "HIGH", "Dynamic code execution", evidence, "Untrusted input can lead to arbitrary code execution.", "Remove dynamic execution or replace it with a strict parser and an allowlisted operation set.", ) if function_name == "os.system": add_finding( findings, path, line, "security", "HIGH", "HIGH", "Shell command execution through os.system", evidence, "User-controlled values can create operating-system command injection.", "Use subprocess with an argument list, shell disabled, validated inputs, and least privilege.", ) if function_name.startswith("subprocess.") and is_true_literal(keyword_value(node, "shell")): add_finding( findings, path, line, "security", "HIGH", "HIGH", "Subprocess executed with shell enabled", evidence, "Shell interpretation increases command-injection risk.", "Disable shell execution and pass validated command arguments as a list.", ) if function_name in {"pickle.load", "pickle.loads"}: add_finding( findings, path, line, "security", "HIGH", "HIGH", "Unsafe pickle deserialization", evidence, "Loading attacker-controlled pickle data can execute arbitrary code.", "Use a non-executable serialization format and validate data before processing it.", ) if function_name == "yaml.load": loader = keyword_value(node, "Loader") loader_name = dotted_name(loader) if loader is not None else "" if loader_name not in {"yaml.SafeLoader", "yaml.CSafeLoader", "SafeLoader", "CSafeLoader"}: add_finding( findings, path, line, "security", "HIGH", "HIGH", "Unsafe YAML loading", evidence, "Unsafe YAML constructors can instantiate arbitrary Python objects.", "Use yaml.safe_load or explicitly select SafeLoader.", ) if function_name.startswith("requests.") and is_false_literal(keyword_value(node, "verify")): add_finding( findings, path, line, "security", "HIGH", "HIGH", "TLS certificate verification disabled", evidence, "Network traffic can be intercepted by an attacker presenting an untrusted certificate.", "Enable certificate verification and configure a trusted CA bundle when required.", ) if function_name in {"hashlib.md5", "hashlib.sha1"}: add_finding( findings, path, line, "security", "MEDIUM", "MEDIUM", "Weak cryptographic hash", evidence, "MD5 and SHA-1 are unsuitable for security-sensitive integrity or credential protection.", "Use a modern hash such as SHA-256, or a password-specific algorithm such as Argon2 or bcrypt.", ) if function_name.endswith(".run") and is_true_literal(keyword_value(node, "debug")): add_finding( findings, path, line, "security", "MEDIUM", "MEDIUM", "Debug mode explicitly enabled", evidence, "Production debug mode can expose sensitive stack traces or interactive debugging features.", "Disable debug mode in deployment and control it through environment-specific configuration.", ) if len(content.splitlines()) > 800: add_finding( findings, path, 1, "maintainability", "LOW", "MEDIUM", "Large source file", f"The submitted file contains {len(content.splitlines())} lines.", "Very large modules are harder to review, test, and maintain safely.", "Split unrelated responsibilities into smaller cohesive modules while preserving behavior.", ) def analyze_java_source(source_file, findings): path = source_file.path content = source_file.content lines = content.splitlines() for index, raw_line in enumerate(lines, start=1): line = raw_line.strip() secret_match = re.search( r'(?i)\b(?:String\s+)?([A-Za-z0-9_]*(?:password|passwd|secret|token|apiKey|api_key|privateKey|accessKey)[A-Za-z0-9_]*)\s*=\s*"([^"]{6,})"', line, ) if secret_match and not is_placeholder_secret(secret_match.group(2)): add_finding( findings, path, index, "security", "HIGH", "HIGH", "Possible hardcoded credential", "A credential-like Java variable is assigned a literal string value.", "The credential can be exposed through source control, build artifacts, or logs.", "Load the value from a protected secret store or environment variable and rotate exposed credentials.", ) if "Runtime.getRuntime().exec" in line: add_finding( findings, path, index, "security", "HIGH", "HIGH", "Runtime command execution", line[:240], "Unvalidated command content can lead to operating-system command injection.", "Use a fixed command with validated arguments and avoid passing user-controlled shell content.", ) if "new ProcessBuilder" in line: add_finding( findings, path, index, "security", "MEDIUM", "MEDIUM", "External process creation requires validation", line[:240], "Untrusted process arguments can create command execution or privilege risks.", "Use allowlisted commands, validated argument arrays, controlled working directories, and least privilege.", ) if re.search(r"\bexecute(?:Query|Update)?\s*\([^)]*\+", line): add_finding( findings, path, index, "security", "HIGH", "MEDIUM", "Possible SQL query concatenation", line[:240], "Concatenated untrusted values can allow SQL injection.", "Use PreparedStatement parameters or an ORM parameter-binding API.", ) if re.search(r'MessageDigest\.getInstance\(\s*"(?:MD5|SHA-1)"', line, re.IGNORECASE): add_finding( findings, path, index, "security", "MEDIUM", "HIGH", "Weak cryptographic hash", line[:240], "MD5 and SHA-1 are unsuitable for security-sensitive integrity or credential protection.", "Use SHA-256 or stronger, and use a password-specific algorithm for password storage.", ) if ".printStackTrace()" in line: add_finding( findings, path, index, "error-handling", "LOW", "HIGH", "Stack trace written directly", line[:240], "Raw stack traces can expose implementation details and bypass structured logging.", "Use the project logger with controlled context and avoid exposing sensitive details to clients.", ) if "System.out.print" in line or "System.err.print" in line: add_finding( findings, path, index, "maintainability", "LOW", "HIGH", "Direct console output in application code", line[:240], "Console output is difficult to classify, filter, and correlate in production.", "Use structured application logging with an appropriate log level.", ) if re.search(r"@CrossOrigin\s*\([^)]*\*", line) or re.search(r"allowedOrigins?\s*\([^)]*\*", line, re.IGNORECASE): add_finding( findings, path, index, "security", "MEDIUM", "HIGH", "Wildcard cross-origin access", line[:240], "Unrestricted origins can expose APIs to untrusted web applications.", "Allow only explicitly trusted origins and review credential-sharing settings.", ) if re.search(r"HostnameVerifier.*->\s*true", line) or "ALLOW_ALL_HOSTNAME_VERIFIER" in line: add_finding( findings, path, index, "security", "CRITICAL", "HIGH", "TLS hostname verification bypass", line[:240], "An attacker can impersonate a remote service even when TLS is used.", "Remove the permissive verifier and use the platform default hostname and certificate validation.", ) for match in re.finditer(r"catch\s*\([^)]*\)\s*\{\s*\}", content, re.DOTALL): line = content.count("\n", 0, match.start()) + 1 add_finding( findings, path, line, "error-handling", "MEDIUM", "HIGH", "Empty Java catch block", "A catch block contains no handling logic.", "The application can hide failures and continue with invalid state.", "Handle the expected exception, preserve safe diagnostic context, or rethrow it appropriately.", ) if len(lines) > 800: add_finding( findings, path, 1, "maintainability", "LOW", "MEDIUM", "Large source file", f"The submitted file contains {len(lines)} lines.", "Very large classes are harder to review, test, and maintain safely.", "Separate unrelated responsibilities into smaller cohesive classes while preserving behavior.", ) def finding_may_need_current_knowledge(finding): text = " ".join( clean_text(finding.get(key), 500).lower() for key in ("title", "category", "evidence", "impact", "recommendation") ) return any(indicator in text for indicator in CURRENT_KNOWLEDGE_INDICATORS) def build_source_line_index(request): index = {} for source_file in request.files: lines = source_file.content.splitlines() index[source_file.path] = lines return index def model_source_payload(request, deterministic_findings): file_payload = [] char_budget = MAX_SOURCE_AI_CONTEXT_CHARS used = 0 for source_file in request.files[:MAX_SOURCE_AI_FILES]: remaining = char_budget - used if remaining <= 0: break content = source_file.content[:remaining] used += len(content) numbered = "\n".join( f"{index}: {line}" for index, line in enumerate(content.splitlines(), start=1) ) file_payload.append( { "path": source_file.path, "truncated": source_file.truncated, "numbered_source": numbered, } ) return { "project_type": request.project_type, "has_tests": request.has_tests, "files": file_payload, "deterministic_findings": [ { "file": item.get("file_path"), "line": item.get("line"), "category": item.get("category"), "severity": item.get("severity"), "title": item.get("title"), } for item in deterministic_findings[:20] ], } def build_source_messages(request, deterministic_findings): system = ( "You are Stitch QA's Source Quality Intelligence Analyst. Review only the supplied Python or Java/Maven application source. " "Deterministic findings are locked facts. You may add contextual source findings only when the exact supplied file and line visibly support the issue. " "Never invent files, lines, runtime results, dependency versions, vulnerabilities, framework behavior, or business requirements. " "Do not repeat deterministic findings. Prefer correctness, reliability, validation, security, performance, maintainability, and testability risks that require contextual reasoning. " "AI-only findings must be conservative: use HIGH only for clearly dangerous source behavior visible at the cited line; otherwise use MEDIUM, LOW, or INFO. " "Set current knowledge true only when implementation decisions genuinely depend on current version, vendor, compatibility, deprecation, or security-advisory facts. " "Return only JSON matching the required schema." ) return [ {"role": "system", "content": system}, { "role": "user", "content": json.dumps(model_source_payload(request, deterministic_findings), ensure_ascii=False, separators=(",", ":")), }, ] def parse_source_model_output(text): try: data = json.loads(str(text or "").strip()) except json.JSONDecodeError as error: raise ValueError("The Source Quality model response was not valid JSON.") from error try: return ModelSourcePlan.model_validate(data) except ValidationError as error: raise ValueError(f"The Source Quality model response failed schema validation: {error}") from error def validate_source_ai_findings(plan, request, deterministic_findings): line_index = build_source_line_index(request) deterministic_keys = { ( str(item.get("file_path") or ""), int(item.get("line") or 0), str(item.get("title") or "").lower(), ) for item in deterministic_findings } accepted = [] for item in plan.findings: lines = line_index.get(item.file_path) if lines is None or item.line > len(lines): continue line_text = lines[item.line - 1].strip() if not line_text: continue title = clean_text(item.title, 180) key = (item.file_path, item.line, title.lower()) if key in deterministic_keys: continue severity = item.severity if severity == "HIGH": high_signal = any( token in line_text.lower() for token in ( "eval(", "exec(", "os.system", "shell=true", "runtime.getruntime().exec", "allow_all_hostname_verifier", "hostnameverifier", "verify=false", "pickle.load", ) ) if not high_signal: severity = "MEDIUM" current_required = bool(item.current_knowledge_required) and any( indicator in " ".join( [item.category.lower(), title.lower(), item.impact.lower(), item.recommendation.lower()] ) for indicator in CURRENT_KNOWLEDGE_INDICATORS ) add_finding( accepted, item.file_path, item.line, clean_text(item.category, 80), severity, "MEDIUM", title, line_text[:240], item.impact, item.recommendation, detector="ai-contextual-validated", current_knowledge_required=current_required, current_knowledge_reason=(item.current_knowledge_reason if current_required else None), ) return accepted def deduplicate_findings(findings): unique = {} for finding in findings: key = ( finding.get("file_path"), finding.get("line"), clean_text(finding.get("title"), 180).lower(), ) existing = unique.get(key) if existing is None: unique[key] = finding continue if SEVERITY_ORDER.get(finding.get("severity"), 99) < SEVERITY_ORDER.get(existing.get("severity"), 99): unique[key] = finding ordered = sorted( unique.values(), key=lambda item: ( SEVERITY_ORDER.get(item.get("severity"), 99), str(item.get("file_path") or ""), item.get("line") or 0, item.get("title") or "", ), ) for index, finding in enumerate(ordered, start=1): finding["id"] = f"SQ-SRC-{index:03d}" return ordered def build_severity_summary(findings): summary = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0, "INFO": 0} for finding in findings: severity = finding.get("severity", "INFO") summary[severity] = summary.get(severity, 0) + 1 return summary def build_category_summary(findings): return dict(sorted(Counter(finding.get("category", "other") for finding in findings).items())) def build_risk_level(severity_summary): for severity in ("CRITICAL", "HIGH", "MEDIUM", "LOW", "INFO"): if severity_summary.get(severity, 0): return severity return "LOW" def build_release_recommendation(risk_level, partial): if risk_level in {"CRITICAL", "HIGH"}: return "BLOCK_RELEASE" if risk_level == "MEDIUM": return "REVIEW_REQUIRED" if partial: return "MANUAL_REVIEW_REQUIRED" return "READY_WITH_CAUTION" def build_source_summary(request, findings, risk_level, release_recommendation): if not findings: return ( f"Reviewed {len(request.files)} application source file(s) and found no blocking issue with the implemented deterministic and validated contextual checks. " "This does not prove the absence of defects, vulnerabilities, dependency risks, business-rule errors, or runtime failures." ) severity_summary = build_severity_summary(findings) return ( f"Reviewed {len(request.files)} application source file(s) and reported {len(findings)} evidence-grounded finding(s). " f"Highest source risk is {risk_level}. Critical: {severity_summary.get('CRITICAL', 0)}, High: {severity_summary.get('HIGH', 0)}, " f"Medium: {severity_summary.get('MEDIUM', 0)}, Low: {severity_summary.get('LOW', 0)}. Release recommendation: {release_recommendation}." ) def review_source_request(request): deterministic = [] warnings = [] limitations = [ "Source Quality Intelligence combines targeted deterministic checks with bounded AI reasoning; it does not prove the absence of defects or vulnerabilities.", "AI contextual reasoning uses a bounded source window for CPU reliability; deterministic checks still evaluate every submitted supported source file.", "Dependency vulnerabilities, infrastructure configuration, business-rule correctness, and runtime behavior require separate evidence.", ] for source_file in request.files: suffix = os.path.splitext(source_file.path.lower())[1] if suffix == ".py": analyze_python_source(source_file, deterministic, warnings) elif suffix == ".java": analyze_java_source(source_file, deterministic) else: warnings.append(f"Unsupported source extension skipped by Source Quality Intelligence: {source_file.path}") if source_file.truncated: warnings.append(f"{source_file.path} was truncated before review; findings may not cover the entire file.") if not request.has_tests: add_finding( deterministic, None, None, "testing", "MEDIUM", "HIGH", "No automated tests detected", "The project scan did not detect a compatible automated test suite.", "Source review alone cannot verify runtime behavior, integrations, regressions, or business rules.", "Add focused automated tests for critical flows, validation, error paths, and security-sensitive behavior.", ) ai_findings = [] llm_error = None ai_confidence = None mode = "deterministic-validated" if model_service.enabled and request.files: try: messages = build_source_messages(request, deterministic) text = model_service.generate_json( messages, SOURCE_AI_SCHEMA, model_service.source_new_tokens, "source-quality", ) plan = parse_source_model_output(text) ai_findings = validate_source_ai_findings(plan, request, deterministic) ai_confidence = plan.confidence mode = "hybrid-ai-validated" if ai_findings else "hybrid-ai-no-additional-findings" except Exception as error: llm_error = repr(error) mode = "deterministic-fallback" findings = deduplicate_findings([*deterministic, *ai_findings]) severity_summary = build_severity_summary(findings) category_summary = build_category_summary(findings) risk_level = build_risk_level(severity_summary) partial = any( ( request.truncated_files_count, request.omitted_files_count, request.read_error_files_count, ) ) release_recommendation = build_release_recommendation(risk_level, partial) if request.omitted_files_count: limitations.append(f"{request.omitted_files_count} discovered source file(s) were omitted by client-side review limits.") if request.read_error_files_count: limitations.append(f"{request.read_error_files_count} source file(s) could not be read by the CLI.") if request.truncated_files_count: limitations.append(f"{request.truncated_files_count} source file(s) were partially reviewed because of size limits.") current_required = any(bool(item.get("current_knowledge_required")) for item in findings) current_reason = next( (item.get("current_knowledge_reason") for item in findings if item.get("current_knowledge_required") and item.get("current_knowledge_reason")), None, ) confidence = "HIGH" if partial: confidence = "MEDIUM" elif ai_confidence in {"HIGH", "MEDIUM", "LOW"} and ai_findings: confidence = ai_confidence elif llm_error: confidence = "MEDIUM" if findings else "LOW" result = { "agent_id": SOURCE_AGENT_ID, "display_name": SOURCE_DISPLAY_NAME, "agent_version": AGENT_VERSION, "agent": LEGACY_AGENT_NAME, "mode": mode, "model": model_service.model_name or model_service.primary_model if model_service.enabled else None, "status": "PARTIAL" if partial else "COMPLETED", "confidence": confidence, "summary": build_source_summary(request, findings, risk_level, release_recommendation), "risk_level": risk_level, "release_recommendation": release_recommendation, "reviewed_files_count": len(request.files), "findings_count": len(findings), "severity_summary": severity_summary, "category_summary": category_summary, "findings": findings, "warnings": list(dict.fromkeys(warnings)), "limitations": limitations, "verification": "Validate each confirmed source finding, preserve unaffected behavior, run targeted checks where available, then rerun the complete supported QA workflow.", "current_knowledge_required": current_required, "current_knowledge_reason": current_reason, "llm_metrics": model_service.last_generation, "llm_error": llm_error, } return SourceReviewResponse.model_validate(result).model_dump() def contract_text(contract): return " ".join( clean_text(value, 800).lower() for value in ( contract.title, contract.priority_reason, contract.repair_objective, contract.repair_strategy, contract.change_boundary, contract.protected_behavior, contract.verification, contract.done_condition, ) ) def contract_is_environment_only(contract): text = contract_text(contract) explicit_no_source = any( phrase in text for phrase in ( "do not modify application source", "keep application source code outside this repair", "environment or build-tool availability", "environment needed to execute", ) ) environment_signal = any(indicator in text for indicator in ENVIRONMENT_INDICATORS) return explicit_no_source and environment_signal def contract_is_testing_only(contract): text = contract_text(contract) test_signal = any(indicator in text for indicator in TESTING_ONLY_INDICATORS) source_exclusion = any( phrase in text for phrase in ( "do not modify application behavior", "keep production code unchanged", "preserve existing application behavior", ) ) return test_signal and source_exclusion def repair_contracts_from_request(request): source = request.repair_plan or {} contracts = [] for item in normalize_list(source.get("stitch_repair_contracts"))[:MAX_REPAIR_CONTRACTS]: if not isinstance(item, dict): continue try: contracts.append(RepairContractInput.model_validate(item)) except ValidationError: continue return contracts def source_finding_map(request): source = request.source_review or {} result = {} for item in normalize_list(source.get("findings")): if isinstance(item, dict) and item.get("id"): result[str(item.get("id"))] = item return result def runtime_group_map(request): analysis = request.runtime_analysis or {} result = {} for item in normalize_list(analysis.get("root_cause_groups")): if isinstance(item, dict) and item.get("group_id"): result[str(item.get("group_id"))] = item return result def known_source_files(request): return {item.path: item for item in request.source_files} def contract_linked_files(contract, request): available = known_source_files(request) source_map = source_finding_map(request) runtime_map = runtime_group_map(request) linked = [] def add(path): value = clean_text(path, 500) if value in available and value not in linked: linked.append(value) for ref in contract.finding_refs: source_finding = source_map.get(ref) if source_finding: add(source_finding.get("file_path")) runtime_group = runtime_map.get(ref) if runtime_group: for evidence in normalize_list(runtime_group.get("evidence")): if isinstance(evidence, dict): add(evidence.get("application_file")) return linked def source_symbol_ranges(source_file): if source_file is None: return [] content = source_file.content path = source_file.path.lower() ranges = [] if path.endswith(".py"): try: tree = ast.parse(content) except SyntaxError: return [] for node in tree.body: if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)): start = getattr(node, "lineno", None) end = getattr(node, "end_lineno", None) or start if start: ranges.append((node.name, start, end)) elif path.endswith(".java"): lines = content.splitlines() class_match = re.search(r"\bclass\s+([A-Za-z_$][A-Za-z0-9_$]*)", content) if class_match: line = content.count("\n", 0, class_match.start()) + 1 ranges.append((class_match.group(1), line, len(lines) or line)) method_pattern = re.compile( r"\b(?:public|protected|private)?\s*(?:static\s+)?(?:final\s+)?(?:[A-Za-z_$][A-Za-z0-9_$<>\[\], ?.]*)\s+([A-Za-z_$][A-Za-z0-9_$]*)\s*\(" ) for match in method_pattern.finditer(content): name = match.group(1) line = content.count("\n", 0, match.start()) + 1 if name not in {item[0] for item in ranges}: ranges.append((name, line, line)) return ranges[:24] def extract_symbols(source_file): return [name for name, _, _ in source_symbol_ranges(source_file)][:12] def symbols_at_lines(source_file, lines): if source_file is None: return [] wanted = [int(line) for line in lines if isinstance(line, int) or str(line).isdigit()] if not wanted: return [] ranges = source_symbol_ranges(source_file) matched = [] for line in wanted: for name, start, end in ranges: if start <= line <= end and name not in matched: matched.append(name) return matched[:12] def contract_linked_locations(contract, request): source_map = source_finding_map(request) runtime_map = runtime_group_map(request) locations = [] def add(path, line): cleaned_path = clean_text(path, 500) if not cleaned_path: return try: cleaned_line = int(line) if line is not None else None except (TypeError, ValueError): cleaned_line = None item = (cleaned_path, cleaned_line) if item not in locations: locations.append(item) for ref in contract.finding_refs: source_finding = source_map.get(ref) if source_finding: add(source_finding.get("file_path"), source_finding.get("line")) runtime_group = runtime_map.get(ref) if runtime_group: for evidence in normalize_list(runtime_group.get("evidence")): if isinstance(evidence, dict): add(evidence.get("application_file"), evidence.get("application_line")) return locations def contract_linked_symbols(contract, request, linked_files=None): files = known_source_files(request) linked_files = linked_files or contract_linked_files(contract, request) locations = contract_linked_locations(contract, request) symbols = [] for path in linked_files: lines = [line for file_path, line in locations if file_path == path and line] for symbol in symbols_at_lines(files.get(path), lines): if symbol not in symbols: symbols.append(symbol) if symbols: return symbols[:12] for path in linked_files: for symbol in extract_symbols(files.get(path)): if symbol not in symbols: symbols.append(symbol) return symbols[:12] def compact_repair_contract(contract): return { "id": contract.contract_id, "refs": list(dict.fromkeys(contract.finding_refs)), "priority": clean_text(contract.priority, 20), "objective": clean_text(contract.repair_objective, 260), "strategy": clean_text(contract.repair_strategy, 360), "boundary": clean_text(contract.change_boundary, 300), "protect": clean_text(contract.protected_behavior, 300), "risk": clean_text(contract.side_effect_risk, 40), "verification": clean_text(contract.verification, 320), "done": clean_text(contract.done_condition, 260), } def compact_runtime_group(group): evidence = [] for item in normalize_list(group.get("evidence"))[:MAX_REPAIR_EVIDENCE_PER_GROUP]: if not isinstance(item, dict): continue evidence.append( { "test": clean_text(item.get("test_name"), 180), "test_file": clean_text(item.get("test_file"), 300), "test_line": item.get("test_line"), "app_file": clean_text(item.get("application_file"), 300), "app_line": item.get("application_line"), "expected": clean_text(item.get("expected"), 120), "actual": clean_text(item.get("actual"), 120), "exception": clean_text(item.get("exception_type"), 120), "message": clean_text(item.get("exception_message"), 220), } ) return { "id": clean_text(group.get("group_id"), 80), "category": clean_text(group.get("category"), 80), "origin": clean_text(group.get("failure_origin"), 80), "cause": clean_text(group.get("root_cause"), 420), "impact": clean_text(group.get("runtime_impact"), 300), "action": clean_text(group.get("required_action"), 360), "tests": [clean_text(value, 180) for value in normalize_list(group.get("affected_tests"))[:12]], "evidence": evidence, } def compact_source_finding(item): return { "id": clean_text(item.get("id"), 80), "file": clean_text(item.get("file_path"), 300), "line": item.get("line"), "category": clean_text(item.get("category"), 80), "severity": clean_text(item.get("severity"), 40), "title": clean_text(item.get("title"), 160), "evidence": clean_text(item.get("evidence"), 260), "impact": clean_text(item.get("impact"), 260), "recommendation": clean_text(item.get("recommendation"), 300), } def repair_payload(request, contracts): linked_refs = { ref for contract in contracts for ref in contract.finding_refs if ref } runtime_groups = [] for group in normalize_list((request.runtime_analysis or {}).get("root_cause_groups")): if not isinstance(group, dict): continue group_id = str(group.get("group_id") or "") if group_id in linked_refs: runtime_groups.append(compact_runtime_group(group)) if len(runtime_groups) >= MAX_REPAIR_RUNTIME_GROUPS: break source_findings = [] for item in normalize_list((request.source_review or {}).get("findings")): if not isinstance(item, dict): continue finding_id = str(item.get("id") or "") if finding_id in linked_refs: source_findings.append(compact_source_finding(item)) if len(source_findings) >= MAX_REPAIR_SOURCE_FINDINGS: break source_files = [] used = 0 for item in request.source_files[:MAX_SOURCE_CONTEXT_FILES]: remaining = MAX_SOURCE_CONTEXT_CHARS - used if remaining <= 0: break content = item.content[:remaining] if not content: continue used += len(content) source_files.append( { "path": item.path, "truncated": item.truncated or len(content) < len(item.content), "symbols": extract_symbols(item), "content": content, } ) return { "project": request.project_type, "execution": { "command": clean_text(request.command, 300), "success": request.success, "exit_code": request.exit_code, "failure_type": clean_text(request.failure_type, 120), }, "runtime_groups": runtime_groups, "source_findings": source_findings, "contracts": [compact_repair_contract(contract) for contract in contracts], "source_files": source_files, } def build_repair_messages(request, contracts): system = ( "You are Stitch QA's Repair Assurance Intelligence Analyst. Agent 2 contracts are authoritative. " "Explain how to implement each supplied contract safely in the supplied Python or Java/Maven source. " "Return exactly one item for every contract ID and exactly the same finding refs. " "Never broaden the contract boundary, override protected behavior, invent files, symbols, tests, runtime facts, dependency versions, or vulnerabilities. " "Target files may only come from source_files. Use concise implementation guidance, not patches. " "Testing-only contracts may have no target production file. Environment-only contracts are handled outside this model. " "Targeted verification confirms the repair objective; regression verification protects unaffected behavior. " "Set current knowledge true only when current version, vendor, compatibility, deprecation, dependency, or security-advisory documentation is genuinely required. " "Return only JSON matching the schema." ) return [ {"role": "system", "content": system}, { "role": "user", "content": json.dumps(repair_payload(request, contracts), ensure_ascii=False, separators=(",", ":")), }, ] def parse_repair_model_output(text): try: data = json.loads(str(text or "").strip()) except json.JSONDecodeError as error: raise ValueError("The Repair Assurance model response was not valid JSON.") from error try: return ModelRepairPlan.model_validate(data) except ValidationError as error: raise ValueError(f"The Repair Assurance model response failed schema validation: {error}") from error def grounded_current_knowledge_for_contract(contract): text = contract_text(contract) if not text: return False explicit_current_signals = ( "current trusted documentation", "current documentation", "current framework", "current vendor", "current security advisory", "security advisory", "cve", "deprecation", "deprecated", "vendor guidance", "vendor documentation", "release notes", "migration guide", "version compatibility", "dependency version", "package version", "plugin version", "jdk version", "framework release", "breaking change", ) if any(signal in text for signal in explicit_current_signals): return True contextual_pairs = ( ("dependency", "version"), ("dependency", "compatibility"), ("dependency", "upgrade"), ("dependency", "release"), ("framework", "version"), ("framework", "compatibility"), ("plugin", "compatibility"), ("jdk", "compatibility"), ) return any(first in text and second in text for first, second in contextual_pairs) def deterministic_repair_item(contract, request, index, reason=None): linked_files = contract_linked_files(contract, request) source_files = known_source_files(request) symbols = contract_linked_symbols(contract, request, linked_files) testing_only = contract_is_testing_only(contract) current_required = grounded_current_knowledge_for_contract(contract) if testing_only: intent = clean_text(contract.repair_objective, 220) or "Add evidence-driven automated test coverage without changing production behavior." approach = ( "Add focused automated tests using the detected project conventions. Cover the application behavior represented by the repair contract and evidence-backed edge or error paths. " "Keep production source unchanged unless a separate validated finding requires a source repair." ) target_files = [] target_symbols = symbols else: intent = clean_text(contract.repair_objective, 220) or "Implement the repair objective within the approved boundary." approach = clean_text(contract.repair_strategy, 420) or "Apply the smallest evidence-backed implementation change within the approved repair boundary." target_files = linked_files target_symbols = symbols verification = clean_text(contract.verification, 240) or "Confirm the referenced repair objective, then run the relevant regression suite." return { "guidance_id": f"RAI-{index:03d}", "repair_contract_ref": contract.contract_id, "finding_refs": list(dict.fromkeys(contract.finding_refs)), "target_files": target_files, "target_symbols": target_symbols, "implementation_intent": intent, "code_level_approach": approach, "change_boundary": clean_text(contract.change_boundary, 400) or "Do not broaden the repair beyond the referenced evidence.", "protected_behavior": clean_text(contract.protected_behavior, 400) or "Preserve unaffected behavior represented by the existing evidence.", "side_effect_considerations": ( clean_text(reason, 220) if reason else f"Respect the Agent 2 side-effect risk classification ({clean_text(contract.side_effect_risk, 40) or 'UNKNOWN'}) and avoid unrelated refactoring." ), "targeted_verification": verification, "regression_verification": "After targeted confirmation, run the complete available test suite or project checks and confirm no new failure is introduced.", "suggested_patch": None, "patch_validation_status": "NOT_GENERATED", "current_knowledge_required": current_required, "current_knowledge_reason": ( "This repair contract contains version, dependency, compatibility, deprecation, vendor, JDK, plugin, or security-advisory context that should be checked against current trusted documentation." if current_required else None ), "status": "PENDING_IMPLEMENTATION", } def environment_repair_item(contract, index): return { "guidance_id": f"RAI-{index:03d}", "repair_contract_ref": contract.contract_id, "finding_refs": list(dict.fromkeys(contract.finding_refs)), "target_files": [], "target_symbols": [], "implementation_intent": clean_text(contract.repair_objective, 220) or "Restore the execution environment required for QA.", "code_level_approach": "No application source change is justified by this repair contract. Resolve only the environment, build-tool, or wrapper availability problem described by Agent 2.", "change_boundary": clean_text(contract.change_boundary, 400) or "Environment and build-tool setup only; application source remains outside this repair.", "protected_behavior": clean_text(contract.protected_behavior, 400) or "Preserve application source and test behavior while restoring the execution environment.", "side_effect_considerations": "Treat any later compile or test failure as a separate evidence-backed finding instead of mixing it into this environment repair.", "targeted_verification": clean_text(contract.verification, 240) or "Confirm the validated command can start and rerun Stitch QA.", "regression_verification": "After the environment blocker is removed, use the resulting complete build or test run as the runtime regression evidence.", "suggested_patch": None, "patch_validation_status": "NOT_APPLICABLE", "current_knowledge_required": grounded_current_knowledge_for_contract(contract), "current_knowledge_reason": None, "status": "NO_CODE_CHANGE_REQUIRED", } def validate_repair_plan(plan, request, contracts): contract_map = {contract.contract_id: contract for contract in contracts} seen = [] available_files = set(known_source_files(request)) normalized = [] for item in plan.guidance: contract = contract_map.get(item.contract_ref) if contract is None: raise ValueError(f"Repair Assurance referenced unknown contract {item.contract_ref}.") seen.append(item.contract_ref) allowed_refs = set(contract.finding_refs) if set(item.finding_refs) != allowed_refs: raise ValueError(f"Repair Assurance finding coverage does not match {item.contract_ref}.") linked_files = contract_linked_files(contract, request) target_files = [path for path in item.target_files if path in available_files] if linked_files: target_files = [path for path in target_files if path in set(linked_files)] or linked_files if contract_is_testing_only(contract): target_files = [] target_symbols = contract_linked_symbols(contract, request, target_files) current_required = grounded_current_knowledge_for_contract(contract) normalized.append( { "contract": contract, "target_files": target_files, "target_symbols": target_symbols, "implementation_intent": clean_text(item.implementation_intent, 260), "code_level_approach": clean_text(item.code_level_approach, 520), "side_effect_considerations": clean_text(item.side_effect_considerations, 280), "targeted_verification": clean_text(item.targeted_verification, 300), "regression_verification": clean_text(item.regression_verification, 300), "suggested_patch": None, "current_knowledge_required": current_required, "current_knowledge_reason": ( clean_text(item.current_knowledge_reason, 220) if current_required and clean_text(item.current_knowledge_reason, 220) else ( "Current trusted documentation is required because the repair contract contains version, dependency, compatibility, deprecation, vendor, JDK, plugin, or security-advisory context." if current_required else None ) ), } ) if set(seen) != set(contract_map) or len(seen) != len(set(seen)): raise ValueError("Repair Assurance did not cover every submitted repair contract exactly once.") return normalized def build_evidence_lineage(contract): return { "repair_contract_ref": contract.contract_id, "finding_refs": list(dict.fromkeys(contract.finding_refs)), "guidance_status": "LINKED", } def repair_assurance_request(request): contracts = repair_contracts_from_request(request) if not contracts: result = { "agent_id": REPAIR_AGENT_ID, "display_name": REPAIR_DISPLAY_NAME, "agent_version": AGENT_VERSION, "agent": LEGACY_AGENT_NAME, "mode": "evidence-validated", "model": None, "status": "NOT_REQUIRED", "confidence": "HIGH", "risk_level": "NONE", "auto_apply": False, "summary": "No Stitch Repair Contract was supplied, so no code-level repair assurance is required.", "guidance": [], "suggested_patch": None, "verification": "No repair implementation is pending from Agent 2.", "current_knowledge_required": False, "current_knowledge_reason": None, "evidence_lineage": [], "shadow_validation_status": "NOT_RUN", "warnings": [], "limitations": ["Repair Assurance only operates on supplied Stitch Repair Contracts."], "llm_metrics": model_service.last_generation, "llm_error": None, } return RepairAssuranceResponse.model_validate(result).model_dump() environment_contracts = [contract for contract in contracts if contract_is_environment_only(contract)] code_contracts = [contract for contract in contracts if contract not in environment_contracts] guidance = [] llm_error = None confidence = "HIGH" mode = "deterministic-validated" for index, contract in enumerate(environment_contracts, start=1): guidance.append(environment_repair_item(contract, index)) model_guidance = [] if code_contracts and model_service.enabled: try: messages = build_repair_messages(request, code_contracts) text = model_service.generate_json( messages, REPAIR_AI_SCHEMA, model_service.repair_new_tokens, "repair-assurance", model_service.repair_max_input_tokens, False, ) plan = parse_repair_model_output(text) model_guidance = validate_repair_plan(plan, request, code_contracts) confidence = plan.confidence mode = "ai-reasoned-validated" except Exception as error: llm_error = repr(error) confidence = "MEDIUM" mode = "deterministic-fallback" start_index = len(guidance) + 1 if model_guidance: for offset, item in enumerate(model_guidance): contract = item["contract"] guidance.append( { "guidance_id": f"RAI-{start_index + offset:03d}", "repair_contract_ref": contract.contract_id, "finding_refs": list(dict.fromkeys(contract.finding_refs)), "target_files": item["target_files"], "target_symbols": item["target_symbols"], "implementation_intent": item["implementation_intent"] or clean_text(contract.repair_objective, 260), "code_level_approach": item["code_level_approach"] or clean_text(contract.repair_strategy, 520), "change_boundary": clean_text(contract.change_boundary, 400) or "Do not broaden the repair beyond the Agent 2 contract.", "protected_behavior": clean_text(contract.protected_behavior, 400) or "Preserve unaffected behavior represented by existing evidence.", "side_effect_considerations": item["side_effect_considerations"], "targeted_verification": item["targeted_verification"] or clean_text(contract.verification, 300), "regression_verification": item["regression_verification"], "suggested_patch": None, "patch_validation_status": "NOT_GENERATED", "current_knowledge_required": item["current_knowledge_required"], "current_knowledge_reason": item["current_knowledge_reason"], "status": "PENDING_IMPLEMENTATION", } ) else: for offset, contract in enumerate(code_contracts): guidance.append( deterministic_repair_item( contract, request, start_index + offset, "AI reasoning was unavailable or rejected, so this guidance preserves the Agent 2 contract without expanding its scope.", ) ) order = {contract.contract_id: index for index, contract in enumerate(contracts)} guidance.sort(key=lambda item: order.get(item["repair_contract_ref"], 999)) for index, item in enumerate(guidance, start=1): item["guidance_id"] = f"RAI-{index:03d}" current_required = any(item.get("current_knowledge_required") for item in guidance) current_reason = next( (item.get("current_knowledge_reason") for item in guidance if item.get("current_knowledge_required") and item.get("current_knowledge_reason")), None, ) risk = highest_risk( *[clean_text(contract.side_effect_risk, 40) or "UNKNOWN" for contract in contracts] ) code_change_count = sum(1 for item in guidance if item.get("status") != "NO_CODE_CHANGE_REQUIRED") if code_change_count: summary = ( f"Prepared {len(guidance)} contract-bound repair assurance item(s) for {len(contracts)} Stitch Repair Contract(s). " "Implementation guidance is evidence-linked, preserves Agent 2 boundaries, and does not modify the project automatically." ) else: summary = ( f"Reviewed {len(contracts)} Stitch Repair Contract(s); all supplied contracts are environment-only, so no application source change is justified." ) result = { "agent_id": REPAIR_AGENT_ID, "display_name": REPAIR_DISPLAY_NAME, "agent_version": AGENT_VERSION, "agent": LEGACY_AGENT_NAME, "mode": mode, "model": model_service.model_name or model_service.primary_model if model_service.enabled and code_contracts else None, "status": "COMPLETED", "confidence": confidence, "risk_level": risk, "auto_apply": False, "summary": summary, "guidance": guidance, "suggested_patch": None, "verification": "Complete targeted confirmation for each repair assurance item, then run the relevant full regression workflow before considering the repair verified.", "current_knowledge_required": current_required, "current_knowledge_reason": current_reason, "evidence_lineage": [build_evidence_lineage(contract) for contract in contracts], "shadow_validation_status": "NOT_RUN", "warnings": [], "limitations": [ "Repair Assurance provides implementation guidance but never modifies the original project automatically.", "Shadow Repair Validation is not executed in this service response; Repair Assurance does not emit an unvalidated patch.", "Current external facts are not fetched inside this agent; cases marked current_knowledge_required need trusted documentation before implementation.", ], "llm_metrics": model_service.last_generation, "llm_error": llm_error, } return RepairAssuranceResponse.model_validate(result).model_dump() def legacy_repair_assurance(request): if request.failure_type in { "MAVEN_NOT_AVAILABLE", "MAVEN_WRAPPER_NOT_AVAILABLE", "MAVEN_WRAPPER_NOT_EXECUTABLE", "PYTHON_NOT_AVAILABLE", "PYTEST_NOT_AVAILABLE", "COMMAND_TIMEOUT", }: summary = "The supplied evidence describes an execution-environment blocker rather than a confirmed application-source defect. No application source change is justified." status = "NO_CODE_CHANGE_REQUIRED" risk = "LOW" elif request.success is True or request.exit_code == 0: summary = "The supplied execution context does not establish a blocking application-source repair target. Preserve the passing state and do not apply an ungrounded source change." status = "NOT_REQUIRED" risk = "LOW" else: summary = "Legacy single-context guidance is available only as a compatibility path. Use Agent 2 Stitch Repair Contracts for professional Repair Assurance guidance." status = "LEGACY_CONTEXT_ONLY" risk = "MEDIUM" result = { "agent_id": REPAIR_AGENT_ID, "display_name": REPAIR_DISPLAY_NAME, "agent_version": AGENT_VERSION, "agent": LEGACY_AGENT_NAME, "mode": "legacy-compatibility", "model": None, "status": status, "confidence": "MEDIUM", "risk_level": risk, "auto_apply": False, "summary": summary, "guidance": [], "suggested_patch": None, "verification": "Run the supported Stitch QA workflow with Agent 2 repair planning before applying any code change.", "current_knowledge_required": False, "current_knowledge_reason": None, "evidence_lineage": [], "shadow_validation_status": "NOT_RUN", "warnings": [], "limitations": ["This compatibility path does not replace contract-bound Repair Assurance."], "llm_metrics": None, "llm_error": None, } return RepairAssuranceResponse.model_validate(result).model_dump() @app.get("/") def health_check(): status = model_service.status() return { "service": SERVICE_NAME, "status": "running", "agent_version": AGENT_VERSION, "model": status, "capabilities": [ { "agent_id": SOURCE_AGENT_ID, "display_name": SOURCE_DISPLAY_NAME, "endpoint": "/review-source", }, { "agent_id": REPAIR_AGENT_ID, "display_name": REPAIR_DISPLAY_NAME, "endpoint": "/assure-repair", }, ], "compatibility_endpoints": ["/suggest-code-fix"], "supported_project_types": ["Python Project", "Java Maven Project"], } @app.get("/ready") def readiness_check(): status = model_service.status() return { "ready": True, "source_quality_ready": True, "repair_assurance_ready": True, "llm_enabled": status["enabled"], "llm_loaded": status["loaded"], "llm_state": status["state"], "configured_model": status["configured_model"], "active_model": status["active_model"], "deterministic_fallback": True, } @app.post("/review-source", response_model=SourceReviewResponse) def review_source(request: SourceReviewRequest): return SourceReviewResponse.model_validate(review_source_request(request)) @app.post("/assure-repair", response_model=RepairAssuranceResponse) def assure_repair(request: RepairAssuranceRequest): if request.repair_plan: return RepairAssuranceResponse.model_validate(repair_assurance_request(request)) return RepairAssuranceResponse.model_validate(legacy_repair_assurance(request)) @app.post("/suggest-code-fix", response_model=RepairAssuranceResponse) def suggest_code_fix(request: RepairAssuranceRequest): if request.repair_plan: return RepairAssuranceResponse.model_validate(repair_assurance_request(request)) return RepairAssuranceResponse.model_validate(legacy_repair_assurance(request))