"""Shared lifecycle and helpers for automatic resource processors.""" import asyncio import hashlib import re import uuid from abc import abstractmethod from collections.abc import Mapping from contextlib import contextmanager from dataclasses import dataclass, field from pathlib import Path, PurePosixPath from typing import Any import frontmatter from agentscope.message import DataBlock, TextBlock, UserMsg from watchfiles import Change from ...components.runtime_context import RuntimeContext from ..base_step import BaseStep from ..file_io import refresh_day_index, validate_filename_component from ..file_io._path import is_relative_to, resolve_path from ._evolve import agent_reply_result_text, now _SOURCE_RESOURCE_KEY = "source_resource" _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$") _UNSAFE_FILENAME_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]+') _LOOKUP_TABLE_KEY = "_auto_resource_lookup_table" @dataclass class _ResourceLookupTable: """Initial source → day/path ownership, shared by this invocation's processors. Build once without refreshes. Repeated sources use the original live lookup. None means history has not been read; an empty dict is a valid result. """ owners: dict[str, dict[str, str]] | None = None seen_sources: set[str] = field(default_factory=set) @contextmanager def _resource_lookup_scope(context: RuntimeContext, *, reuse: bool = False): """Share a router's index with processors, never with the next watch batch.""" previous = context.get(_LOOKUP_TABLE_KEY) if reuse and isinstance(previous, _ResourceLookupTable): yield previous return existed = _LOOKUP_TABLE_KEY in context table = _ResourceLookupTable() context[_LOOKUP_TABLE_KEY] = table try: yield table finally: table.owners = None table.seen_sources.clear() if existed: context[_LOOKUP_TABLE_KEY] = previous else: del context[_LOOKUP_TABLE_KEY] @dataclass(frozen=True) class _ResourceNoteState: """Snapshot a linked note, or a collision-free path reserved for a new one.""" path: str created: bool before_bytes: bytes | None def _compute_note_stem(filename: str) -> str: """Return the daily note stem for a resource filename.""" return PurePosixPath(filename).stem def _compute_agent_session_id(path: str) -> str: """Return a stable UUID session id for agent backends.""" return str(uuid.uuid5(uuid.NAMESPACE_URL, path)) def _parse_resource_path(file_path: str, resource_dir: str) -> tuple[str, str]: """Extract (date, filename) from a resource path like 'resource/2026-06-06/report.pdf'. Returns (date_str, filename) where filename may contain subdirectories. """ parts = PurePosixPath(file_path).parts # Strip leading resource_dir prefix prefix_parts = PurePosixPath(resource_dir).parts if parts[: len(prefix_parts)] != prefix_parts: return "", "" parts = parts[len(prefix_parts) :] # First segment is date, rest is filename date_str = parts[0] if parts else "" if not _DATE_RE.match(date_str): return "", "" filename = str(PurePosixPath(*parts[1:])) if len(parts) > 1 else "" return date_str, filename def _loose_resource_filename(file_path: str, resource_dir: str) -> str: """Return filename for a root-level resource path like 'resource/report.txt'.""" parts = PurePosixPath(file_path).parts prefix_parts = PurePosixPath(resource_dir).parts if parts[: len(prefix_parts)] != prefix_parts: return "" rest = parts[len(prefix_parts) :] if len(rest) != 1: return "" filename = rest[0] return "" if filename in ("", ".", "..") else filename def _results_answer(results: list[dict], processed_answer: str) -> str: """Return the actual per-change answer while preserving a batch fallback.""" answers = [str(item.get("answer") or "").strip() for item in results] answers = [item for item in answers if item] if len(answers) == 1: return answers[0] if len(answers) > 1: return "\n\n".join(f"{index}. {answer}" for index, answer in enumerate(answers, start=1)) return processed_answer def _source_suffix(file_path: str) -> str: """Return a short stable suffix for source-path collision handling.""" return hashlib.sha1(file_path.encode("utf-8")).hexdigest()[:8] def _sanitize_note_name(raw: str, fallback: str) -> str: """Return a safe single filename component from an LLM-suggested name.""" name = str(raw or "").strip() name = _UNSAFE_FILENAME_CHARS.sub("-", name) name = re.sub(r"\s+", " ", name).strip(" .") if not name: name = str(fallback or "").strip() name = _UNSAFE_FILENAME_CHARS.sub("-", name).strip(" .") if not name or validate_filename_component(name, kind="name"): name = f"resource-{_source_suffix(fallback or raw or 'note')}" if validate_filename_component(name, kind="name"): name = f"resource-{_source_suffix(name)}" return name class BaseAutoResourceStep(BaseStep): """Shared source-linked daily-note lifecycle for resource processors.""" _resource_lookup: _ResourceLookupTable | None = None resource_fallback = False resource_suffixes: frozenset[str] = frozenset() router_inherit_keys = frozenset({"file_store", "language"}) def __init__(self, **kwargs): super().__init__(**kwargs) self.create_tools: list[str] = ["write"] self.update_tools: list[str] = ["read", "edit", "frontmatter_update", "write"] @classmethod def matches_change(cls, change: Mapping[str, Any]) -> bool: """Return whether this processor accepts a change before fallback. The predicate must stay synchronous and file-system independent because deleted resources no longer exist when the router evaluates them. """ file_path = change.get("path") or change.get("file_path", "") return Path(str(file_path)).suffix.lower() in cls.resource_suffixes def _normalize_change(self, raw) -> Change | None: if isinstance(raw, Change): return raw if isinstance(raw, str): return Change.__members__.get(raw) return None def _today(self) -> str: tz = self.app_context.app_config.timezone if self.app_context is not None else None return now(tz).strftime("%Y-%m-%d") def _daily_note_path(self, day: str, name: str) -> str: return f"{self.config_value('daily_dir')}/{day}/{name}.md" def _resource_directory(self) -> tuple[str, Path]: """Return the workspace-relative identity and resolved resource root.""" workspace = self.workspace_path.resolve() configured = str(self.config_value("resource_dir")) resolved, error = resolve_path(workspace, configured) if error or resolved is None: raise ValueError(f"invalid resource_dir {configured!r}: {error or 'cannot resolve path'}") logical = Path(configured) if logical.is_absolute(): for workspace_variant in (self.workspace_path.absolute(), workspace): try: logical = logical.relative_to(workspace_variant) break except ValueError: continue else: raise ValueError("resource_dir must stay inside the workspace") return logical.as_posix(), resolved def _resolve_resource_source(self, raw_path: str) -> tuple[str, Path, str]: """Return a logical resource path, safe read path, and logical resource root. The logical path is kept for ``source_resource`` provenance. The resolved path is used for stat/read so a symlink cannot escape the configured resource directory. """ value = str(raw_path or "").strip() if not value: raise ValueError("resource path is required") supplied = Path(value) if any(part in {".", ".."} for part in supplied.parts): raise ValueError(f"resource path cannot contain '.' or '..': {value!r}") workspace = self.workspace_path.resolve() resource_dir, resource_root = self._resource_directory() resolved, path_error = resolve_path(workspace, value) if path_error or resolved is None: raise ValueError(f"invalid resource path {value!r}: {path_error or 'cannot resolve path'}") if not is_relative_to(resolved, resource_root): raise ValueError("resource path must stay inside the configured resource directory") if supplied.is_absolute(): try: logical = supplied.relative_to(self.workspace_path.absolute()) except ValueError: try: logical = supplied.relative_to(workspace) except ValueError as exc: raise ValueError("resource path must stay inside the workspace") from exc else: logical = supplied resource_logical = Path(resource_dir) try: logical.relative_to(resource_logical) except ValueError as exc: raise ValueError("resource path must stay inside the configured resource directory") from exc return logical.as_posix(), resolved, resource_dir @staticmethod def _source_resource_link(file_path: str) -> str: return f"[[{file_path}]]" def _frontmatter(self, path: str) -> dict: data = self._note_bytes(path) if data is None: raise FileNotFoundError(path) post = frontmatter.loads(data.decode("utf-8")) return dict(post.metadata or {}) def _note_bytes(self, path: str) -> bytes | None: note_path, error = resolve_path(self.file_store.workspace_path, path) if error or note_path is None: raise ValueError(f"invalid resource note path {path!r}: {error}") if not note_path.is_file(): return None return note_path.read_bytes() def _note_modified(self, before_path: str, before_bytes: bytes | None, after_path: str) -> bool: if not after_path: return False after_bytes = self._note_bytes(after_path) if after_bytes is None: return before_bytes is not None return after_path != before_path or before_bytes != after_bytes async def _refresh_day_index(self, day: str) -> dict: """Refresh and return the derived daily index for a resource-note change.""" daily_dir = self.config_value("daily_dir") self.logger.info(f"[{self.name}] refresh index start date={day} daily_dir={daily_dir}") index_payload = await refresh_day_index(self.file_store, day, daily_dir) self.logger.info(f"[{self.name}] refresh index done date={day}") return index_payload def _find_resource_note(self, notes: list[dict], file_path: str) -> dict | None: """Return only a note explicitly owned by ``file_path``.""" source = self._source_resource_link(file_path) for note in notes: if str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() == source: return note return None async def _list_resource_note(self, day: str, file_path: str) -> dict | None: notes = await self._list_daily_notes(day) return self._find_resource_note(notes, file_path) async def _lookup_resource_note_path(self, day: str, file_path: str) -> str | None: """Reuse the initial path before processing a source, not after writing it.""" if self._resource_lookup is not None: assert self._resource_lookup.owners is not None source = self._source_resource_link(file_path) return self._resource_lookup.owners.get(source, {}).get(day) note = await self._list_resource_note(day, file_path) return str(note["path"]) if note is not None else None async def _list_daily_notes(self, day: str) -> list[dict]: """Use the configured daily_list job for both history and targeted checks.""" list_response = await self.run_job("daily_list", date=day) if not list_response.success: raise RuntimeError(f"daily_list failed: {list_response.answer}") return list_response.metadata.get("notes") or [] def _daily_note_days(self) -> list[str]: """Return safe, deterministic daily subdirectories available for lookup.""" workspace = self.workspace_path.resolve() daily_dir = str(self.config_value("daily_dir")) daily_root, error = resolve_path(workspace, daily_dir) if error or daily_root is None: raise ValueError(f"invalid daily_dir {daily_dir!r}: {error or 'cannot resolve path'}") if not daily_root.is_dir(): return [] days = [] for entry in daily_root.iterdir(): if not _DATE_RE.fullmatch(entry.name): continue try: resolved, path_error = resolve_path(workspace, f"{daily_dir}/{entry.name}") if not path_error and resolved is not None and resolved.is_dir(): days.append(entry.name) except (OSError, RuntimeError): # Broken or cyclic links must not prevent lookup in other days. continue return sorted(days) async def _find_loose_resource_day(self, file_path: str) -> str | None: """Use initial ownership once per source; repeat events retain the original lookup.""" assert self.context is not None self._resource_lookup = None with _resource_lookup_scope(self.context, reuse=True) as table: if file_path in table.seen_sources: matches = [] for day in self._daily_note_days(): note = await self._list_resource_note(day, file_path) if note is not None: matches.append((day, str(note["path"]))) else: table.seen_sources.add(file_path) if table.owners is None: owners = {} for day in self._daily_note_days(): for note in await self._list_daily_notes(day): source = str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() if source: # Preserve daily_list's first match within each day. owners.setdefault(source, {}).setdefault(day, str(note["path"])) # Publish only after the entire initial scan succeeds. table.owners = owners self._resource_lookup = table source = self._source_resource_link(file_path) matches = sorted(table.owners.get(source, {}).items()) if len(matches) > 1: paths = ", ".join(path for _, path in matches) raise RuntimeError(f"Multiple daily resource notes claim {file_path}: {paths}") return matches[0][0] if matches else None async def _prepare_resource_note(self, day: str, file_path: str, note_stem: str) -> _ResourceNoteState: """Find the owned note or allocate a safe path before the first write.""" note_path = await self._lookup_resource_note_path(day, file_path) if note_path is not None: return _ResourceNoteState(path=note_path, created=False, before_bytes=self._note_bytes(note_path)) _, note_path = self._unique_daily_note_path(day, note_stem, file_path, current_path="") return _ResourceNoteState(path=note_path, created=True, before_bytes=None) async def _interpret_resource( self, file_path: str, day: str, note_stem: str, added: bool, file_content: str, *, input_blocks: list[TextBlock | DataBlock] | None = None, resource_instructions: str = "", note_metadata: dict | None = None, reply_kwargs: dict | None = None, ) -> str | None: """Run the same note-writing agent for text and native multimodal inputs.""" state = await self._prepare_resource_note(day, file_path, note_stem) prompt_name = "user_message_create" if state.created else "user_message_update" prompt = self.prompt_format( prompt_name, workspace_dir=str(self.workspace_path), note_path=state.path, note_stem=note_stem, file_path=file_path, source_resource=self._source_resource_link(file_path), file_content=file_content, resource_instructions=f"\n\n{resource_instructions}" if resource_instructions else "", date=day, ) if resource_instructions and "{resource_instructions}" not in self.get_prompt(prompt_name): prompt = f"{prompt}\n\n{resource_instructions}" inputs = UserMsg(name="user", content=[*input_blocks, TextBlock(text=prompt)]) if input_blocks else prompt self.logger.info(f"[{self.name}] agent start file_path={file_path} note_path={state.path}") agent_kwargs = { "system_prompt": self.prompt_format("system_prompt"), "job_tools": self.create_tools if state.created else self.update_tools, **(reply_kwargs or {}), } if "session_id" not in agent_kwargs: agent_kwargs["session_id"] = _compute_agent_session_id(file_path) # Consume the resource-only option before forwarding kwargs to the wrapper. if agent_kwargs.pop("scope_note_tools", False): # Bind the final allocated path last, never a path supplied by the agent. agent_kwargs["injected_job_kwargs"] = { **(agent_kwargs.get("injected_job_kwargs") or {}), "file_store": self.file_store.name, "_allowed_paths": [state.path], } try: result = await self.agent_wrapper.reply(inputs, **agent_kwargs) except (Exception, asyncio.CancelledError): self.context.response.success = False try: await self._recover_resource_note(state, day, file_path, note_stem, added, note_metadata) except Exception: self.logger.exception(f"[{self.name}] failed to finalize resource after agent error: {file_path}") raise note_path = await self._finalize_resource_note( state, day, file_path, note_stem, added, metadata=note_metadata, ) self.context.response.success = True self.context.response.answer = agent_reply_result_text(result) if note_path is None: self.logger.info(f"[{self.name}] done without note file_path={file_path} modified=False") return None session_id = agent_kwargs.get("session_id") if session_id is None and isinstance(result, Mapping): session_id = result.get("session_id") if session_id: self.context.response.metadata["agent_session_id"] = session_id self.logger.info( f"[{self.name}] agent done file_path={file_path} modified={self.context.response.metadata['modified']}", ) return note_path async def _recover_resource_note( self, state: _ResourceNoteState, day: str, file_path: str, note_stem: str, added: bool, metadata: dict | None, ) -> None: """Finalize only a changed, explicitly owned note, without claiming another file.""" after_bytes = self._note_bytes(state.path) modified = after_bytes != state.before_bytes self.context.response.metadata.update( { "path": state.path, "created": state.created and after_bytes is not None, "modified": modified, }, ) if ( modified and after_bytes is not None and str(self._frontmatter(state.path).get(_SOURCE_RESOURCE_KEY, "")).strip() == self._source_resource_link(file_path) ): await self._finalize_resource_note( state, day, file_path, note_stem, added, metadata=metadata, ) def _validate_resource_note(self, path: str, file_path: str, before_bytes: bytes | None) -> None: """Optional modality-specific acceptance check; text keeps its existing behavior.""" async def _resolve_written_note( self, state: _ResourceNoteState, day: str, file_path: str, ) -> str | None: """Resolve a processor write without claiming a pre-existing same-stem note.""" if not state.created: if self._note_bytes(state.path) is None: raise RuntimeError(f"linked resource note disappeared: {state.path}") return state.path note = await self._list_resource_note(day, file_path) if note is not None: return str(note["path"]) if self._note_bytes(state.path) is None: return None # The path was absent when this invocation allocated it, so a note # created there without provenance can be repaired conservatively. A # different explicit owner is always a conflict and is never claimed. source = str(self._frontmatter(state.path).get(_SOURCE_RESOURCE_KEY, "")).strip() expected = self._source_resource_link(file_path) if source and source != expected: raise RuntimeError(f"resource note path is owned by another source: {state.path}") return state.path async def _ensure_resource_frontmatter(self, path: str, file_path: str, metadata: dict | None = None) -> None: metadata = {**(metadata or {}), _SOURCE_RESOURCE_KEY: self._source_resource_link(file_path)} current = self._frontmatter(path) if all(current.get(key) == value for key, value in metadata.items()): return response = await self.run_job( "frontmatter_update", path=path, metadata=metadata, ) if not response.success: raise RuntimeError(f"frontmatter_update failed: {response.answer}") async def _set_frontmatter_name(self, path: str, name: str) -> None: if self._frontmatter(path).get("name") == name: return response = await self.run_job("frontmatter_update", path=path, metadata={"name": name}) if not response.success: raise RuntimeError(f"frontmatter_update failed: {response.answer}") def _unique_daily_note_path(self, day: str, name: str, file_path: str, current_path: str) -> tuple[str, str]: """Return a collision-free (name, path), preserving current_path when possible.""" target_path = self._daily_note_path(day, name) target_abs = self.file_store.workspace_path / target_path if target_path == current_path or not target_abs.exists(): return name, target_path suffixed = f"{name}--{_source_suffix(file_path)}" target_path = self._daily_note_path(day, suffixed) target_abs = self.file_store.workspace_path / target_path if target_path == current_path or not target_abs.exists(): return suffixed, target_path for index in range(2, 100): candidate = f"{suffixed}-{index}" target_path = self._daily_note_path(day, candidate) target_abs = self.file_store.workspace_path / target_path if target_path == current_path or not target_abs.exists(): return candidate, target_path raise RuntimeError(f"cannot allocate unique note name for: {name!r}") async def _rename_from_frontmatter_name( self, path: str, day: str, file_path: str, fallback_name: str, *, allow_rename: bool, ) -> str: meta = self._frontmatter(path) current_name = PurePosixPath(path).stem suggested_name = str(meta.get("name", "")).strip() if not allow_rename: name = _sanitize_note_name(current_name, fallback_name) if suggested_name != name: await self._set_frontmatter_name(path, name) return path name = _sanitize_note_name(suggested_name, fallback_name) name, target_path = self._unique_daily_note_path(day, name, file_path, path) if suggested_name != name: await self._set_frontmatter_name(path, name) if target_path == path: return path move_response = await self.run_job( "move", src_path=path, dst_path=target_path, overwrite=False, retarget=True, ) if not move_response.success: raise RuntimeError(f"move failed: {move_response.answer}") return target_path async def _finalize_resource_note( self, state: _ResourceNoteState, day: str, file_path: str, note_stem: str, added: bool, *, metadata: dict | None = None, ) -> str | None: """Resolve, source-link, rename, index, and report one processor write.""" staged_bytes = self._note_bytes(state.path) self.context.response.metadata.update( { "path": state.path if staged_bytes is not None else None, "created": state.created and staged_bytes is not None, "modified": self._note_modified(state.path, state.before_bytes, state.path), }, ) note_path = await self._resolve_written_note(state, day, file_path) if note_path is None: self.context.response.metadata.update({"path": None, "created": False, "modified": False}) return None modified = self._note_modified(state.path, state.before_bytes, note_path) self.context.response.metadata.update({"path": note_path, "created": state.created, "modified": modified}) await self._ensure_resource_frontmatter(note_path, file_path, metadata) note_path = await self._rename_from_frontmatter_name( note_path, day, file_path, note_stem, allow_rename=state.created, ) modified = self._note_modified(state.path, state.before_bytes, note_path) self.context.response.metadata.update({"path": note_path, "created": state.created, "modified": modified}) index_payload = await self._refresh_day_index(day) self.context.response.metadata.update( { "path": note_path, "created": state.created, "modified": modified, "session_id": note_stem, "source_resource": self._source_resource_link(file_path), "action": "added" if added else "modified", "index": index_payload, }, ) self._validate_resource_note(note_path, file_path, state.before_bytes) return note_path async def _handle_delete(self, file_path: str, date_str: str, note_stem: str) -> None: note_rel = await self._lookup_resource_note_path(date_str, file_path) if note_rel is None: self.context.response.success = True self.context.response.answer = f"No linked resource note to delete: {file_path}" self.context.response.metadata.update( { "path": None, "session_id": note_stem, "source_resource": self._source_resource_link(file_path), "action": "skipped", "reason": "resource_note_not_found", "modified": False, }, ) self.logger.info(f"[{self.name}] delete skipped; no owned note file_path={file_path}") return note_abs = self.workspace_path / note_rel note_existed = note_abs.is_file() self.logger.info(f"[{self.name}] delete start note={note_rel}") if note_existed: note_abs.unlink() self.logger.info(f"[{self.name}] Deleted file: {note_rel}") self.context.response.metadata.update( { "path": note_rel, "session_id": note_stem, "source_resource": self._source_resource_link(file_path), "action": "deleted", "modified": note_existed, }, ) await self.file_store.delete([note_rel]) self.logger.info(f"[{self.name}] catalog delete done note={note_rel}") index_payload = await self._refresh_day_index(date_str) self.context.response.success = True self.context.response.answer = f"Deleted resource note: {note_rel}" self.context.response.metadata["index"] = index_payload @abstractmethod async def _handle_upsert( self, file_path: str, date_str: str, note_stem: str, added: bool, source_path: Path, ) -> None: """Interpret one added or modified resource into its daily note.""" def _skip_resource_change(self, file_path: str) -> bool: """Allow a processor to disable its lifecycle after common input validation.""" del file_path return False async def _handle_change(self, file_path: str, raw_change) -> dict: assert self.context is not None self._resource_lookup = None # Handlers write item-scoped fields into the shared response. Start each # change with a fresh mapping so one result cannot inherit another's metadata. self.context.response.metadata = {} if not file_path: self.context.response.success = False self.context.response.answer = "Missing file_path" self.logger.warning(f"[{self.name}] missing file_path change={raw_change!r}") return {"success": False, "path": file_path, "change": raw_change, "answer": self.context.response.answer} change = self._normalize_change(raw_change) if change is None: self.context.response.success = False self.context.response.answer = f"Invalid change type: {raw_change}" self.logger.warning(f"[{self.name}] invalid change file_path={file_path} change={raw_change!r}") return {"success": False, "path": file_path, "change": raw_change, "answer": self.context.response.answer} try: file_path, source_path, resource_dir = self._resolve_resource_source(file_path) except ValueError as exc: self.context.response.success = False self.context.response.answer = str(exc) self.context.response.metadata.update( {"path": str(file_path), "action": "failed", "error": str(exc), "modified": False}, ) self.logger.warning(f"[{self.name}] invalid resource path file_path={file_path!r} error={exc}") return { "success": False, "path": str(file_path), "change": change.name, "answer": self.context.response.answer, "metadata": dict(self.context.response.metadata), } if self._skip_resource_change(file_path): return { "success": self.context.response.success, "path": file_path, "change": change.name, "answer": self.context.response.answer, "metadata": dict(self.context.response.metadata), } loose_filename = _loose_resource_filename(file_path, resource_dir) if loose_filename: existing_day = await self._find_loose_resource_day(file_path) date_str, filename = existing_day or self._today(), loose_filename self.logger.info( f"[{self.name}] loose resource file_path={file_path} date={date_str} existing={bool(existing_day)}", ) else: date_str, filename = _parse_resource_path(file_path, resource_dir) if not date_str or not filename: self.context.response.success = False self.context.response.answer = f"Cannot parse date/filename from: {file_path}" self.logger.warning(f"[{self.name}] parse path failed file_path={file_path} resource_dir={resource_dir}") return {"success": False, "path": file_path, "change": change.name, "answer": self.context.response.answer} note_stem = _compute_note_stem(filename) self.logger.info(f"[{self.name}] {change.name} file_path={file_path} note_stem={note_stem}") if change == Change.deleted: await self._handle_delete(file_path, date_str, note_stem) else: await self._handle_upsert( file_path, date_str, note_stem, change == Change.added, source_path, ) return { "success": self.context.response.success, "path": file_path, "change": change.name, "answer": self.context.response.answer, "metadata": dict(self.context.response.metadata), } def _failed_change_result(self, file_path: str, raw_change, exc: Exception) -> dict: """Convert one unexpected processor error into an item-scoped result.""" assert self.context is not None file_path = str(file_path or "") if Path(file_path).is_absolute(): file_path = self.to_workspace_relative(file_path) change = self._normalize_change(raw_change) change_name = change.name if change is not None else str(raw_change) answer = f"Failed to process resource: {file_path}: {exc}" metadata = dict(self.context.response.metadata) metadata.setdefault("path", file_path) metadata.setdefault("modified", False) metadata.update({"action": "failed", "error": str(exc)}) self.context.response.success = False self.context.response.answer = answer self.context.response.metadata = metadata self.logger.exception(f"[{self.name}] resource failed file_path={file_path} error={exc}") return { "success": False, "path": file_path, "change": change_name, "answer": answer, "metadata": dict(metadata), } async def execute(self): assert self.context is not None with _resource_lookup_scope(self.context, reuse=True): return await self._execute_changes() async def _execute_changes(self): """Process a sub-batch within the router's or a standalone lookup scope.""" assert self.context is not None changes = self.context.get("changes") if not isinstance(changes, list): self.context.response.success = False self.context.response.answer = "AutoResourceStep requires changes: list[dict]" self.logger.warning(f"[{self.name}] invalid changes payload type={type(changes).__name__}") return self.context.response self.logger.info(f"[{self.name}] start changes={len(changes)}") results = [] for index, item in enumerate(changes, start=1): if not isinstance(item, dict): self.logger.warning(f"[{self.name}] skip invalid change item index={index} type={type(item).__name__}") continue self.logger.info(f"[{self.name}] process change {index}/{len(changes)}") file_path = item.get("path") or item.get("file_path", "") raw_change = item.get("change", "") self.context.response.metadata = {} try: result = await self._handle_change(file_path, raw_change) except Exception as exc: # pylint: disable=broad-except result = self._failed_change_result(file_path, raw_change, exc) results.append(result) success_count = sum(1 for item in results if item.get("success")) self.context.response.success = success_count == len(changes) processed_answer = f"Processed {success_count}/{len(changes)} resource change(s)" self.context.response.answer = _results_answer(results, processed_answer) self.context.response.metadata["processed"] = len(results) self.context.response.metadata["results"] = results self.context.response.metadata["modified"] = any( bool((item.get("metadata") or {}).get("modified")) for item in results ) self.logger.info( f"[{self.name}] done success={success_count}/{len(changes)} " f"processed={len(results)} modified={self.context.response.metadata['modified']}", ) return self.context.response