bobseares-2/support-to-fix-engineer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

a2a-source-edit: write agent.py

Browse Source
This commit is contained in:
a2a-cloud
2026-07-13 03:07:14 +00:00
parent 7e0dd78cd4
commit 2140614c2d
1 changed files with 236 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files

656
agent.py
View File

@@ -24,26 +24,23 @@ from a2a_pack import (
WorkspaceMode, WorkspaceMode,
) )
from a2a_pack.context import AgentEvent from a2a_pack.context import AgentEvent
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator from pydantic import BaseModel, ConfigDict, Field, field_validator
APPROVAL_ACK = "I_APPROVE_THE_LISTED_STEPS" APPROVAL_ACK = "I_APPROVE_THE_LISTED_STEPS"
OUTPUT_ROOT = "outputs/support-cases" OUTPUT_ROOT = "outputs/support-cases"
MAX_TEXT_CHARS = 24000 MAX_TEXT = 24000
MAX_EVIDENCE_ITEMS = 80 MAX_ITEMS = 80
MAX_PATHS = 40
class SupportToFixEngineerConfig(BaseModel):
model_config = ConfigDict(extra="forbid")
max_log_chars: int = Field(default=12000, ge=1000, le=50000)
default_dry_run: bool = True
class StrictModel(BaseModel): class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid", use_enum_values=True) model_config = ConfigDict(extra="forbid", use_enum_values=True)
class SupportToFixEngineerConfig(StrictModel):
max_log_chars: int = Field(default=12000, ge=1000, le=50000)
default_dry_run: bool = True
class Severity(str, Enum): class Severity(str, Enum):
sev1 = "sev1" sev1 = "sev1"
sev2 = "sev2" sev2 = "sev2"
@@ -70,6 +67,16 @@ class DeliveryAction(str, Enum):
customer_summary = "customer_summary" customer_summary = "customer_summary"
class KeyValue(StrictModel):
key: str = Field(min_length=1, max_length=120)
value: str = Field(default="", max_length=1200)
@field_validator("key", "value")
@classmethod
def clean(cls, value: str) -> str:
return redact(value.strip())[:1200]
class EvidenceItem(StrictModel): class EvidenceItem(StrictModel):
source_id: str = Field(min_length=1, max_length=160) source_id: str = Field(min_length=1, max_length=160)
kind: EvidenceKind kind: EvidenceKind
@@ -80,17 +87,15 @@ class EvidenceItem(StrictModel):
@field_validator("source_id", "summary", "excerpt") @field_validator("source_id", "summary", "excerpt")
@classmethod @classmethod
def _redact_text_fields(cls, value: str | None) -> str | None: def clean_text(cls, value: str | None) -> str | None:
if value is None: return None if value is None else redact(value)[:4000]
return None
return redact_secrets(value)[:4000]
@field_validator("url") @field_validator("url")
@classmethod @classmethod
def _validate_url(cls, value: str | None) -> str | None: def safe_url(cls, value: str | None) -> str | None:
if value in (None, ""): if not value:
return None return None
return validate_safe_url(value, allowed_hosts=()) return validate_safe_url(value, allowed_hosts=set())
class HttpCheck(StrictModel): class HttpCheck(StrictModel):
@@ -106,13 +111,13 @@ class RepoFileChange(StrictModel):
@field_validator("path") @field_validator("path")
@classmethod @classmethod
def _safe_path(cls, value: str) -> str: def path_ok(cls, value: str) -> str:
return safe_repo_path(value) return safe_path(value)
@field_validator("proposed_content") @field_validator("reason", "proposed_content")
@classmethod @classmethod
def _redact_content(cls, value: str | None) -> str | None: def redact_text(cls, value: str | None) -> str | None:
return redact_secrets(value) if value is not None else None return None if value is None else redact(value)
class TriageCaseInput(StrictModel): class TriageCaseInput(StrictModel):
@@ -120,58 +125,58 @@ class TriageCaseInput(StrictModel):
customer_id: str = Field(min_length=1, max_length=120) customer_id: str = Field(min_length=1, max_length=120)
tenant_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) release: str = Field(min_length=1, max_length=120)
ticket_text: str = Field(min_length=1, max_length=MAX_TEXT_CHARS) ticket_text: str = Field(min_length=1, max_length=MAX_TEXT)
evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_EVIDENCE_ITEMS) evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_ITEMS)
affected_services: list[str] = Field(default_factory=list, max_length=20) affected_services: list[str] = Field(default_factory=list, max_length=20)
recent_changes: list[str] = Field(default_factory=list, max_length=40) recent_changes: list[str] = Field(default_factory=list, max_length=40)
dry_run: bool = True dry_run: bool = True
@field_validator("ticket_text") @field_validator("ticket_text", "customer_id", "tenant_id", "release")
@classmethod @classmethod
def _redact_ticket(cls, value: str) -> str: def redact_fields(cls, value: str) -> str:
return redact_secrets(value) return redact(value)
@field_validator("affected_services", "recent_changes") @field_validator("affected_services", "recent_changes")
@classmethod @classmethod
def _bounded_strings(cls, values: list[str]) -> list[str]: def clean_list(cls, values: list[str]) -> list[str]:
return [redact_secrets(v.strip())[:500] for v in values if v.strip()] return [redact(v.strip())[:500] for v in values if v.strip()]
class ReproduceIssueInput(StrictModel): 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}$") 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) redacted_symptom: str = Field(min_length=1, max_length=8000)
environment: dict[str, str] = Field(default_factory=dict, max_length=30) environment: list[KeyValue] = Field(default_factory=list, max_length=30)
observed_errors: list[str] = Field(default_factory=list, max_length=40) observed_errors: list[str] = Field(default_factory=list, max_length=40)
candidate_files: list[str] = Field(default_factory=list, max_length=MAX_PATHS) candidate_files: list[str] = Field(default_factory=list, max_length=40)
health_checks: list[HttpCheck] = Field(default_factory=list, max_length=20) health_checks: list[HttpCheck] = Field(default_factory=list, max_length=20)
allowed_health_hosts: list[str] = Field(default_factory=list, max_length=30) allowed_health_hosts: list[str] = Field(default_factory=list, max_length=30)
dry_run: bool = True 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") @field_validator("redacted_symptom")
@classmethod @classmethod
def _redact_symptom(cls, value: str) -> str: def redact_symptom(cls, value: str) -> str:
return redact_secrets(value) return redact(value)
@field_validator("candidate_files")
@classmethod
def safe_paths(cls, values: list[str]) -> list[str]:
return [safe_path(v) for v in values]
class ProposeFixInput(StrictModel): 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}$") 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) reproduction_summary: str = Field(min_length=1, max_length=12000)
hypotheses: list[str] = Field(default_factory=list, max_length=20) hypotheses: list[str] = Field(default_factory=list, max_length=20)
supporting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_EVIDENCE_ITEMS) supporting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_ITEMS)
contradicting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_EVIDENCE_ITEMS) contradicting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_ITEMS)
proposed_changes: list[RepoFileChange] = Field(default_factory=list, max_length=MAX_PATHS) proposed_changes: list[RepoFileChange] = Field(default_factory=list, max_length=40)
repository: str | None = Field(default=None, max_length=200) repository: str | None = Field(default=None, max_length=200)
dry_run: bool = True dry_run: bool = True
@field_validator("reproduction_summary") @field_validator("reproduction_summary")
@classmethod @classmethod
def _redact_repro(cls, value: str) -> str: def redact_summary(cls, value: str) -> str:
return redact_secrets(value) return redact(value)
class ValidateFixInput(StrictModel): class ValidateFixInput(StrictModel):
@@ -185,21 +190,20 @@ class ValidateFixInput(StrictModel):
@field_validator("test_selectors", "lint_selectors", "security_checks") @field_validator("test_selectors", "lint_selectors", "security_checks")
@classmethod @classmethod
def _selectors(cls, values: list[str]) -> list[str]: def selector_ok(cls, values: list[str]) -> list[str]:
cleaned: list[str] = [] out: list[str] = []
for value in values: for item in values:
item = value.strip() cleaned = item.strip()
if not item: if cleaned:
continue if re.search(r"[;&|`$<>]", cleaned):
if re.search(r"[;&|`$<>]", item): raise ValueError("selectors must not contain shell metacharacters")
raise ValueError("selectors must not contain shell metacharacters") out.append(cleaned[:180])
cleaned.append(item[:180]) return out
return cleaned
@field_validator("patch_diff") @field_validator("patch_diff")
@classmethod @classmethod
def _patch_without_secrets(cls, value: str) -> str: def patch_clean(cls, value: str) -> str:
return redact_secrets(value) return redact(value)
class PrepareDeliveryInput(StrictModel): class PrepareDeliveryInput(StrictModel):
@@ -217,7 +221,7 @@ class PrepareDeliveryInput(StrictModel):
@field_validator("repository", "target_branch", "proposed_branch") @field_validator("repository", "target_branch", "proposed_branch")
@classmethod @classmethod
def _safe_refish(cls, value: str) -> str: def ref_ok(cls, value: str) -> str:
if re.search(r"[\s;&|`$<>]", value): if re.search(r"[\s;&|`$<>]", value):
raise ValueError("repository and branch values must not contain whitespace or shell metacharacters") raise ValueError("repository and branch values must not contain whitespace or shell metacharacters")
return value return value
@@ -234,7 +238,20 @@ class AuditRecord(StrictModel):
event: str event: str
at: str at: str
case_id: str case_id: str
details: dict[str, Any] = Field(default_factory=dict) details: list[KeyValue] = Field(default_factory=list, max_length=20)
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 CheckResult(StrictModel):
name: str
status: str
class TriageCaseOutput(StrictModel): class TriageCaseOutput(StrictModel):
@@ -264,14 +281,6 @@ class ReproduceIssueOutput(StrictModel):
warnings: list[str] 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): class ProposeFixOutput(StrictModel):
status: str status: str
case_id: str case_id: str
@@ -290,7 +299,7 @@ class ValidateFixOutput(StrictModel):
case_id: str case_id: str
validation_digest: str validation_digest: str
patch_digest: str patch_digest: str
checks: dict[str, str] checks: list[CheckResult]
artifacts: list[ArtifactRecord] artifacts: list[ArtifactRecord]
audit: list[AuditRecord] audit: list[AuditRecord]
warnings: list[str] warnings: list[str]
@@ -311,20 +320,11 @@ class PrepareDeliveryOutput(StrictModel):
class SupportToFixEngineer(A2AAgent[SupportToFixEngineerConfig, NoAuth]): class SupportToFixEngineer(A2AAgent[SupportToFixEngineerConfig, NoAuth]):
name = "support-to-fix-engineer" name = "support-to-fix-engineer"
description = ( description = "Turns support cases into evidence-backed fix artifacts with approval-gated delivery."
"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" version = "0.1.0"
config_model = SupportToFixEngineerConfig config_model = SupportToFixEngineerConfig
auth_model = NoAuth auth_model = NoAuth
pricing = Pricing(price_per_call_usd=0.0, caller_pays_llm=False, notes="Deterministic policy and artifact generation; no LLM credential required.")
pricing = Pricing(
price_per_call_usd=0.0,
caller_pays_llm=False,
notes="Deterministic policy and artifact generation; no LLM credential required.",
)
resources = Resources(cpu="1", memory="512Mi", max_runtime_seconds=600) resources = Resources(cpu="1", memory="512Mi", max_runtime_seconds=600)
workspace_access = WorkspaceAccess.dynamic( workspace_access = WorkspaceAccess.dynamic(
max_files=160, max_files=160,
@@ -337,288 +337,164 @@ class SupportToFixEngineer(A2AAgent[SupportToFixEngineerConfig, NoAuth]):
consumer_setup = ConsumerSetup.from_fields( consumer_setup = ConsumerSetup.from_fields(
ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", input_type="url", required=False), ConsumerSetupField.config("GITEA_BASE_URL", label="Gitea base URL", input_type="url", required=False),
ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea token", 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("GITEA_ALLOWED_REPOS", label="Allowed repositories", required=False),
ConsumerSetupField.config("KUBERNETES_API_URL", label="Kubernetes API URL", input_type="url", required=False), ConsumerSetupField.config("KUBERNETES_API_URL", label="Kubernetes API URL", input_type="url", required=False),
ConsumerSetupField.secret("KUBERNETES_TOKEN", label="Kubernetes token", 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.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.secret("ARGOCD_TOKEN", label="Argo CD token", required=False),
ConsumerSetupField.config("OBSERVABILITY_BASE_URL", label="Observability base URL", input_type="url", 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.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("TENANT_ALLOWLIST", label="Tenant allowlist", required=False),
ConsumerSetupField.config("HEALTHCHECK_ALLOWED_HOSTS", label="HTTP health check hosts", required=False, description="Comma-separated exact host allowlist."), ConsumerSetupField.config("HEALTHCHECK_ALLOWED_HOSTS", label="HTTP health check hosts", required=False),
) )
egress = EgressPolicy(allow_hosts=(), deny_internet_by_default=True) egress = EgressPolicy(allow_hosts=(), deny_internet_by_default=True)
tools_used = ("pydantic", "workspace") tools_used = ("pydantic", "workspace")
@a2a.tool(description="Intake, redact, correlate evidence, classify severity and ownership for a bounded support case", timeout_seconds=120, idempotent=True) @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: 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}) await emit_audit(ctx, case.case_id, "triage_started", [kv("dry_run", case.dry_run)])
warnings = policy_warnings(case.ticket_text, [e.summary for e in case.evidence]) warnings = warnings_for(case.ticket_text, [e.summary for e in case.evidence])
setup = missing_setup(ctx, include_tokens=False) allowed = tenant_allowed(ctx, case.tenant_id)
tenant_allowed = tenant_is_allowed(ctx, case.tenant_id)
severity = classify_severity(case.ticket_text, case.evidence) severity = classify_severity(case.ticket_text, case.evidence)
owner = classify_owner(case.affected_services, case.ticket_text, case.recent_changes) 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) evidence = normalize_evidence(case)
payload = { payload = {"case_id": case.case_id, "customer_id": case.customer_id, "tenant_id": case.tenant_id, "release": case.release, "severity": severity.value, "owner": owner, "allowed_tenant": allowed, "evidence": [e.model_dump(mode="json") for e in evidence]}
"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) digest = stable_digest(payload)
artifacts = [ artifacts = [await write_case_file(ctx, case.case_id, "redacted_case.json", dumps(payload)), await write_case_file(ctx, case.case_id, "triage.json", dumps({**payload, "plan_digest": digest}))]
await write_case_file(ctx, case.case_id, "redacted_case.json", json.dumps(payload, indent=2, sort_keys=True)), audit = [audit_record(case.case_id, "triage_completed", [kv("plan_digest", digest), kv("severity", severity.value)])]
await write_case_file(ctx, case.case_id, "triage.json", json.dumps({**payload, "plan_digest": digest}, indent=2, sort_keys=True)), await emit_audit(ctx, case.case_id, "triage_completed", [kv("plan_digest", digest)])
] setup = missing_setup(ctx)
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: if setup:
warnings.append("Some integrations are not configured; analysis used caller-supplied evidence only.") warnings.append("Some integrations are not configured; analysis used caller-supplied evidence only.")
return TriageCaseOutput( return TriageCaseOutput(status="triaged" if allowed else "blocked_ungranted_tenant", case_id=case.case_id, severity=severity, owner=owner, redaction_count=count_redactions(case.ticket_text), missing_setup=setup, allowed_tenant=allowed, plan_digest=digest, evidence_index=evidence, artifacts=artifacts, audit=audit, warnings=warnings)
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,
)
@a2a.tool(description="Create a deterministic minimal reproduction and validate HTTP health-check targets against SSRF protections", timeout_seconds=120, idempotent=True) @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: 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}) await emit_audit(ctx, reproduction.case_id, "reproduction_started", [kv("dry_run", reproduction.dry_run)])
allowed_hosts = set(reproduction.allowed_health_hosts or split_csv(ctx.consumer_config("HEALTHCHECK_ALLOWED_HOSTS", ""))) allowed_hosts = set(reproduction.allowed_health_hosts or split_csv(ctx.consumer_config("HEALTHCHECK_ALLOWED_HOSTS", "")))
safe_checks: list[HttpCheck] = [] safe: list[HttpCheck] = []
blocked: list[str] = [] blocked: list[str] = []
for check in reproduction.health_checks: for check in reproduction.health_checks:
try: try:
validate_safe_url(check.url, allowed_hosts=allowed_hosts) validate_safe_url(check.url, allowed_hosts=allowed_hosts)
safe_checks.append(check) safe.append(check)
except ValueError as exc: except ValueError as exc:
blocked.append(f"{check.name}: {exc}") blocked.append(f"{check.name}: {exc}")
steps = build_reproduction_steps(reproduction) steps = repro_steps(reproduction)
doc = reproduction_markdown(reproduction, steps, safe_checks, blocked) digest = stable_digest({"steps": steps, "safe": [c.model_dump(mode="json") for c in safe], "blocked": 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", repro_markdown(reproduction, steps, safe, blocked))
artifact = await write_case_file(ctx, reproduction.case_id, "reproduction.md", doc) audit = [audit_record(reproduction.case_id, "reproduction_completed", [kv("reproduction_digest", digest)])]
audit = [audit_record(reproduction.case_id, "reproduction_completed", {"reproduction_digest": digest})] await emit_audit(ctx, reproduction.case_id, "reproduction_completed", [kv("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, blocked_health_checks=blocked, artifacts=[artifact], audit=audit, warnings=warnings_for(reproduction.redacted_symptom, reproduction.observed_errors))
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),
)
@a2a.tool(description="Rank hypotheses with provenance and produce a scoped patch.diff without modifying a repository", timeout_seconds=180, idempotent=True) @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: 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}) await emit_audit(ctx, proposal.case_id, "proposal_started", [kv("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) ranked = rank_hypotheses(proposal.hypotheses, proposal.supporting_evidence, proposal.contradicting_evidence)
patch_diff = build_patch_diff(proposal.proposed_changes) paths = [c.path for c in proposal.proposed_changes]
diagnosis = { patch = patch_diff(proposal.proposed_changes)
"case_id": proposal.case_id, patch_digest = stable_digest({"patch_diff": patch, "paths": paths})
"repository": proposal.repository, plan_digest = stable_digest({"case_id": proposal.case_id, "repository": proposal.repository, "patch_digest": patch_digest, "paths": paths})
"hypotheses": [item.model_dump(mode="json") for item in ranked], diagnosis = {"case_id": proposal.case_id, "repository": proposal.repository, "hypotheses": [h.model_dump(mode="json") for h in ranked], "scoped_paths": paths, "dry_run": proposal.dry_run, "note": "No repository mutation performed. Patch is an output artifact only."}
"scoped_paths": scoped_paths, artifacts = [await write_case_file(ctx, proposal.case_id, "diagnosis.json", dumps(diagnosis)), await write_case_file(ctx, proposal.case_id, "patch.diff", patch)]
"dry_run": proposal.dry_run, audit = [audit_record(proposal.case_id, "proposal_completed", [kv("patch_digest", patch_digest), kv("plan_digest", plan_digest)])]
"note": "No repository mutation performed. Patch is an output artifact only.", await emit_audit(ctx, proposal.case_id, "proposal_completed", [kv("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=paths, artifacts=artifacts, audit=audit, warnings=warnings_for(proposal.reproduction_summary, proposal.hypotheses))
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,
)
@a2a.tool(description="Validate a proposed patch with deterministic static checks and test-plan selectors without executing arbitrary shell", timeout_seconds=180, idempotent=True) @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: 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}) await emit_audit(ctx, validation.case_id, "validation_started", [kv("dry_run", validation.dry_run)])
checks = deterministic_validation(validation) checks = validation_checks(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": [c.model_dump(mode="json") for c in 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."}
payload = { digest = stable_digest(payload)
"case_id": validation.case_id, artifact = await write_case_file(ctx, validation.case_id, "validation.json", dumps({**payload, "validation_digest": digest}))
"patch_digest": validation.patch_digest, audit = [audit_record(validation.case_id, "validation_completed", [kv("validation_digest", digest)])]
"checks": checks, await emit_audit(ctx, validation.case_id, "validation_completed", [kv("validation_digest", digest)])
"test_selectors": validation.test_selectors, warnings = [] if all(c.status == "pass" for c in checks) else ["One or more deterministic checks require human review."]
"lint_selectors": validation.lint_selectors, return ValidateFixOutput(status="validated_dry_run" if validation.dry_run else "validated_without_deploy", case_id=validation.case_id, validation_digest=digest, patch_digest=validation.patch_digest, checks=checks, artifacts=[artifact], audit=audit, warnings=warnings)
"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,
)
@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) @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: 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}) await emit_audit(ctx, delivery.case_id, "delivery_started", [kv("dry_run", delivery.dry_run)])
actions = [str(action.value if isinstance(action, DeliveryAction) else action) for action in delivery.delivery_actions] actions = [str(a.value if isinstance(a, DeliveryAction) else a) for a in delivery.delivery_actions]
immutable_plan = { 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}
"case_id": delivery.case_id, digest = stable_digest(plan)
"repository": delivery.repository, consequential = any(a in {"draft_branch", "draft_commit", "draft_pr"} for a in actions)
"target_branch": delivery.target_branch, approved = consequential and not delivery.dry_run and delivery.approval_acknowledgement == APPROVAL_ACK and delivery.approval_plan_digest == digest
"proposed_branch": delivery.proposed_branch, approval_required = consequential and not approved
"patch_digest": delivery.patch_digest, executed = actions if approved else (["customer_summary"] if "customer_summary" in actions else [])
"validation_digest": delivery.validation_digest, blocked = [a for a in actions if a not in executed]
"actions": actions, artifact = await write_case_file(ctx, delivery.case_id, "delivery.md", delivery_doc(delivery, digest, approval_required, approved, executed, blocked))
} audit = [audit_record(delivery.case_id, "delivery_prepared", [kv("plan_digest", digest), kv("approved", approved)])]
plan_digest = stable_digest(immutable_plan) await emit_audit(ctx, delivery.case_id, "delivery_prepared", [kv("plan_digest", digest), kv("approved", approved)])
consequential = any(action in {DeliveryAction.draft_branch.value, DeliveryAction.draft_commit.value, DeliveryAction.draft_pr.value} for action in actions) return PrepareDeliveryOutput(status="approval_required" if approval_required else "delivery_prepared", case_id=delivery.case_id, plan_digest=digest, approval_required=approval_required, approved=bool(approved), executed_actions=executed, blocked_actions=blocked, artifacts=[artifact], audit=audit, 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."])
approved = bool(
consequential
and not delivery.dry_run def kv(key: str, value: Any) -> KeyValue:
and delivery.approval_acknowledgement == APPROVAL_ACK return KeyValue(key=str(key), value=str(value))
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 split_csv(value: Any) -> list[str]: def split_csv(value: Any) -> list[str]:
return [item.strip() for item in str(value or "").split(",") if item.strip()] return [x.strip() for x in str(value or "").split(",") if x.strip()]
def missing_setup(ctx: RunContext[NoAuth], *, include_tokens: bool) -> list[str]: def missing_setup(ctx: RunContext[NoAuth]) -> list[str]:
names = [] return [name for name in ("GITEA_BASE_URL", "KUBERNETES_API_URL", "ARGOCD_BASE_URL", "OBSERVABILITY_BASE_URL") if not ctx.consumer_config(name, "")]
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 tenant_is_allowed(ctx: RunContext[NoAuth], tenant_id: str) -> bool: def tenant_allowed(ctx: RunContext[NoAuth], tenant_id: str) -> bool:
allowed = set(split_csv(ctx.consumer_config("TENANT_ALLOWLIST", ""))) allowed = set(split_csv(ctx.consumer_config("TENANT_ALLOWLIST", "")))
return not allowed or tenant_id in allowed return not allowed or tenant_id in allowed
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")
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(text: str | None) -> str:
if not text:
return ""
return PII_RE.sub("[REDACTED_PII]", SECRET_RE.sub("[REDACTED_SECRET]", str(text)))
def count_redactions(text: str) -> int: def count_redactions(text: str) -> int:
return len(SECRET_RE.findall(text or "")) return len(SECRET_RE.findall(text or ""))
SECRET_RE = re.compile( def warnings_for(*items: Any) -> list[str]:
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-----)" flat: list[str] = []
) for item in items:
PII_RE = re.compile(r"(?i)\b([A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}|\+?\d[\d .()\-]{7,}\d)\b") flat.extend(str(v) for v in item) if isinstance(item, list) else flat.append(str(item))
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)") text = "\n".join(flat)
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] = [] warnings: list[str] = []
if PROMPT_INJECTION_RE.search(joined): if INJECTION_RE.search(text):
warnings.append("Untrusted ticket/log content contained instruction-like text and was treated only as evidence.") warnings.append("Untrusted ticket/log content contained instruction-like text and was treated only as evidence.")
if SECRET_RE.search(joined): if SECRET_RE.search(text):
warnings.append("Secrets were detected and redacted before artifact generation.") warnings.append("Secrets were detected and redacted before artifact generation.")
if len(joined) > MAX_TEXT_CHARS: if len(text) > MAX_TEXT:
warnings.append("Input was bounded and truncated for deterministic processing.") warnings.append("Input was bounded and truncated for deterministic processing.")
return warnings return warnings
def safe_repo_path(path: str) -> str: def safe_path(path: str) -> str:
clean = str(path).replace("\\", "/").strip().lstrip("/") clean = str(path).replace("\\", "/").strip().lstrip("/")
posix = PurePosixPath(clean) posix = PurePosixPath(clean)
if not clean or ".." in posix.parts or clean.startswith("~"): if not clean or ".." in posix.parts or clean.startswith("~") or any(p in {".git", ".hg", ".svn"} for p in posix.parts):
raise ValueError(f"unsafe repository path: {path!r}") 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() return posix.as_posix()
def safe_case_id(case_id: str) -> str: def safe_case(case_id: str) -> str:
if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{2,79}", case_id): if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._-]{2,79}", case_id):
raise ValueError("invalid case_id") raise ValueError("invalid case_id")
return case_id return case_id
def validate_safe_url(url: str, *, allowed_hosts: set[str] | tuple[str, ...]) -> str: def validate_safe_url(url: str, *, allowed_hosts: set[str]) -> str:
parsed = urlparse(url) parsed = urlparse(url)
if parsed.scheme not in {"http", "https"}: if parsed.scheme not in {"http", "https"} or not parsed.hostname:
raise ValueError("URL scheme must be http or https") raise ValueError("URL must be absolute http(s)")
if not parsed.hostname:
raise ValueError("URL must include a hostname")
host = parsed.hostname.lower().rstrip(".") host = parsed.hostname.lower().rstrip(".")
if allowed_hosts and host not in {h.lower().rstrip(".") for h in allowed_hosts}: 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") raise ValueError(f"host {host!r} is not in the allowlist")
@@ -629,200 +505,140 @@ def validate_safe_url(url: str, *, allowed_hosts: set[str] | tuple[str, ...]) ->
except ValueError: except ValueError:
ip = None 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): 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") raise ValueError("unsafe IP address is blocked")
if parsed.username or parsed.password: if parsed.username or parsed.password:
raise ValueError("credentials in URLs are not allowed") raise ValueError("credentials in URLs are not allowed")
return url return url
def stable_digest(payload: Any) -> str: def stable_digest(payload: Any) -> str:
encoded = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode()).hexdigest()
return hashlib.sha256(encoded).hexdigest()
def classify_severity(ticket_text: str, evidence: list[EvidenceItem]) -> Severity: def classify_severity(ticket: str, evidence: list[EvidenceItem]) -> Severity:
text = (ticket_text + "\n" + "\n".join(e.summary for e in evidence)).lower() text = (ticket + "\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")): if any(x in text for x in ("outage", "data loss", "security incident", "all customers", "sev1")):
return Severity.sev1 return Severity.sev1
if any(term in text for term in ("unable", "down", "500", "payments failing", "sev2")): if any(x in text for x in ("unable", "down", "500", "payments failing", "sev2")):
return Severity.sev2 return Severity.sev2
if any(term in text for term in ("degraded", "intermittent", "slow", "error")): if any(x in text for x in ("degraded", "intermittent", "slow", "error")):
return Severity.sev3 return Severity.sev3
return Severity.sev4 return Severity.sev4
def classify_owner(services: list[str], ticket_text: str, changes: list[str]) -> str: def classify_owner(services: list[str], ticket: str, changes: list[str]) -> str:
haystack = " ".join([ticket_text, *services, *changes]).lower() text = " ".join([ticket, *services, *changes]).lower()
if "payment" in haystack or "checkout" in haystack: if "payment" in text or "checkout" in text:
return "payments" return "payments"
if "auth" in haystack or "login" in haystack: if "auth" in text or "login" in text:
return "identity" return "identity"
if "kubernetes" in haystack or "pod" in haystack or "argocd" in haystack: if "kubernetes" in text or "pod" in text or "argocd" in text:
return "platform" return "platform"
if services: return services[0][:80] if services else "unassigned"
return services[0][:80]
return "unassigned"
def normalized_evidence(evidence: list[EvidenceItem], ticket_text: str, release: str, changes: list[str]) -> list[EvidenceItem]: def normalize_evidence(case: TriageCaseInput) -> list[EvidenceItem]:
out = list(evidence) out = list(case.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="ticket", kind=EvidenceKind.ticket, summary=case.ticket_text[:1200], excerpt=case.ticket_text[:2000]))
out.append(EvidenceItem(source_id="release", kind=EvidenceKind.release, summary=f"Release: {redact_secrets(release)}")) out.append(EvidenceItem(source_id="release", kind=EvidenceKind.release, summary=f"Release: {case.release}"))
for idx, change in enumerate(changes[:20], start=1): for i, change in enumerate(case.recent_changes[:20], 1):
out.append(EvidenceItem(source_id=f"recent-change-{idx}", kind=EvidenceKind.change, summary=change[:1200])) out.append(EvidenceItem(source_id=f"recent-change-{i}", kind=EvidenceKind.change, summary=change[:1200]))
return out[:MAX_EVIDENCE_ITEMS] return out[:MAX_ITEMS]
def build_reproduction_steps(inp: ReproduceIssueInput) -> list[str]: def repro_steps(inp: ReproduceIssueInput) -> list[str]:
steps = [ env = {item.key: item.value for item in inp.environment}
"Use only the redacted support case and explicitly granted repository/workspace evidence.", steps = ["Use only redacted case data and explicitly granted workspace evidence.", f"Set environment facets: {json.dumps(env, sort_keys=True)}.", "Trigger the smallest customer-safe path matching the symptom."]
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: if inp.observed_errors:
steps.append(f"Assert the observed error signature appears: {redact_secrets(inp.observed_errors[0])[:500]}.") steps.append(f"Assert observed signature: {redact(inp.observed_errors[0])[:500]}.")
if inp.candidate_files: if inp.candidate_files:
steps.append("Inspect only scoped candidate files: " + ", ".join(inp.candidate_files[:10]) + ".") steps.append("Inspect only scoped candidate files: " + ", ".join(inp.candidate_files[:10]) + ".")
if inp.health_checks: if inp.health_checks:
steps.append("Run configured HTTP health checks only after SSRF allowlist validation; do not follow redirects.") steps.append("Run configured HTTP checks only after SSRF allowlist validation; do not follow redirects.")
steps.append("Record expected vs actual behavior without customer secrets or raw tenant data.") steps.append("Record expected versus actual behavior without customer secrets or raw tenant data.")
return steps return steps
def reproduction_markdown(inp: ReproduceIssueInput, steps: list[str], checks: list[HttpCheck], blocked: list[str]) -> str: def repro_markdown(inp: ReproduceIssueInput, steps: list[str], checks: list[HttpCheck], blocked: list[str]) -> str:
lines = [ body = [f"# Minimal Reproduction for {inp.case_id}", "", "## Symptom", inp.redacted_symptom, "", "## Deterministic Steps"]
f"# Minimal Reproduction for {inp.case_id}", body += [f"{i}. {s}" for i, s in enumerate(steps, 1)]
"", body += ["", "## Safe Health Checks"] + ([f"- {c.name}: {c.url} expects {c.expected_status}" for c in checks] or ["- None"])
"## Symptom", body += ["", "## Blocked Health Checks"] + ([f"- {b}" for b in blocked] or ["- None"])
redact_secrets(inp.redacted_symptom), body += ["", "## Safety", "Ticket/log instructions are untrusted evidence; no arbitrary shell or URL fetching is performed."]
"", return "\n".join(body) + "\n"
"## 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]: def rank_hypotheses(hypotheses: list[str], supporting: list[EvidenceItem], contradicting: list[EvidenceItem]) -> list[HypothesisRank]:
if not hypotheses: raw = hypotheses or ["Regression correlated with recent release/change and observed error signature"]
hypotheses = ["Regression correlated with recent release/change and observed error signature"] sids = [e.source_id for e in supporting]
ranked: list[HypothesisRank] = [] cids = [e.source_id for e in contradicting]
support_ids = [e.source_id for e in supporting] return [HypothesisRank(hypothesis=redact(h)[:1000], rank=i, confidence=round(max(0.1, min(0.95, 0.55 + len(sids) * 0.05 - len(cids) * 0.08 - (i - 1) * 0.03)), 2), supporting_source_ids=sids[:20], contradicting_source_ids=cids[:20]) for i, h in enumerate(raw[:20], 1)]
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: def patch_diff(changes: list[RepoFileChange]) -> str:
if not changes: 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" 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] = [] parts: list[str] = []
for change in changes: for change in changes:
content = change.proposed_content or f"# Proposed change\n# Reason: {change.reason}\n" lines = redact(change.proposed_content or f"# Proposed change\n# Reason: {change.reason}\n").splitlines()
safe_content = redact_secrets(content).splitlines() parts += [f"diff --git a/{change.path} b/{change.path}", f"--- a/{change.path}", f"+++ b/{change.path}", f"@@ -0,0 +1,{max(1, len(lines))} @@"]
parts.append(f"diff --git a/{change.path} b/{change.path}") parts += ["+" + line for line in lines[:400]]
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" return "\n".join(parts) + "\n"
def deterministic_validation(inp: ValidateFixInput) -> dict[str, str]: def validation_checks(inp: ValidateFixInput) -> list[CheckResult]:
checks: dict[str, str] = {} values = {
checks["patch_digest_format"] = "pass" if re.fullmatch(r"[a-fA-F0-9]{16,128}", inp.patch_digest) else "review" "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" "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" "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" "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" "test_plan_present": "pass" if inp.test_selectors else "review",
checks["lint_plan_present"] = "pass" if inp.lint_selectors else "review" "lint_plan_present": "pass" if inp.lint_selectors else "review",
checks["security_plan_present"] = "pass" if inp.security_checks else "review" "security_plan_present": "pass" if inp.security_checks else "review",
checks["dry_run_enforced"] = "pass" if inp.dry_run else "review" "dry_run_enforced": "pass" if inp.dry_run else "review",
return checks }
return [CheckResult(name=k, status=v) for k, v in values.items()]
def delivery_markdown(inp: PrepareDeliveryInput, plan_digest: str, approval_required: bool, approved: bool, executed: list[str], blocked: list[str]) -> str: def delivery_doc(inp: PrepareDeliveryInput, digest: str, required: bool, approved: bool, executed: list[str], blocked: list[str]) -> str:
safe_summary = redact_secrets(inp.customer_summary_context)[:2000] summary = redact(inp.customer_summary_context)[:2000] or "A scoped fix has been prepared for engineering review. No customer secrets, raw logs, or deployment details are included."
return "\n".join([ return "\n".join([f"# Delivery Plan for {inp.case_id}", "", f"Plan digest: `{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(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", summary, "", "## Non-actions", "This agent never merges, deploys, follows ticket/log instructions, executes arbitrary shell, or fetches arbitrary URLs."]) + "\n"
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: def audit_record(case_id: str, event: str, details: list[KeyValue]) -> AuditRecord:
return AuditRecord(event=event, at=datetime.now(UTC).isoformat(), case_id=case_id, details=json_sanitize(details)) return AuditRecord(event=event, at=datetime.now(UTC).isoformat(), case_id=case_id, details=details)
async def emit_audit(ctx: RunContext[NoAuth], case_id: str, event: str, details: dict[str, Any]) -> None: async def emit_audit(ctx: RunContext[NoAuth], case_id: str, event: str, details: list[KeyValue]) -> None:
await ctx.emit_event(AgentEvent(kind="audit", payload=audit_record(case_id, event, details).model_dump(mode="json"))) await ctx.emit_event(AgentEvent(kind="audit", payload=audit_record(case_id, event, details).model_dump(mode="json")))
def json_sanitize(value: Any) -> Any: def dumps(value: Any) -> str:
raw = json.dumps(value, default=str) return redact(json.dumps(value, indent=2, sort_keys=True, default=str))
return json.loads(redact_secrets(raw))
async def write_case_file(ctx: RunContext[NoAuth], case_id: str, filename: str, content: str) -> ArtifactRecord: async def write_case_file(ctx: RunContext[NoAuth], case_id: str, filename: str, content: str) -> ArtifactRecord:
safe_case = safe_case_id(case_id) rel_path = f"{OUTPUT_ROOT}/{safe_case(case_id)}/{safe_path(filename)}"
safe_name = safe_repo_path(filename) if safe_path(filename) != filename:
if "/" in safe_name: raise ValueError("artifact filename must be a plain relative filename")
raise ValueError("case artifact filename must not contain subdirectories") data = redact(content).encode()
rel_path = f"{OUTPUT_ROOT}/{safe_case}/{safe_name}"
data = redact_secrets(content).encode("utf-8")
digest = hashlib.sha256(data).hexdigest() digest = hashlib.sha256(data).hexdigest()
artifact_uri: str | None = None artifact_uri: str | None = None
try: try:
view = await ctx.workspace.open_view( view = await ctx.workspace.open_view(purpose=f"Persist support case artifact {filename}", hints=[case_id, filename], file_types=(), max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="Persist generated support-to-fix engineering artifact under outputs/support-cases/{case_id}/")
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) await view.write(rel_path, data)
except Exception as exc: # workspace may be unavailable in local smoke invocations except Exception as exc:
await ctx.emit_event(AgentEvent(kind="workspace_write_warning", payload={"path": rel_path, "message": redact_secrets(str(exc))[:500]})) await ctx.emit_event(AgentEvent(kind="workspace_write_warning", payload={"path": rel_path, "message": redact(str(exc))[:500]}))
try: try:
ref = await ctx.write_artifact(rel_path.replace("/", "__"), data, mime_type_for(filename)) ref = await ctx.write_artifact(rel_path.replace("/", "__"), data, mime(filename))
await ctx.emit_artifact(ref) await ctx.emit_artifact(ref)
artifact_uri = ref.uri artifact_uri = ref.uri
except Exception as exc: except Exception as exc:
await ctx.emit_event(AgentEvent(kind="artifact_warning", payload={"path": rel_path, "message": redact_secrets(str(exc))[:500]})) await ctx.emit_event(AgentEvent(kind="artifact_warning", payload={"path": rel_path, "message": redact(str(exc))[:500]}))
return ArtifactRecord(path=rel_path, artifact_uri=artifact_uri, sha256=digest, size_bytes=len(data)) return ArtifactRecord(path=rel_path, artifact_uri=artifact_uri, sha256=digest, size_bytes=len(data))
def mime_type_for(filename: str) -> str: def mime(filename: str) -> str:
if filename.endswith(".json"): if filename.endswith(".json"):
return "application/json" return "application/json"
if filename.endswith(".md"): if filename.endswith(".md"):