Add InfraGuard review pipeline

src/iac_code/a2a/artifacts.py

@@ -23,6 +23,11 @@
 from iac_code.utils.public_errors import sanitize_public_text
 from iac_code.utils.public_paths import relativize_public_file_uri
 from iac_code.utils.state_io import atomic_write_bytes
+from iac_code.utils.tool_result_redaction import (
+    REDACTED_TOOL_RESULT_VALUE,
+    is_file_content_key,
+    redact_file_content_from_json_string,
+)
 
 PUBLIC_ARTIFACT_URI_PREFIX = "iac-code-artifact://"
 _PUBLIC_ARTIFACT_URI_SCHEME = "iac-code-artifact"
@@ -137,6 +142,9 @@ def sanitize_public_artifact_data(
         if not key_is_public_path and _is_sensitive_output_key(key_name):
             sanitized[output_key] = "[REDACTED]"
             continue
+        if not key_is_public_path and is_file_content_key(key_name):
+            sanitized[output_key] = REDACTED_TOOL_RESULT_VALUE
+            continue
         if _is_artifact_uri_key(key_name):
             if isinstance(raw_value, str):
                 sanitized_value = _sanitize_artifact_string(key_name, raw_value, public_path_roots=public_path_roots)
@@ -159,6 +167,7 @@ def sanitize_public_tool_output_data(
     public_path_roots: Iterable[Mapping[str, str]] | None = None,
 ) -> Any:
     if isinstance(value, str):
+        value = redact_file_content_from_json_string(value)
         return sanitize_public_artifact_text(value, fallback_summary="", public_path_roots=public_path_roots)
     if isinstance(value, list):
         return [sanitize_public_tool_output_data(item, public_path_roots=public_path_roots) for item in value]
@@ -177,6 +186,8 @@ def sanitize_public_tool_output_data(
             sanitized[output_key] = sanitize_public_artifact_data(raw_value, public_path_roots=public_path_roots)
         elif not key_is_public_path and _is_sensitive_output_key(key_name):
             sanitized[output_key] = "[REDACTED]"
+        elif not key_is_public_path and is_file_content_key(key_name):
+            sanitized[output_key] = REDACTED_TOOL_RESULT_VALUE
         else:
             sanitized[output_key] = sanitize_public_tool_output_data(raw_value, public_path_roots=public_path_roots)
     return sanitized

src/iac_code/a2a/pipeline_events.py

@@ -1366,13 +1366,29 @@ def _artifact_from_spec(spec: Any, root: dict[str, Any]) -> dict[str, Any] | Non
     media_type = _artifact_spec_field(spec, "media_type") or _artifact_spec_field(spec, "mediaType") or "auto"
     if media_type == "auto":
         media_type = _media_type_for_filename(path)
+    role = _artifact_spec_field(spec, "role") or "final"
+    supersedes_path = path
+    supersedes_expression = _artifact_spec_field(spec, "supersedes_path") or _artifact_spec_field(
+        spec, "supersedesPath"
+    )
+    if supersedes_expression is not None:
+        resolved_supersedes_path = _resolve_artifact_expression(root, supersedes_expression)
+        if isinstance(resolved_supersedes_path, str):
+            supersedes_path = resolved_supersedes_path
 
     try:
         filename = artifact_filename_from_path(path)
     except UnsafeArtifactNameError:
         filename = "artifact.txt"
 
-    return {"filename": filename, "mediaType": media_type, "content": content}
+    return {
+        "filename": filename,
+        "mediaType": media_type,
+        "content": content,
+        "role": role,
+        "supersedesPath": sanitize_public_artifact_text(supersedes_path),
+        "supersedesKey": fingerprint_text(supersedes_path),
+    }
 
 
 def _artifact_spec_field(spec: Any, field_name: str) -> str | None:

src/iac_code/a2a/pipeline_executor.py

@@ -12,6 +12,7 @@
 from typing import Any
 
 import httpx
+import yaml
 from a2a.types import Message, Role, TaskState, TaskStatus, TaskStatusUpdateEvent
 from a2a.utils.errors import InvalidParamsError
 
@@ -51,7 +52,8 @@
 from iac_code.pipeline.engine.cleanup import CleanupLedger
 from iac_code.pipeline.engine.events import PipelineEvent, PipelineEventType
 from iac_code.pipeline.engine.handoff import build_handoff_summary, terminal_outcome_from_completed_event
-from iac_code.pipeline.engine.loader import load_pipeline_dir
+from iac_code.pipeline.engine.loader import _resolve_feature_flags, load_pipeline_dir
+from iac_code.pipeline.engine.prerequisites import inspect_prerequisites
 from iac_code.pipeline.engine.public_errors import public_error
 from iac_code.pipeline.engine.session import PipelineSession
 from iac_code.pipeline.engine.user_input import PipelineUserInput, normalize_pipeline_user_input
@@ -398,7 +400,16 @@ def fresh_pipeline_factory() -> Any:
                     )
                 if selected.pipeline is not pipeline:
                     pipeline = selected.pipeline
+                    publisher = self._publisher(
+                        event_queue=event_queue,
+                        pipeline=pipeline,
+                        task_id=task_id,
+                        context_id=context_id,
+                        session_id=ctx.session_id,
+                        cwd=cwd,
+                    )
                     pipeline_runtime.pipeline = pipeline
+                    pipeline_runtime.publisher = publisher
                     self._task_store.mirror_context(ctx)
                 stream = selected.stream
                 ctx.active_task_id = task.task_id
@@ -947,8 +958,19 @@ def _create_pipeline(
         session_storage: SessionStorage,
         resume_from_sidecar: bool = True,
     ) -> Any:
+        pipeline_name = get_pipeline_name()
+        prerequisite_resolution = self._inspect_pipeline_prerequisite_metadata(
+            pipeline_name=pipeline_name,
+            cwd=cwd,
+            session_id=session_id,
+            session_storage=session_storage,
+            resume_from_sidecar=resume_from_sidecar,
+        )
+        create_kwargs: dict[str, Any] = {}
+        if prerequisite_resolution is not None:
+            create_kwargs["prerequisite_resolution"] = prerequisite_resolution
         return create_pipeline(
-            get_pipeline_name(),
+            pipeline_name,
             provider_manager=runtime.provider_manager,
             base_tool_registry=runtime.tool_registry,
             session_storage=session_storage,
@@ -957,8 +979,64 @@ def _create_pipeline(
             resume_from_sidecar=resume_from_sidecar,
             surface="a2a",
             backup_service=self._backup_service,
+            **create_kwargs,
         )
 
+    def _inspect_pipeline_prerequisite_metadata(
+        self,
+        *,
+        pipeline_name: str,
+        cwd: str,
+        session_id: str,
+        session_storage: SessionStorage,
+        resume_from_sidecar: bool,
+    ) -> dict[str, Any] | None:
+        if resume_from_sidecar:
+            sidecar_metadata = self._sidecar_prerequisite_metadata(
+                cwd=cwd,
+                session_id=session_id,
+                session_storage=session_storage,
+            )
+            if sidecar_metadata is not None:
+                return sidecar_metadata
+
+        raw = self._load_pipeline_raw_config(pipeline_name)
+        raw_prerequisites = raw.get("prerequisites") or {}
+        if not isinstance(raw_prerequisites, dict):
+            raw_prerequisites = {}
+        raw_feature_flags = raw.get("feature_flags")
+        feature_flags = _resolve_feature_flags(raw_feature_flags if isinstance(raw_feature_flags, dict) else None)
+        resolution = inspect_prerequisites(raw_prerequisites, feature_flags=feature_flags)
+        return resolution.to_metadata()
+
+    def _sidecar_prerequisite_metadata(
+        self,
+        *,
+        cwd: str,
+        session_id: str,
+        session_storage: SessionStorage,
+    ) -> dict[str, Any] | None:
+        try:
+            raw_session_dir = session_storage.session_dir(cwd, session_id)
+            meta_path = Path(raw_session_dir) / "pipeline" / "meta.yaml"
+            if not meta_path.exists():
+                return None
+            raw = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
+        except Exception:
+            logger.debug("Failed to peek A2A pipeline sidecar prerequisites", exc_info=True)
+            return None
+        if not isinstance(raw, dict):
+            return None
+        metadata = raw.get("prerequisites")
+        return dict(metadata) if isinstance(metadata, dict) else None
+
+    def _load_pipeline_raw_config(self, pipeline_name: str) -> dict[str, Any]:
+        pipeline_dir = discover_pipelines().get(pipeline_name)
+        if pipeline_dir is None:
+            return {}
+        raw = yaml.safe_load((pipeline_dir / "pipeline.yaml").read_text(encoding="utf-8")) or {}
+        return raw if isinstance(raw, dict) else {}
+
     def _set_pipeline_telemetry_correlation(self, pipeline: Any, *, task_id: str, context_id: str) -> None:
         set_correlation = getattr(pipeline, "set_telemetry_correlation", None)
         if not callable(set_correlation):

src/iac_code/a2a/pipeline_snapshot.py

@@ -218,7 +218,7 @@ def _hydrate_existing_snapshot(self, existing_snapshot: dict[str, Any] | None) -
         self._hydrate_messages()
         self._hydrate_display_indexes("candidateDetails", self._candidate_detail_indexes, ("detailId", "id"))
         self._hydrate_display_indexes("diagrams", self._diagram_indexes, ("diagramId", "id"))
-        self._hydrate_display_indexes("artifacts", self._artifact_indexes, ("artifactId", "artifact_id", "id"))
+        self._hydrate_artifact_indexes()
         self._hydrate_display_indexes("permissions", self._permission_indexes, ("permissionId", "toolUseId", "id"))
         self._hydrate_display_indexes("toolResults", self._tool_result_indexes, ("toolUseId", "resultId", "id"))
         self._hydrate_control_history("inputHistory", self._input_history_keys)
@@ -294,6 +294,26 @@ def _hydrate_messages(self) -> None:
                 valid_messages.append(message)
         self._snapshot["display"]["messages"] = valid_messages
 
+    def _hydrate_artifact_indexes(self) -> None:
+        unique_items: list[dict[str, Any]] = []
+        seen_item_keys: set[str] = set()
+        for item in self._snapshot["display"]["artifacts"]:
+            if not isinstance(item, dict):
+                continue
+            event_id = _string_or_none(item.get("eventId"))
+            if event_id is not None:
+                self._seen_event_ids.add(event_id)
+            item_id = _first_string_value(item, ("artifactId", "artifact_id", "id", "eventId"))
+            if item_id is None:
+                continue
+            item_key = _artifact_replacement_key(item_id, item, None)
+            if item_key in seen_item_keys:
+                continue
+            seen_item_keys.add(item_key)
+            self._artifact_indexes[item_key] = len(unique_items)
+            unique_items.append(item)
+        self._snapshot["display"]["artifacts"] = unique_items
+
     def _hydrate_display_indexes(
         self,
         display_key: str,
@@ -788,9 +808,10 @@ def _upsert_artifact_item(self, event: dict[str, Any]) -> None:
         _merge_event_coordinates(item, event)
 
         items = self._snapshot["display"]["artifacts"]
-        index = self._artifact_indexes.get(item_id)
+        item_key = _artifact_replacement_key(item_id, item, event)
+        index = self._artifact_indexes.get(item_key)
         if index is None:
-            self._artifact_indexes[item_id] = len(items)
+            self._artifact_indexes[item_key] = len(items)
             items.append(item)
         else:
             existing = copy.deepcopy(items[index])
@@ -1460,6 +1481,43 @@ def _artifact_item_id(artifact: dict[str, Any] | None, data: dict[str, Any], eve
     return f"artifact-{_sequence_value(event)}"
 
 
+def _artifact_replacement_key(
+    item_id: str,
+    item: dict[str, Any],
+    event: dict[str, Any] | None,
+) -> str:
+    supersedes_key = (
+        _string_or_none(item.get("supersedesKey"))
+        or _string_or_none(item.get("supersedes_key"))
+        or _string_or_none(item.get("supersedesFingerprint"))
+        or _string_or_none(item.get("supersedes_fingerprint"))
+    )
+    if supersedes_key is not None:
+        return f"supersedes-key:{_artifact_replacement_scope(item, event)}:{supersedes_key}"
+    supersedes_path = _public_supersedes_path_for_replacement(item)
+    if supersedes_path is None:
+        return f"id:{item_id}"
+    return f"supersedes:{_artifact_replacement_scope(item, event)}:{supersedes_path}"
+
+
+def _public_supersedes_path_for_replacement(item: dict[str, Any]) -> str | None:
+    supersedes_path = _string_or_none(item.get("supersedesPath")) or _string_or_none(item.get("supersedes_path"))
+    if supersedes_path in {None, "[PATH]"}:
+        return None
+    return supersedes_path
+
+
+def _artifact_replacement_scope(item: dict[str, Any], event: dict[str, Any] | None) -> str:
+    for source in (event, item):
+        if not isinstance(source, dict):
+            continue
+        candidate = _dict_or_none(source.get("candidate"))
+        run_id = _run_id(candidate)
+        if run_id is not None:
+            return f"candidate:{run_id}"
+    return "pipeline"
+
+
 def _drop_legacy_artifact_uri(data: dict[str, Any] | None) -> None:
     if data is None:
         return

src/iac_code/a2a/pipeline_stream.py

@@ -39,6 +39,7 @@
 PENDING_BACKUP_VISIBILITY = "pending_backup"
 COMMITTED_BACKUP_VISIBILITY = "committed"
 BACKUP_COMMITTED_EVENT_TYPE = "backup_committed"
+_ARTIFACT_SEMANTIC_METADATA_KEYS = ("role", "supersedesPath", "supersedesKey", "supersedesFingerprint")
 _RECOVERY_SEMANTIC_EVENT_TYPES = {
     "pipeline_started",
     "pipeline_resumed",
@@ -748,11 +749,14 @@ async def _maybe_externalize_artifact(
 
         try:
             if event_type == "artifact_created":
-                result = {"artifact": envelope.get("artifact")}
+                original_artifact = envelope.get("artifact")
+                result = {"artifact": original_artifact}
             elif tool_result is not None:
+                original_artifact = None
                 result = tool_result.result
             else:
                 return None
+            artifact_semantic_metadata = _artifact_semantic_metadata(original_artifact, envelope.get("data"))
             artifact_metadata = _extract_artifact_metadata(result, self.artifact_store)
         except Exception:
             logger.warning("Failed to externalize A2A pipeline tool artifact", exc_info=True)
@@ -763,7 +767,7 @@ async def _maybe_externalize_artifact(
         envelope["eventType"] = "artifact_created"
         envelope["scope"] = envelope.get("scope") or "pipeline"
         envelope["status"] = "working"
-        envelope["artifact"] = artifact_metadata
+        envelope["artifact"] = {**artifact_metadata, **artifact_semantic_metadata}
         base_data: dict[str, Any] = {}
         envelope_data = envelope.get("data")
         if event_type == "artifact_created" and isinstance(envelope_data, dict):
@@ -776,6 +780,7 @@ async def _maybe_externalize_artifact(
             "byteSize": artifact_metadata.get("byteSize"),
             "sha256": artifact_metadata.get("sha256"),
             "uri": artifact_metadata.get("uri"),
+            **artifact_semantic_metadata,
         }
         if tool_result is not None and A2AExposureType.TOOL_TRACE in self.exposure_types:
             envelope["data"].update(
@@ -848,6 +853,18 @@ def _tool_result_from(event: Any) -> ToolResultEvent | None:
     return inner if isinstance(inner, ToolResultEvent) else None
 
 
+def _artifact_semantic_metadata(*sources: Any) -> dict[str, Any]:
+    metadata: dict[str, Any] = {}
+    for source in sources:
+        if not isinstance(source, dict):
+            continue
+        for key in _ARTIFACT_SEMANTIC_METADATA_KEYS:
+            value = source.get(key)
+            if isinstance(value, str) and value:
+                metadata.setdefault(key, value)
+    return metadata
+
+
 def _unwrap_stream_event(event: Any) -> Any:
     while isinstance(event, SubPipelineStreamEvent):
         event = event.inner