diff --git a/agent.py b/agent.py index 22d573a..e13bd2d 100644 --- a/agent.py +++ b/agent.py @@ -24,26 +24,23 @@ from a2a_pack import ( WorkspaceMode, ) 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" OUTPUT_ROOT = "outputs/support-cases" -MAX_TEXT_CHARS = 24000 -MAX_EVIDENCE_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 +MAX_TEXT = 24000 +MAX_ITEMS = 80 class StrictModel(BaseModel): 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): sev1 = "sev1" sev2 = "sev2" @@ -70,6 +67,16 @@ class DeliveryAction(str, Enum): 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): source_id: str = Field(min_length=1, max_length=160) kind: EvidenceKind @@ -80,17 +87,15 @@ class EvidenceItem(StrictModel): @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] + def clean_text(cls, value: str | None) -> str | None: + return None if value is None else redact(value)[:4000] @field_validator("url") @classmethod - def _validate_url(cls, value: str | None) -> str | None: - if value in (None, ""): + def safe_url(cls, value: str | None) -> str | None: + if not value: return None - return validate_safe_url(value, allowed_hosts=()) + return validate_safe_url(value, allowed_hosts=set()) class HttpCheck(StrictModel): @@ -106,13 +111,13 @@ class RepoFileChange(StrictModel): @field_validator("path") @classmethod - def _safe_path(cls, value: str) -> str: - return safe_repo_path(value) + def path_ok(cls, value: str) -> str: + return safe_path(value) - @field_validator("proposed_content") + @field_validator("reason", "proposed_content") @classmethod - def _redact_content(cls, value: str | None) -> str | None: - return redact_secrets(value) if value is not None else None + def redact_text(cls, value: str | None) -> str | None: + return None if value is None else redact(value) class TriageCaseInput(StrictModel): @@ -120,58 +125,58 @@ class TriageCaseInput(StrictModel): 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) + ticket_text: str = Field(min_length=1, max_length=MAX_TEXT) + evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_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") + @field_validator("ticket_text", "customer_id", "tenant_id", "release") @classmethod - def _redact_ticket(cls, value: str) -> str: - return redact_secrets(value) + def redact_fields(cls, value: str) -> str: + return redact(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()] + def clean_list(cls, values: list[str]) -> list[str]: + return [redact(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) + environment: list[KeyValue] = Field(default_factory=list, 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) + candidate_files: list[str] = Field(default_factory=list, max_length=40) 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) + def redact_symptom(cls, value: str) -> str: + 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): 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) + supporting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_ITEMS) + contradicting_evidence: list[EvidenceItem] = Field(default_factory=list, max_length=MAX_ITEMS) + proposed_changes: list[RepoFileChange] = Field(default_factory=list, max_length=40) 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) + def redact_summary(cls, value: str) -> str: + return redact(value) class ValidateFixInput(StrictModel): @@ -185,21 +190,20 @@ class ValidateFixInput(StrictModel): @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 + def selector_ok(cls, values: list[str]) -> list[str]: + out: list[str] = [] + for item in values: + cleaned = item.strip() + if cleaned: + if re.search(r"[;&|`$<>]", cleaned): + raise ValueError("selectors must not contain shell metacharacters") + out.append(cleaned[:180]) + return out @field_validator("patch_diff") @classmethod - def _patch_without_secrets(cls, value: str) -> str: - return redact_secrets(value) + def patch_clean(cls, value: str) -> str: + return redact(value) class PrepareDeliveryInput(StrictModel): @@ -217,7 +221,7 @@ class PrepareDeliveryInput(StrictModel): @field_validator("repository", "target_branch", "proposed_branch") @classmethod - def _safe_refish(cls, value: str) -> str: + def ref_ok(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 @@ -234,7 +238,20 @@ class AuditRecord(StrictModel): event: str at: 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): @@ -264,14 +281,6 @@ class ReproduceIssueOutput(StrictModel): 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 @@ -290,7 +299,7 @@ class ValidateFixOutput(StrictModel): case_id: str validation_digest: str patch_digest: str - checks: dict[str, str] + checks: list[CheckResult] artifacts: list[ArtifactRecord] audit: list[AuditRecord] warnings: list[str] @@ -311,20 +320,11 @@ class PrepareDeliveryOutput(StrictModel): class SupportToFixEngineer(A2AAgent[SupportToFixEngineerConfig, NoAuth]): name = "support-to-fix-engineer" - description = ( - "Turns authenticated support cases into evidence-backed engineering fix artifacts " - "with deterministic triage, reproduction, patch proposal, validation, and approval-gated delivery." - ) + description = "Turns support cases into evidence-backed fix artifacts with approval-gated delivery." version = "0.1.0" - config_model = SupportToFixEngineerConfig 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) workspace_access = WorkspaceAccess.dynamic( max_files=160, @@ -337,288 +337,164 @@ class SupportToFixEngineer(A2AAgent[SupportToFixEngineerConfig, NoAuth]): 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("GITEA_ALLOWED_REPOS", label="Allowed repositories", 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.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."), + ConsumerSetupField.config("TENANT_ALLOWLIST", label="Tenant allowlist", required=False), + ConsumerSetupField.config("HEALTHCHECK_ALLOWED_HOSTS", label="HTTP health check hosts", required=False), ) egress = EgressPolicy(allow_hosts=(), deny_internet_by_default=True) 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) 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) + await emit_audit(ctx, case.case_id, "triage_started", [kv("dry_run", case.dry_run)]) + warnings = warnings_for(case.ticket_text, [e.summary for e in case.evidence]) + allowed = tenant_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], - } + evidence = normalize_evidence(case) + 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]} 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" + 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}))] + audit = [audit_record(case.case_id, "triage_completed", [kv("plan_digest", digest), kv("severity", severity.value)])] + await emit_audit(ctx, case.case_id, "triage_completed", [kv("plan_digest", digest)]) + setup = missing_setup(ctx) 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, - ) + 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) @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}) + 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", ""))) - safe_checks: list[HttpCheck] = [] + safe: 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) + safe.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), - ) + steps = repro_steps(reproduction) + digest = stable_digest({"steps": steps, "safe": [c.model_dump(mode="json") for c in safe], "blocked": blocked}) + artifact = await write_case_file(ctx, reproduction.case_id, "reproduction.md", repro_markdown(reproduction, steps, safe, blocked)) + audit = [audit_record(reproduction.case_id, "reproduction_completed", [kv("reproduction_digest", digest)])] + await emit_audit(ctx, reproduction.case_id, "reproduction_completed", [kv("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)) @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] + await emit_audit(ctx, proposal.case_id, "proposal_started", [kv("dry_run", proposal.dry_run)]) 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, - ) + paths = [c.path for c in proposal.proposed_changes] + patch = patch_diff(proposal.proposed_changes) + patch_digest = stable_digest({"patch_diff": patch, "paths": paths}) + plan_digest = stable_digest({"case_id": proposal.case_id, "repository": proposal.repository, "patch_digest": patch_digest, "paths": paths}) + 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."} + artifacts = [await write_case_file(ctx, proposal.case_id, "diagnosis.json", dumps(diagnosis)), await write_case_file(ctx, proposal.case_id, "patch.diff", patch)] + audit = [audit_record(proposal.case_id, "proposal_completed", [kv("patch_digest", patch_digest), kv("plan_digest", plan_digest)])] + 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)) @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, - ) + await emit_audit(ctx, validation.case_id, "validation_started", [kv("dry_run", validation.dry_run)]) + checks = validation_checks(validation) + 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."} + digest = stable_digest(payload) + artifact = await write_case_file(ctx, validation.case_id, "validation.json", dumps({**payload, "validation_digest": digest})) + audit = [audit_record(validation.case_id, "validation_completed", [kv("validation_digest", digest)])] + await emit_audit(ctx, validation.case_id, "validation_completed", [kv("validation_digest", digest)]) + warnings = [] if all(c.status == "pass" for c in checks) else ["One or more deterministic checks require human review."] + 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) @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, - ) + await emit_audit(ctx, delivery.case_id, "delivery_started", [kv("dry_run", delivery.dry_run)]) + actions = [str(a.value if isinstance(a, DeliveryAction) else a) for a in delivery.delivery_actions] + 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} + digest = stable_digest(plan) + consequential = any(a in {"draft_branch", "draft_commit", "draft_pr"} for a in actions) + approved = consequential and not delivery.dry_run and delivery.approval_acknowledgement == APPROVAL_ACK and delivery.approval_plan_digest == digest + approval_required = consequential and not approved + executed = actions if approved else (["customer_summary"] if "customer_summary" in actions else []) + blocked = [a for a in actions if a not in executed] + 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)])] + await emit_audit(ctx, delivery.case_id, "delivery_prepared", [kv("plan_digest", digest), kv("approved", approved)]) + 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."]) + + +def kv(key: str, value: Any) -> KeyValue: + return KeyValue(key=str(key), value=str(value)) 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]: - 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 missing_setup(ctx: RunContext[NoAuth]) -> list[str]: + return [name for name in ("GITEA_BASE_URL", "KUBERNETES_API_URL", "ARGOCD_BASE_URL", "OBSERVABILITY_BASE_URL") if not ctx.consumer_config(name, "")] -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", ""))) 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: 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) +def warnings_for(*items: Any) -> list[str]: + flat: list[str] = [] + for item in items: + flat.extend(str(v) for v in item) if isinstance(item, list) else flat.append(str(item)) + text = "\n".join(flat) 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.") - if SECRET_RE.search(joined): + if SECRET_RE.search(text): 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.") return warnings -def safe_repo_path(path: str) -> str: +def safe_path(path: str) -> str: clean = str(path).replace("\\", "/").strip().lstrip("/") 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}") - 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: +def safe_case(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: +def validate_safe_url(url: str, *, allowed_hosts: set[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") + if parsed.scheme not in {"http", "https"} or not parsed.hostname: + raise ValueError("URL must be absolute http(s)") 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") @@ -629,200 +505,140 @@ def validate_safe_url(url: str, *, allowed_hosts: set[str] | tuple[str, ...]) -> 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") + raise ValueError("unsafe IP address is 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() + return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode()).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")): +def classify_severity(ticket: str, evidence: list[EvidenceItem]) -> Severity: + text = (ticket + "\n" + "\n".join(e.summary for e in evidence)).lower() + if any(x in text for x 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")): + if any(x in text for x in ("unable", "down", "500", "payments failing", "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.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: +def classify_owner(services: list[str], ticket: str, changes: list[str]) -> str: + text = " ".join([ticket, *services, *changes]).lower() + if "payment" in text or "checkout" in text: return "payments" - if "auth" in haystack or "login" in haystack: + if "auth" in text or "login" in text: 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" - if services: - return services[0][:80] - return "unassigned" + return services[0][:80] if services else "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 normalize_evidence(case: TriageCaseInput) -> list[EvidenceItem]: + out = list(case.evidence) + 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: {case.release}")) + for i, change in enumerate(case.recent_changes[:20], 1): + out.append(EvidenceItem(source_id=f"recent-change-{i}", kind=EvidenceKind.change, summary=change[:1200])) + return out[:MAX_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.", - ] +def repro_steps(inp: ReproduceIssueInput) -> list[str]: + env = {item.key: item.value for item in inp.environment} + 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."] 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: 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.") + steps.append("Run configured HTTP checks only after SSRF allowlist validation; do not follow redirects.") + steps.append("Record expected versus 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 repro_markdown(inp: ReproduceIssueInput, steps: list[str], checks: list[HttpCheck], blocked: list[str]) -> str: + body = [f"# Minimal Reproduction for {inp.case_id}", "", "## Symptom", inp.redacted_symptom, "", "## Deterministic Steps"] + 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"]) + body += ["", "## Blocked Health Checks"] + ([f"- {b}" for b in blocked] or ["- None"]) + body += ["", "## Safety", "Ticket/log instructions are untrusted evidence; no arbitrary shell or URL fetching is performed."] + return "\n".join(body) + "\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 + raw = hypotheses or ["Regression correlated with recent release/change and observed error signature"] + sids = [e.source_id for e in supporting] + cids = [e.source_id for e in contradicting] + 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)] -def build_patch_diff(changes: list[RepoFileChange]) -> str: +def 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]) + lines = redact(change.proposed_content or f"# Proposed change\n# Reason: {change.reason}\n").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 += ["+" + line for line in lines[: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 validation_checks(inp: ValidateFixInput) -> list[CheckResult]: + values = { + "patch_digest_format": "pass" if re.fullmatch(r"[a-fA-F0-9]{16,128}", inp.patch_digest) else "review", + "patch_has_diff_header": "pass" if "diff --git" in inp.patch_diff and "+++ b/" in inp.patch_diff else "review", + "no_secret_literals": "pass" if not SECRET_RE.search(inp.patch_diff) else "fail", + "path_containment": "pass" if not re.search(r"(^|\n)(---|\+\+\+) [ab]/(\.\.|/|.*\.git)", inp.patch_diff) else "fail", + "test_plan_present": "pass" if inp.test_selectors else "review", + "lint_plan_present": "pass" if inp.lint_selectors else "review", + "security_plan_present": "pass" if inp.security_checks else "review", + "dry_run_enforced": "pass" if inp.dry_run else "review", + } + 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: - 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 delivery_doc(inp: PrepareDeliveryInput, digest: str, required: bool, approved: bool, executed: list[str], blocked: list[str]) -> str: + 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([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" -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)) +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=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"))) -def json_sanitize(value: Any) -> Any: - raw = json.dumps(value, default=str) - return json.loads(redact_secrets(raw)) +def dumps(value: Any) -> str: + return redact(json.dumps(value, indent=2, sort_keys=True, default=str)) 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") + rel_path = f"{OUTPUT_ROOT}/{safe_case(case_id)}/{safe_path(filename)}" + if safe_path(filename) != filename: + raise ValueError("artifact filename must be a plain relative filename") + data = redact(content).encode() 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}/", - ) + 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}/") 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]})) + except Exception as exc: + await ctx.emit_event(AgentEvent(kind="workspace_write_warning", payload={"path": rel_path, "message": redact(str(exc))[:500]})) 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) 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]})) + 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)) -def mime_type_for(filename: str) -> str: +def mime(filename: str) -> str: if filename.endswith(".json"): return "application/json" if filename.endswith(".md"):