[codex] Add session-scoped backup layout

src/iac_code/a2a/artifacts.py

@@ -12,7 +12,15 @@
 from urllib.parse import quote, unquote, urlsplit
 
 from iac_code.i18n import _
+from iac_code.services.session_layout import (
+    SESSION_LAYOUT_VERSION_V2,
+    SessionPaths,
+    UnsupportedSessionLayoutError,
+    ensure_session_owned_dir,
+    require_supported_session_layout,
+)
 from iac_code.utils.public_errors import sanitize_public_text
+from iac_code.utils.state_io import atomic_write_bytes
 
 PUBLIC_ARTIFACT_URI_PREFIX = "iac-code-artifact://"
 _PUBLIC_ARTIFACT_URI_SCHEME = "iac-code-artifact"
@@ -333,8 +341,9 @@ def to_dict(self) -> dict[str, object]:
 
 
 class A2AArtifactStore:
-    def __init__(self, root: str | Path) -> None:
+    def __init__(self, root: str | Path, *, session_dir: str | Path | None = None) -> None:
         self.root = Path(root)
+        self._session_dir = Path(session_dir) if session_dir is not None else None
 
     def save_text(self, *, filename: str, content: str, media_type: str) -> A2AArtifactMetadata:
         encoded = content.encode("utf-8")
@@ -348,9 +357,12 @@ def save_bytes(self, *, filename: str, content: bytes, media_type: str) -> A2AAr
         safe_name = self._safe_filename(filename)
         artifact_id = str(uuid.uuid4())
         artifact_dir = self.root / artifact_id
-        artifact_dir.mkdir(parents=True, exist_ok=False)
+        if self._session_dir is not None:
+            ensure_session_owned_dir(self._session_dir, artifact_dir)
+        else:
+            artifact_dir.mkdir(parents=True, exist_ok=False)
         path = artifact_dir / safe_name
-        path.write_bytes(content)
+        atomic_write_bytes(path, content)
         return A2AArtifactMetadata(
             artifact_id=artifact_id,
             filename=safe_name,
@@ -373,3 +385,13 @@ def _public_uri(artifact_id: str, filename: str) -> str:
     @staticmethod
     def _safe_filename(filename: str) -> str:
         return safe_artifact_filename(filename)
+
+
+def artifact_store_for_session(session_dir: str | Path) -> A2AArtifactStore:
+    session_path = Path(session_dir)
+    version = require_supported_session_layout(session_path)
+    if version != SESSION_LAYOUT_VERSION_V2:
+        raise UnsupportedSessionLayoutError(_("Unsupported session layout version: {version}").format(version="legacy"))
+    session_paths = SessionPaths.from_session_dir(session_path)
+    root = ensure_session_owned_dir(session_path, session_paths.a2a_artifacts_dir)
+    return A2AArtifactStore(root, session_dir=session_path)

src/iac_code/a2a/backup.py

@@ -0,0 +1,91 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from typing import Any
+
+from iac_code.i18n import _
+from iac_code.services.session_backup import BackupReason, SessionBackupBlocked
+
+logger = logging.getLogger(__name__)
+
+
+async def backup_session_async(
+    backup_service: Any,
+    cwd: str,
+    session_id: str,
+    *,
+    reason: BackupReason,
+    critical: bool,
+    metrics: Any | None = None,
+) -> Any | None:
+    failed_recorded = False
+    try:
+        result = await asyncio.to_thread(
+            backup_service.backup_session,
+            cwd,
+            session_id,
+            reason=reason,
+            critical=critical,
+        )
+        retry_count = _retry_count(result)
+        if getattr(result, "enabled", False) and not getattr(result, "succeeded", True):
+            message = str(
+                getattr(result, "error", None) or _("Session backup failed. Retry after the backup path is available.")
+            )
+            _record_backup_failed(metrics, reason=reason, critical=critical, retry_count=retry_count)
+            failed_recorded = True
+            if critical:
+                raise SessionBackupBlocked(message, retry_count=retry_count, result=result)
+            logger.warning(
+                "A2A session backup failed reason=%s critical=%s retry_count=%s: %s",
+                reason.value,
+                critical,
+                retry_count,
+                message,
+            )
+        elif getattr(result, "enabled", False):
+            _record_backup_succeeded(metrics, reason=reason, critical=critical, retry_count=retry_count)
+        return result
+    except Exception as exc:
+        retry_count = _retry_count_from_exception(exc)
+        if not failed_recorded:
+            _record_backup_failed(metrics, reason=reason, critical=critical, retry_count=retry_count)
+        if critical:
+            raise
+        logger.warning(
+            "A2A session backup failed reason=%s critical=%s retry_count=%s error_type=%s",
+            reason.value,
+            critical,
+            retry_count,
+            type(exc).__name__,
+        )
+        return None
+
+
+def _retry_count(result: Any) -> int:
+    value = getattr(result, "retry_count", 0)
+    return value if isinstance(value, int) and value >= 0 else 0
+
+
+def _retry_count_from_exception(exc: BaseException) -> int:
+    value = getattr(exc, "retry_count", 0)
+    return value if isinstance(value, int) and value >= 0 else 0
+
+
+def _record_backup_succeeded(metrics: Any | None, *, reason: BackupReason, critical: bool, retry_count: int) -> None:
+    record = getattr(metrics, "record_backup_succeeded", None)
+    if callable(record):
+        try:
+            record(reason=reason.value, critical=critical, retry_count=retry_count)
+        except Exception as exc:
+            logger.debug("Failed to record A2A backup_succeeded metric: %s", type(exc).__name__)
+
+
+def _record_backup_failed(metrics: Any | None, *, reason: BackupReason, critical: bool, retry_count: int) -> None:
+    record = getattr(metrics, "record_backup_failed", None)
+    if callable(record):
+        try:
+            record(reason=reason.value, critical=critical, retry_count=retry_count)
+        except Exception as exc:
+            logger.debug("Failed to record A2A backup_failed metric: %s", type(exc).__name__)

src/iac_code/a2a/events.py

@@ -455,7 +455,7 @@ async def publish_stream_event(
         if event.error_id:
             error_metadata["errorId"] = event.error_id
         if event.is_retryable:
-            text = "A temporary error occurred. Please retry."
+            text = _("A temporary error occurred. Please retry.")
             state = TaskState.TASK_STATE_INPUT_REQUIRED
         else:
             raw = event.error or "Unknown error"