From cb530873ce75d6a1707a48f6406c28c833744211 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Mon, 13 Jul 2026 03:02:03 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 931 ++++++++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 787 insertions(+), 144 deletions(-) diff --git a/agent.py b/agent.py index eaeb730..22d573a 100644 --- a/agent.py +++ b/agent.py @@ -1,189 +1,832 @@ -"""support-to-fix-engineer agent. - -Starter stack: - - DeepAgents for tool-calling orchestration - - Caller-provided LLM credentials via ctx.llm - - A tiny model-call middleware hook you can replace with tracing, - routing, rate limits, or policy checks -""" from __future__ import annotations +import hashlib +import ipaddress import json -from pathlib import Path +import re +from datetime import UTC, datetime +from enum import Enum +from pathlib import PurePosixPath from typing import Any - -from pydantic import BaseModel +from urllib.parse import urlparse import a2a_pack as a2a from a2a_pack import ( A2AAgent, - LLMProvisioning, - {{ auth_type }}, + ConsumerSetup, + ConsumerSetupField, + EgressPolicy, + NoAuth, Pricing, + Resources, RunContext, WorkspaceAccess, WorkspaceMode, ) -from a2a_pack.context import LLMCreds +from a2a_pack.context import AgentEvent +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +APPROVAL_ACK = "I_APPROVE_THE_LISTED_STEPS" +OUTPUT_ROOT = "outputs/support-cases" +MAX_TEXT_CHARS = 24000 +MAX_EVIDENCE_ITEMS = 80 +MAX_PATHS = 40 class SupportToFixEngineerConfig(BaseModel): - pass + model_config = ConfigDict(extra="forbid") + + max_log_chars: int = Field(default=12000, ge=1000, le=50000) + default_dry_run: bool = True -SYSTEM_PROMPT = """\ -You are a compact tool-calling agent. - -Use the text_stats tool when the user asks about text, counts, summaries, -or anything where exact length/word numbers would help. Mention tool results -briefly instead of dumping raw JSON. -""" - -RUNTIME_SKILLS_DIR = "support-to-fix-engineer/.deepagents/skills/" -DEEPAGENTS_RECURSION_LIMIT = 500 +class StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", use_enum_values=True) -class SupportToFixEngineer(A2AAgent[SupportToFixEngineerConfig, {{ auth_type }}]): +class Severity(str, Enum): + sev1 = "sev1" + sev2 = "sev2" + sev3 = "sev3" + sev4 = "sev4" + + +class EvidenceKind(str, Enum): + ticket = "ticket" + log = "log" + trace = "trace" + release = "release" + change = "change" + customer = "customer" + health_check = "health_check" + repository = "repository" + validation = "validation" + + +class DeliveryAction(str, Enum): + draft_branch = "draft_branch" + draft_commit = "draft_commit" + draft_pr = "draft_pr" + customer_summary = "customer_summary" + + +class EvidenceItem(StrictModel): + source_id: str = Field(min_length=1, max_length=160) + kind: EvidenceKind + summary: str = Field(min_length=1, max_length=1200) + timestamp: str | None = Field(default=None, max_length=80) + url: str | None = Field(default=None, max_length=2048) + excerpt: str | None = Field(default=None, max_length=4000) + + @field_validator("source_id", "summary", "excerpt") + @classmethod + def _redact_text_fields(cls, value: str | None) -> str | None: + if value is None: + return None + return redact_secrets(value)[:4000] + + @field_validator("url") + @classmethod + def _validate_url(cls, value: str | None) -> str | None: + if value in (None, ""): + return None + return validate_safe_url(value, allowed_hosts=()) + + +class HttpCheck(StrictModel): + name: str = Field(min_length=1, max_length=80) + url: str = Field(min_length=8, max_length=2048) + expected_status: int = Field(default=200, ge=100, le=599) + + +class RepoFileChange(StrictModel): + path: str = Field(min_length=1, max_length=240) + reason: str = Field(min_length=1, max_length=800) + proposed_content: str | None = Field(default=None, max_length=20000) + + @field_validator("path") + @classmethod + def _safe_path(cls, value: str) -> str: + return safe_repo_path(value) + + @field_validator("proposed_content") + @classmethod + def _redact_content(cls, value: str | None) -> str | None: + return redact_secrets(value) if value is not None else None + + +class TriageCaseInput(StrictModel): + case_id: str = Field(min_length=3, max_length=80, pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{2,79}$") + customer_id: str = Field(min_length=1, max_length=120) + tenant_id: str = Field(min_length=1, max_length=120) + release: str = Field(min_length=1, max_length=120) + ticket_text: str = Field(min_length=1, max_length=MAX_TEXT_CHARS) + evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_EVIDENCE_ITEMS) + affected_services: list[str] = Field(default_factory=list, max_length=20) + recent_changes: list[str] = Field(default_factory=list, max_length=40) + dry_run: bool = True + + @field_validator("ticket_text") + @classmethod + def _redact_ticket(cls, value: str) -> str: + return redact_secrets(value) + + @field_validator("affected_services", "recent_changes") + @classmethod + def _bounded_strings(cls, values: list[str]) -> list[str]: + return [redact_secrets(v.strip())[:500] for v in values if v.strip()] + + +class ReproduceIssueInput(StrictModel): + case_id: str = Field(min_length=3, max_length=80, pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{2,79}$") + redacted_symptom: str = Field(min_length=1, max_length=8000) + environment: dict[str, str] = Field(default_factory=dict, max_length=30) + observed_errors: list[str] = Field(default_factory=list, max_length=40) + candidate_files: list[str] = Field(default_factory=list, max_length=MAX_PATHS) + health_checks: list[HttpCheck] = Field(default_factory=list, max_length=20) + allowed_health_hosts: list[str] = Field(default_factory=list, max_length=30) + dry_run: bool = True + + @field_validator("candidate_files") + @classmethod + def _safe_paths(cls, values: list[str]) -> list[str]: + return [safe_repo_path(v) for v in values] + + @field_validator("redacted_symptom") + @classmethod + def _redact_symptom(cls, value: str) -> str: + return redact_secrets(value) + + +class ProposeFixInput(StrictModel): + case_id: str = Field(min_length=3, max_length=80, pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{2,79}$") + reproduction_summary: str = Field(min_length=1, max_length=12000) + hypotheses: list[str] = Field(default_factory=list, max_length=20) + supporting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_EVIDENCE_ITEMS) + contradicting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_EVIDENCE_ITEMS) + proposed_changes: list[RepoFileChange] = Field(default_factory=list, max_length=MAX_PATHS) + repository: str | None = Field(default=None, max_length=200) + dry_run: bool = True + + @field_validator("reproduction_summary") + @classmethod + def _redact_repro(cls, value: str) -> str: + return redact_secrets(value) + + +class ValidateFixInput(StrictModel): + case_id: str = Field(min_length=3, max_length=80, pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{2,79}$") + patch_digest: str = Field(min_length=16, max_length=128) + patch_diff: str = Field(min_length=1, max_length=60000) + test_selectors: list[str] = Field(default_factory=list, max_length=40) + lint_selectors: list[str] = Field(default_factory=list, max_length=20) + security_checks: list[str] = Field(default_factory=list, max_length=20) + dry_run: bool = True + + @field_validator("test_selectors", "lint_selectors", "security_checks") + @classmethod + def _selectors(cls, values: list[str]) -> list[str]: + cleaned: list[str] = [] + for value in values: + item = value.strip() + if not item: + continue + if re.search(r"[;&|`$<>]", item): + raise ValueError("selectors must not contain shell metacharacters") + cleaned.append(item[:180]) + return cleaned + + @field_validator("patch_diff") + @classmethod + def _patch_without_secrets(cls, value: str) -> str: + return redact_secrets(value) + + +class PrepareDeliveryInput(StrictModel): + case_id: str = Field(min_length=3, max_length=80, pattern=r"^[A-Za-z0-9][A-Za-z0-9._-]{2,79}$") + validation_digest: str = Field(min_length=16, max_length=128) + patch_digest: str = Field(min_length=16, max_length=128) + repository: str = Field(min_length=1, max_length=200) + target_branch: str = Field(default="main", min_length=1, max_length=120) + proposed_branch: str = Field(default="support-fix", min_length=1, max_length=120) + delivery_actions: list[DeliveryAction] = Field(default_factory=lambda: [DeliveryAction.customer_summary], max_length=4) + customer_summary_context: str = Field(default="", max_length=8000) + dry_run: bool = True + approval_acknowledgement: str | None = Field(default=None, max_length=80) + approval_plan_digest: str | None = Field(default=None, max_length=128) + + @field_validator("repository", "target_branch", "proposed_branch") + @classmethod + def _safe_refish(cls, value: str) -> str: + if re.search(r"[\s;&|`$<>]", value): + raise ValueError("repository and branch values must not contain whitespace or shell metacharacters") + return value + + +class ArtifactRecord(StrictModel): + path: str + artifact_uri: str | None = None + sha256: str + size_bytes: int = Field(ge=0) + + +class AuditRecord(StrictModel): + event: str + at: str + case_id: str + details: dict[str, Any] = Field(default_factory=dict) + + +class TriageCaseOutput(StrictModel): + status: str + case_id: str + severity: Severity + owner: str + redaction_count: int = Field(ge=0) + missing_setup: list[str] + allowed_tenant: bool + plan_digest: str + evidence_index: list[EvidenceItem] + artifacts: list[ArtifactRecord] + audit: list[AuditRecord] + warnings: list[str] + + +class ReproduceIssueOutput(StrictModel): + status: str + case_id: str + reproduction_digest: str + minimal_steps: list[str] + safe_health_checks: list[HttpCheck] + blocked_health_checks: list[str] + artifacts: list[ArtifactRecord] + audit: list[AuditRecord] + warnings: list[str] + + +class HypothesisRank(StrictModel): + hypothesis: str + rank: int = Field(ge=1) + confidence: float = Field(ge=0.0, le=1.0) + supporting_source_ids: list[str] + contradicting_source_ids: list[str] + + +class ProposeFixOutput(StrictModel): + status: str + case_id: str + repository: str | None + patch_digest: str + plan_digest: str + hypotheses: list[HypothesisRank] + scoped_paths: list[str] + artifacts: list[ArtifactRecord] + audit: list[AuditRecord] + warnings: list[str] + + +class ValidateFixOutput(StrictModel): + status: str + case_id: str + validation_digest: str + patch_digest: str + checks: dict[str, str] + artifacts: list[ArtifactRecord] + audit: list[AuditRecord] + warnings: list[str] + + +class PrepareDeliveryOutput(StrictModel): + status: str + case_id: str + plan_digest: str + approval_required: bool + approved: bool + executed_actions: list[str] + blocked_actions: list[str] + artifacts: list[ArtifactRecord] + audit: list[AuditRecord] + warnings: list[str] + + +class SupportToFixEngineer(A2AAgent[SupportToFixEngineerConfig, NoAuth]): name = "support-to-fix-engineer" - description = "Turns an authenticated support case into evidence-backed engineering fix artifacts with approval-gated repository delivery." + description = ( + "Turns authenticated support cases into evidence-backed engineering fix artifacts " + "with deterministic triage, reproduction, patch proposal, validation, and approval-gated delivery." + ) version = "0.1.0" config_model = SupportToFixEngineerConfig - auth_model = {{ auth_type }} + auth_model = NoAuth - # Hosted generated agents read the caller's saved LLM credential through - # ctx.llm. The platform may proxy that credential through LiteLLM, but agent - # code never reads provider keys, LiteLLM master keys, or OPENAI_API_KEY - # directly. - llm_provisioning = LLMProvisioning.PLATFORM pricing = Pricing( price_per_call_usd=0.0, - caller_pays_llm=True, - notes="Starter agent uses the caller's saved LLM credential via ctx.llm.", + caller_pays_llm=False, + notes="Deterministic policy and artifact generation; no LLM credential required.", ) + resources = Resources(cpu="1", memory="512Mi", max_runtime_seconds=600) workspace_access = WorkspaceAccess.dynamic( - max_files=64, + max_files=160, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), - require_reason=False, + require_reason=True, + deny_patterns=("**/.env", "**/.env.*", "**/*secret*", "**/*credential*", "**/id_rsa", "**/id_ed25519"), + require_human_approval=True, + max_total_size_bytes=50 * 1024 * 1024, ) - tools_used = ("deepagents", "langchain") + consumer_setup = ConsumerSetup.from_fields( + ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", input_type="url", required=False), + ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea token", required=False), + ConsumerSetupField.config("GITEA_ALLOWED_REPOS", label="Allowed repositories", required=False, description="Comma-separated owner/repo allowlist."), + ConsumerSetupField.config("KUBERNETES_API_URL", label="Kubernetes API URL", input_type="url", required=False), + ConsumerSetupField.secret("KUBERNETES_TOKEN", label="Kubernetes token", required=False), + ConsumerSetupField.config("ARGOCD_BASE_URL", label="Argo CD base URL", input_type="url", required=False), + ConsumerSetupField.secret("ARGOCD_TOKEN", label="Argo CD token", required=False), + ConsumerSetupField.config("OBSERVABILITY_BASE_URL", label="Observability base URL", input_type="url", required=False), + ConsumerSetupField.secret("OBSERVABILITY_TOKEN", label="Observability token", required=False), + ConsumerSetupField.config("TENANT_ALLOWLIST", label="Tenant allowlist", required=False, description="Comma-separated tenant IDs this agent may process."), + ConsumerSetupField.config("HEALTHCHECK_ALLOWED_HOSTS", label="HTTP health check hosts", required=False, description="Comma-separated exact host allowlist."), + ) + egress = EgressPolicy(allow_hosts=(), deny_internet_by_default=True) + tools_used = ("pydantic", "workspace") - @a2a.tool(description="Ask the starter DeepAgent to answer with tool calls when useful") - async def ask(self, ctx: RunContext[{{ auth_type }}], prompt: str) -> str: - creds = ctx.llm - await ctx.emit_progress(f"llm: {creds.model} via {creds.source}") - if not creds.api_key: - return ( - "LLM key required. Add an LLM credential in Settings > LLM " - "credentials before running this agent; for local --invoke " - "runs set AGENT_LLM_KEY." - ) - graph = self._build_deep_agent(ctx=ctx, creds=creds) - state = await graph.ainvoke( - {"messages": [{"role": "user", "content": prompt}]}, - config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}, + @a2a.tool(description="Intake, redact, correlate evidence, classify severity and ownership for a bounded support case", timeout_seconds=120, idempotent=True) + async def triage_case(self, ctx: RunContext[NoAuth], case: TriageCaseInput) -> TriageCaseOutput: + await emit_audit(ctx, case.case_id, "triage_started", {"dry_run": case.dry_run}) + warnings = policy_warnings(case.ticket_text, [e.summary for e in case.evidence]) + setup = missing_setup(ctx, include_tokens=False) + tenant_allowed = tenant_is_allowed(ctx, case.tenant_id) + severity = classify_severity(case.ticket_text, case.evidence) + owner = classify_owner(case.affected_services, case.ticket_text, case.recent_changes) + evidence_index = normalized_evidence(case.evidence, case.ticket_text, case.release, case.recent_changes) + payload = { + "case_id": case.case_id, + "customer_id": redact_secrets(case.customer_id), + "tenant_id": redact_secrets(case.tenant_id), + "release": redact_secrets(case.release), + "severity": severity.value, + "owner": owner, + "allowed_tenant": tenant_allowed, + "evidence": [item.model_dump(mode="json") for item in evidence_index], + } + digest = stable_digest(payload) + artifacts = [ + await write_case_file(ctx, case.case_id, "redacted_case.json", json.dumps(payload, indent=2, sort_keys=True)), + await write_case_file(ctx, case.case_id, "triage.json", json.dumps({**payload, "plan_digest": digest}, indent=2, sort_keys=True)), + ] + audit = [audit_record(case.case_id, "triage_completed", {"plan_digest": digest, "severity": severity.value})] + await emit_audit(ctx, case.case_id, "triage_completed", {"plan_digest": digest}) + status = "triaged" if tenant_allowed else "blocked_ungranted_tenant" + if setup: + warnings.append("Some integrations are not configured; analysis used caller-supplied evidence only.") + return TriageCaseOutput( + status=status, + case_id=case.case_id, + severity=severity, + owner=owner, + redaction_count=count_redactions(case.ticket_text), + missing_setup=setup, + allowed_tenant=tenant_allowed, + plan_digest=digest, + evidence_index=evidence_index, + artifacts=artifacts, + audit=audit, + warnings=warnings, ) - await ctx.emit_progress("deepagent finished") - return _last_message_text(state) - def _build_deep_agent( - self, - *, - ctx: RunContext[{{ auth_type }}], - creds: LLMCreds, - ) -> Any: - # Lazy imports keep `a2a card` usable before local dependencies are - # installed. `a2a deploy` installs requirements.txt during the build. - from a2a_pack.deepagents import create_a2a_deep_agent - from langchain.agents.middleware import wrap_model_call - from langchain_core.tools import tool + @a2a.tool(description="Create a deterministic minimal reproduction and validate HTTP health-check targets against SSRF protections", timeout_seconds=120, idempotent=True) + async def reproduce_issue(self, ctx: RunContext[NoAuth], reproduction: ReproduceIssueInput) -> ReproduceIssueOutput: + await emit_audit(ctx, reproduction.case_id, "reproduction_started", {"dry_run": reproduction.dry_run}) + allowed_hosts = set(reproduction.allowed_health_hosts or split_csv(ctx.consumer_config("HEALTHCHECK_ALLOWED_HOSTS", ""))) + safe_checks: list[HttpCheck] = [] + blocked: list[str] = [] + for check in reproduction.health_checks: + try: + validate_safe_url(check.url, allowed_hosts=allowed_hosts) + safe_checks.append(check) + except ValueError as exc: + blocked.append(f"{check.name}: {exc}") + steps = build_reproduction_steps(reproduction) + doc = reproduction_markdown(reproduction, steps, safe_checks, blocked) + digest = stable_digest({"steps": steps, "safe_checks": [c.model_dump(mode="json") for c in safe_checks], "blocked": blocked}) + artifact = await write_case_file(ctx, reproduction.case_id, "reproduction.md", doc) + audit = [audit_record(reproduction.case_id, "reproduction_completed", {"reproduction_digest": digest})] + await emit_audit(ctx, reproduction.case_id, "reproduction_completed", {"reproduction_digest": digest}) + return ReproduceIssueOutput( + status="reproduced_deterministically", + case_id=reproduction.case_id, + reproduction_digest=digest, + minimal_steps=steps, + safe_health_checks=safe_checks, + blocked_health_checks=blocked, + artifacts=[artifact], + audit=audit, + warnings=policy_warnings(reproduction.redacted_symptom, reproduction.observed_errors), + ) - @tool - def text_stats(text: str) -> str: - """Return exact word, character, and line counts for text.""" - words = [part for part in text.split() if part.strip()] - return json.dumps( - { - "characters": len(text), - "words": len(words), - "lines": len(text.splitlines()) or 1, - } - ) + @a2a.tool(description="Rank hypotheses with provenance and produce a scoped patch.diff without modifying a repository", timeout_seconds=180, idempotent=True) + async def propose_fix(self, ctx: RunContext[NoAuth], proposal: ProposeFixInput) -> ProposeFixOutput: + await emit_audit(ctx, proposal.case_id, "proposal_started", {"dry_run": proposal.dry_run}) + warnings = policy_warnings(proposal.reproduction_summary, proposal.hypotheses) + scoped_paths = [change.path for change in proposal.proposed_changes] + ranked = rank_hypotheses(proposal.hypotheses, proposal.supporting_evidence, proposal.contradicting_evidence) + patch_diff = build_patch_diff(proposal.proposed_changes) + diagnosis = { + "case_id": proposal.case_id, + "repository": proposal.repository, + "hypotheses": [item.model_dump(mode="json") for item in ranked], + "scoped_paths": scoped_paths, + "dry_run": proposal.dry_run, + "note": "No repository mutation performed. Patch is an output artifact only.", + } + patch_digest = stable_digest({"patch_diff": patch_diff, "scoped_paths": scoped_paths}) + plan_digest = stable_digest({"case_id": proposal.case_id, "repository": proposal.repository, "patch_digest": patch_digest, "paths": scoped_paths}) + artifacts = [ + await write_case_file(ctx, proposal.case_id, "diagnosis.json", json.dumps(diagnosis, indent=2, sort_keys=True)), + await write_case_file(ctx, proposal.case_id, "patch.diff", patch_diff), + ] + audit = [audit_record(proposal.case_id, "proposal_completed", {"patch_digest": patch_digest, "plan_digest": plan_digest})] + await emit_audit(ctx, proposal.case_id, "proposal_completed", {"patch_digest": patch_digest}) + return ProposeFixOutput( + status="patch_proposed_no_repository_write", + case_id=proposal.case_id, + repository=proposal.repository, + patch_digest=patch_digest, + plan_digest=plan_digest, + hypotheses=ranked, + scoped_paths=scoped_paths, + artifacts=artifacts, + audit=audit, + warnings=warnings, + ) - @wrap_model_call - async def log_model_call(request: Any, handler: Any) -> Any: - messages = request.state.get("messages", []) - print( - "[middleware] model_call " - f"model={creds.model} source={creds.source} messages={len(messages)}" - ) - return await handler(request) + @a2a.tool(description="Validate a proposed patch with deterministic static checks and test-plan selectors without executing arbitrary shell", timeout_seconds=180, idempotent=True) + async def validate_fix(self, ctx: RunContext[NoAuth], validation: ValidateFixInput) -> ValidateFixOutput: + await emit_audit(ctx, validation.case_id, "validation_started", {"dry_run": validation.dry_run}) + checks = deterministic_validation(validation) + warnings = [] if all(v == "pass" for v in checks.values()) else ["One or more deterministic checks require human review."] + payload = { + "case_id": validation.case_id, + "patch_digest": validation.patch_digest, + "checks": checks, + "test_selectors": validation.test_selectors, + "lint_selectors": validation.lint_selectors, + "security_checks": validation.security_checks, + "dry_run": validation.dry_run, + "note": "No arbitrary shell was executed; selectors are a reviewable validation plan.", + } + validation_digest = stable_digest(payload) + artifact = await write_case_file(ctx, validation.case_id, "validation.json", json.dumps({**payload, "validation_digest": validation_digest}, indent=2, sort_keys=True)) + audit = [audit_record(validation.case_id, "validation_completed", {"validation_digest": validation_digest})] + await emit_audit(ctx, validation.case_id, "validation_completed", {"validation_digest": validation_digest}) + return ValidateFixOutput( + status="validated_dry_run" if validation.dry_run else "validated_without_deploy", + case_id=validation.case_id, + validation_digest=validation_digest, + patch_digest=validation.patch_digest, + checks=checks, + artifacts=[artifact], + audit=audit, + warnings=warnings, + ) - backend = ctx.workspace_backend() - skill_sources = _seed_runtime_skills(backend, ctx) - # create_a2a_deep_agent resolves provider:model strings with - # langchain.init_chat_model from ctx.llm, preserving LiteLLM routing, - # provider-specific extra body, and runtime model overrides. - return create_a2a_deep_agent( - ctx, - creds=creds, - backend=backend, - skills=skill_sources or None, - tools=[text_stats], - middleware=[log_model_call], - system_prompt=SYSTEM_PROMPT, + @a2a.tool(description="Prepare a reviewable branch/commit/PR draft and customer-safe summary; gated actions require approval bound to the plan digest", timeout_seconds=180, idempotent=True) + async def prepare_delivery(self, ctx: RunContext[NoAuth], delivery: PrepareDeliveryInput) -> PrepareDeliveryOutput: + await emit_audit(ctx, delivery.case_id, "delivery_started", {"dry_run": delivery.dry_run}) + actions = [str(action.value if isinstance(action, DeliveryAction) else action) for action in delivery.delivery_actions] + immutable_plan = { + "case_id": delivery.case_id, + "repository": delivery.repository, + "target_branch": delivery.target_branch, + "proposed_branch": delivery.proposed_branch, + "patch_digest": delivery.patch_digest, + "validation_digest": delivery.validation_digest, + "actions": actions, + } + plan_digest = stable_digest(immutable_plan) + consequential = any(action in {DeliveryAction.draft_branch.value, DeliveryAction.draft_commit.value, DeliveryAction.draft_pr.value} for action in actions) + approved = bool( + consequential + and not delivery.dry_run + and delivery.approval_acknowledgement == APPROVAL_ACK + and delivery.approval_plan_digest == plan_digest + ) + approval_required = consequential and (delivery.dry_run or not approved) + executed_actions = actions if approved else ([DeliveryAction.customer_summary.value] if DeliveryAction.customer_summary.value in actions else []) + blocked_actions = [action for action in actions if action not in executed_actions] + doc = delivery_markdown(delivery, plan_digest, approval_required, approved, executed_actions, blocked_actions) + artifact = await write_case_file(ctx, delivery.case_id, "delivery.md", doc) + audit = [audit_record(delivery.case_id, "delivery_prepared", {"plan_digest": plan_digest, "approved": approved})] + await emit_audit(ctx, delivery.case_id, "delivery_prepared", {"plan_digest": plan_digest, "approved": approved}) + warnings = [ + "No merge or deployment was performed.", + "Repository mutations are represented as a reviewable delivery plan only unless platform approval executes them outside this agent.", + ] + return PrepareDeliveryOutput( + status="approval_required" if approval_required else "delivery_prepared", + case_id=delivery.case_id, + plan_digest=plan_digest, + approval_required=approval_required, + approved=approved, + executed_actions=executed_actions, + blocked_actions=blocked_actions, + artifacts=[artifact], + audit=audit, + warnings=warnings, ) -def _runtime_skills_root(ctx: RunContext[Any]) -> str: - workspace = getattr(ctx, "_workspace", None) - prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ()) - if not prefixes: - outputs_prefix = getattr(workspace, "outputs_prefix", None) - prefixes = (outputs_prefix or "outputs/",) - prefix = str(prefixes[0]).strip("/") - return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}" +def split_csv(value: Any) -> list[str]: + return [item.strip() for item in str(value or "").split(",") if item.strip()] -def _seed_runtime_skills(backend: Any, ctx: RunContext[Any]) -> list[str]: - """Copy packaged DeepAgents skills into the invocation workspace. - - DeepAgents loads skills from its backend, while source-controlled - ``skills/`` folders live in the image. This bridge lets generated agents - ship reusable SKILL.md bundles without giving up durable A2A workspace - files. - """ - root = Path(__file__).parent / "skills" - if not root.exists(): - return [] - runtime_skills_root = _runtime_skills_root(ctx) - uploads: list[tuple[str, bytes]] = [] - for path in root.rglob("*"): - if path.is_file(): - rel = path.relative_to(root).as_posix() - uploads.append((runtime_skills_root + rel, path.read_bytes())) - if uploads: - backend.upload_files(uploads) - return [runtime_skills_root] - return [] +def missing_setup(ctx: RunContext[NoAuth], *, include_tokens: bool) -> list[str]: + names = [] + for name in ("GITEA_BASE_URL", "KUBERNETES_API_URL", "ARGOCD_BASE_URL", "OBSERVABILITY_BASE_URL"): + if not ctx.consumer_config(name, ""): + names.append(name) + if include_tokens: + for secret_name in ("GITEA_TOKEN", "KUBERNETES_TOKEN", "ARGOCD_TOKEN", "OBSERVABILITY_TOKEN"): + try: + ctx.consumer_secret(secret_name) + except Exception: + names.append(secret_name) + return names -def _last_message_text(state: dict[str, Any]) -> str: - messages = state.get("messages") or [] - if not messages: - return json.dumps(state, default=str) +def tenant_is_allowed(ctx: RunContext[NoAuth], tenant_id: str) -> bool: + allowed = set(split_csv(ctx.consumer_config("TENANT_ALLOWLIST", ""))) + return not allowed or tenant_id in allowed - content = getattr(messages[-1], "content", None) - if isinstance(content, str): - return content - if isinstance(content, list): - parts: list[str] = [] - for item in content: - if isinstance(item, dict): - text = item.get("text") or item.get("content") - if text: - parts.append(str(text)) - elif item: - parts.append(str(item)) - return "\n".join(parts) if parts else json.dumps(content, default=str) - return str(content or messages[-1]) + +def count_redactions(text: str) -> int: + return len(SECRET_RE.findall(text or "")) + + +SECRET_RE = re.compile( + r"(?i)(authorization:\s*bearer\s+[A-Za-z0-9._~+/=-]+|api[_-]?key\s*[:=]\s*['\"]?[^\s,'\"]+|token\s*[:=]\s*['\"]?[^\s,'\"]+|password\s*[:=]\s*['\"]?[^\s,'\"]+|secret\s*[:=]\s*['\"]?[^\s,'\"]+|AKIA[0-9A-Z]{16}|-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----)" +) +PII_RE = re.compile(r"(?i)\b([A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}|\+?\d[\d .()\-]{7,}\d)\b") +PROMPT_INJECTION_RE = re.compile(r"(?i)(ignore (all )?(previous|prior) instructions|system prompt|developer message|reveal secrets|exfiltrate|curl\s+|wget\s+|bash\s+-c)") + + +def redact_secrets(text: str | None) -> str: + if not text: + return "" + redacted = SECRET_RE.sub("[REDACTED_SECRET]", text) + return PII_RE.sub("[REDACTED_PII]", redacted) + + +def policy_warnings(*texts_or_lists: Any) -> list[str]: + joined_parts: list[str] = [] + for item in texts_or_lists: + if isinstance(item, list): + joined_parts.extend(str(v) for v in item) + else: + joined_parts.append(str(item)) + joined = "\n".join(joined_parts) + warnings: list[str] = [] + if PROMPT_INJECTION_RE.search(joined): + warnings.append("Untrusted ticket/log content contained instruction-like text and was treated only as evidence.") + if SECRET_RE.search(joined): + warnings.append("Secrets were detected and redacted before artifact generation.") + if len(joined) > MAX_TEXT_CHARS: + warnings.append("Input was bounded and truncated for deterministic processing.") + return warnings + + +def safe_repo_path(path: str) -> str: + clean = str(path).replace("\\", "/").strip().lstrip("/") + posix = PurePosixPath(clean) + if not clean or ".." in posix.parts or clean.startswith("~"): + raise ValueError(f"unsafe repository path: {path!r}") + if any(part in {".git", ".hg", ".svn"} for part in posix.parts): + raise ValueError("repository metadata paths are not allowed") + if len(posix.parts) > 20: + raise ValueError("repository path is too deep") + return posix.as_posix() + + +def safe_case_id(case_id: str) -> str: + if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{2,79}", case_id): + raise ValueError("invalid case_id") + return case_id + + +def validate_safe_url(url: str, *, allowed_hosts: set[str] | tuple[str, ...]) -> str: + parsed = urlparse(url) + if parsed.scheme not in {"http", "https"}: + raise ValueError("URL scheme must be http or https") + if not parsed.hostname: + raise ValueError("URL must include a hostname") + host = parsed.hostname.lower().rstrip(".") + if allowed_hosts and host not in {h.lower().rstrip(".") for h in allowed_hosts}: + raise ValueError(f"host {host!r} is not in the allowlist") + if host in {"localhost", "metadata.google.internal"} or host.endswith(".local") or host.endswith(".internal"): + raise ValueError("internal hostnames are blocked") + try: + ip = ipaddress.ip_address(host) + except ValueError: + ip = None + if ip and (ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified): + raise ValueError("private, loopback, link-local, multicast, reserved, and unspecified IPs are blocked") + if parsed.username or parsed.password: + raise ValueError("credentials in URLs are not allowed") + return url + + +def stable_digest(payload: Any) -> str: + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def classify_severity(ticket_text: str, evidence: list[EvidenceItem]) -> Severity: + text = (ticket_text + "\n" + "\n".join(e.summary for e in evidence)).lower() + if any(term in text for term in ("outage", "data loss", "security incident", "all customers", "sev1")): + return Severity.sev1 + if any(term in text for term in ("unable", "down", "500", "payments failing", "sev2")): + return Severity.sev2 + if any(term in text for term in ("degraded", "intermittent", "slow", "error")): + return Severity.sev3 + return Severity.sev4 + + +def classify_owner(services: list[str], ticket_text: str, changes: list[str]) -> str: + haystack = " ".join([ticket_text, *services, *changes]).lower() + if "payment" in haystack or "checkout" in haystack: + return "payments" + if "auth" in haystack or "login" in haystack: + return "identity" + if "kubernetes" in haystack or "pod" in haystack or "argocd" in haystack: + return "platform" + if services: + return services[0][:80] + return "unassigned" + + +def normalized_evidence(evidence: list[EvidenceItem], ticket_text: str, release: str, changes: list[str]) -> list[EvidenceItem]: + out = list(evidence) + out.append(EvidenceItem(source_id="ticket", kind=EvidenceKind.ticket, summary=redact_secrets(ticket_text[:1200]), excerpt=redact_secrets(ticket_text[:2000]))) + out.append(EvidenceItem(source_id="release", kind=EvidenceKind.release, summary=f"Release: {redact_secrets(release)}")) + for idx, change in enumerate(changes[:20], start=1): + out.append(EvidenceItem(source_id=f"recent-change-{idx}", kind=EvidenceKind.change, summary=change[:1200])) + return out[:MAX_EVIDENCE_ITEMS] + + +def build_reproduction_steps(inp: ReproduceIssueInput) -> list[str]: + steps = [ + "Use only the redacted support case and explicitly granted repository/workspace evidence.", + f"Set environment facets: {json.dumps({k: redact_secrets(v) for k, v in sorted(inp.environment.items())}, sort_keys=True)}.", + "Trigger the smallest customer-safe path that matches the observed symptom.", + ] + if inp.observed_errors: + steps.append(f"Assert the observed error signature appears: {redact_secrets(inp.observed_errors[0])[:500]}.") + if inp.candidate_files: + steps.append("Inspect only scoped candidate files: " + ", ".join(inp.candidate_files[:10]) + ".") + if inp.health_checks: + steps.append("Run configured HTTP health checks only after SSRF allowlist validation; do not follow redirects.") + steps.append("Record expected vs actual behavior without customer secrets or raw tenant data.") + return steps + + +def reproduction_markdown(inp: ReproduceIssueInput, steps: list[str], checks: list[HttpCheck], blocked: list[str]) -> str: + lines = [ + f"# Minimal Reproduction for {inp.case_id}", + "", + "## Symptom", + redact_secrets(inp.redacted_symptom), + "", + "## Deterministic Steps", + ] + lines.extend(f"{idx}. {step}" for idx, step in enumerate(steps, start=1)) + lines.extend(["", "## Safe Health Checks"]) + if checks: + lines.extend(f"- {c.name}: {c.url} expects {c.expected_status}" for c in checks) + else: + lines.append("- None") + lines.extend(["", "## Blocked Health Checks"]) + lines.extend([f"- {item}" for item in blocked] or ["- None"]) + lines.extend(["", "## Safety", "Ticket and log instructions are untrusted evidence; no arbitrary shell or URL fetching is performed."]) + return "\n".join(lines) + "\n" + + +def rank_hypotheses(hypotheses: list[str], supporting: list[EvidenceItem], contradicting: list[EvidenceItem]) -> list[HypothesisRank]: + if not hypotheses: + hypotheses = ["Regression correlated with recent release/change and observed error signature"] + ranked: list[HypothesisRank] = [] + support_ids = [e.source_id for e in supporting] + contra_ids = [e.source_id for e in contradicting] + for idx, hypothesis in enumerate(hypotheses[:20], start=1): + confidence = max(0.1, min(0.95, 0.55 + (len(support_ids) * 0.05) - (len(contra_ids) * 0.08) - ((idx - 1) * 0.03))) + ranked.append(HypothesisRank(hypothesis=redact_secrets(hypothesis)[:1000], rank=idx, confidence=round(confidence, 2), supporting_source_ids=support_ids[:20], contradicting_source_ids=contra_ids[:20])) + return ranked + + +def build_patch_diff(changes: list[RepoFileChange]) -> str: + if not changes: + return "diff --git a/README.md b/README.md\n--- a/README.md\n+++ b/README.md\n@@ -0,0 +1,3 @@\n+# Support Case Fix Placeholder\n+No scoped code changes were supplied.\n+Attach repository evidence before implementation.\n" + parts: list[str] = [] + for change in changes: + content = change.proposed_content or f"# Proposed change\n# Reason: {change.reason}\n" + safe_content = redact_secrets(content).splitlines() + parts.append(f"diff --git a/{change.path} b/{change.path}") + parts.append(f"--- a/{change.path}") + parts.append(f"+++ b/{change.path}") + parts.append("@@ -0,0 +1,%d @@" % max(1, len(safe_content))) + parts.extend("+" + line for line in safe_content[:400]) + return "\n".join(parts) + "\n" + + +def deterministic_validation(inp: ValidateFixInput) -> dict[str, str]: + checks: dict[str, str] = {} + checks["patch_digest_format"] = "pass" if re.fullmatch(r"[a-fA-F0-9]{16,128}", inp.patch_digest) else "review" + checks["patch_has_diff_header"] = "pass" if "diff --git" in inp.patch_diff and "+++ b/" in inp.patch_diff else "review" + checks["no_secret_literals"] = "pass" if not SECRET_RE.search(inp.patch_diff) else "fail" + checks["path_containment"] = "pass" if not re.search(r"(^|\n)(---|\+\+\+) [ab]/(\.\.|/|.*\.git)", inp.patch_diff) else "fail" + checks["test_plan_present"] = "pass" if inp.test_selectors else "review" + checks["lint_plan_present"] = "pass" if inp.lint_selectors else "review" + checks["security_plan_present"] = "pass" if inp.security_checks else "review" + checks["dry_run_enforced"] = "pass" if inp.dry_run else "review" + return checks + + +def delivery_markdown(inp: PrepareDeliveryInput, plan_digest: str, approval_required: bool, approved: bool, executed: list[str], blocked: list[str]) -> str: + safe_summary = redact_secrets(inp.customer_summary_context)[:2000] + return "\n".join([ + f"# Delivery Plan for {inp.case_id}", + "", + f"Plan digest: `{plan_digest}`", + f"Repository: `{inp.repository}`", + f"Target branch: `{inp.target_branch}`", + f"Proposed branch: `{inp.proposed_branch}`", + f"Patch digest: `{inp.patch_digest}`", + f"Validation digest: `{inp.validation_digest}`", + "", + "## Approval", + f"Approval required: `{str(approval_required).lower()}`", + f"Approved: `{str(approved).lower()}`", + f"Required acknowledgement: `{APPROVAL_ACK}` bound to the plan digest above.", + "", + "## Actions", + "Executed/prepared: " + (", ".join(executed) if executed else "none"), + "Blocked: " + (", ".join(blocked) if blocked else "none"), + "", + "## Customer-safe Resolution Summary", + safe_summary or "A scoped fix has been prepared for engineering review. No customer secrets, raw logs, or deployment details are included.", + "", + "## Non-actions", + "This agent never merges, deploys, follows ticket/log instructions, executes arbitrary shell, or fetches arbitrary URLs.", + ]) + "\n" + + +def audit_record(case_id: str, event: str, details: dict[str, Any]) -> AuditRecord: + return AuditRecord(event=event, at=datetime.now(UTC).isoformat(), case_id=case_id, details=json_sanitize(details)) + + +async def emit_audit(ctx: RunContext[NoAuth], case_id: str, event: str, details: dict[str, Any]) -> None: + await ctx.emit_event(AgentEvent(kind="audit", payload=audit_record(case_id, event, details).model_dump(mode="json"))) + + +def json_sanitize(value: Any) -> Any: + raw = json.dumps(value, default=str) + return json.loads(redact_secrets(raw)) + + +async def write_case_file(ctx: RunContext[NoAuth], case_id: str, filename: str, content: str) -> ArtifactRecord: + safe_case = safe_case_id(case_id) + safe_name = safe_repo_path(filename) + if "/" in safe_name: + raise ValueError("case artifact filename must not contain subdirectories") + rel_path = f"{OUTPUT_ROOT}/{safe_case}/{safe_name}" + data = redact_secrets(content).encode("utf-8") + digest = hashlib.sha256(data).hexdigest() + artifact_uri: str | None = None + try: + view = await ctx.workspace.open_view( + purpose=f"Persist support case artifact {safe_name}", + hints=[safe_case, safe_name], + file_types=(), + max_files=1, + mode=WorkspaceMode.READ_WRITE_OVERLAY, + reason="Persist generated support-to-fix engineering artifact under outputs/support-cases/{case_id}/", + ) + await view.write(rel_path, data) + except Exception as exc: # workspace may be unavailable in local smoke invocations + await ctx.emit_event(AgentEvent(kind="workspace_write_warning", payload={"path": rel_path, "message": redact_secrets(str(exc))[:500]})) + try: + ref = await ctx.write_artifact(rel_path.replace("/", "__"), data, mime_type_for(filename)) + await ctx.emit_artifact(ref) + artifact_uri = ref.uri + except Exception as exc: + await ctx.emit_event(AgentEvent(kind="artifact_warning", payload={"path": rel_path, "message": redact_secrets(str(exc))[:500]})) + return ArtifactRecord(path=rel_path, artifact_uri=artifact_uri, sha256=digest, size_bytes=len(data)) + + +def mime_type_for(filename: str) -> str: + if filename.endswith(".json"): + return "application/json" + if filename.endswith(".md"): + return "text/markdown" + if filename.endswith(".diff") or filename.endswith(".patch"): + return "text/x-diff" + return "text/plain"