"""auto_resource — interpret resource files into source-linked daily notes via an agent.""" import hashlib import inspect import re import uuid from pathlib import Path, PurePosixPath import aiofiles import frontmatter from watchfiles import Change from ..base_step import BaseStep from ..file_io import refresh_day_index, validate_filename_component from ...components import R 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]+') 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 _compute_note_stem(filename: str) -> str: """Return the daily note stem for a resource filename.""" return PurePosixPath(filename).stem 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 @R.register("auto_resource_step") class AutoResourceStep(BaseStep): """Interpret resource files into daily notes via an Agent.""" def __init__(self, **kwargs): super().__init__(**kwargs) self.create_tools: list[str] = ["write"] self.update_tools: list[str] = ["read", "edit", "frontmatter_update", "write"] 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" @staticmethod def _source_resource_link(file_path: str) -> str: return f"[[{file_path}]]" def _frontmatter(self, path: str) -> dict: post = frontmatter.loads((self.file_store.workspace_path / path).read_text(encoding="utf-8")) return dict(post.metadata or {}) def _note_bytes(self, path: str) -> bytes | None: note_path = self.file_store.workspace_path / path 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 def _find_resource_note(self, notes: list[dict], file_path: str, fallback_path: str) -> dict | None: source = self._source_resource_link(file_path) for note in notes: if str(note.get(_SOURCE_RESOURCE_KEY, "")).strip() == source: return note for note in notes: if str(note.get("path", "")).strip() == fallback_path: return note return None async def _list_resource_note(self, day: str, file_path: str, fallback_path: str) -> dict | None: list_response = await self.run_job("daily_list", date=day) if not list_response.success: raise RuntimeError(f"daily_list failed: {list_response.answer}") notes = list_response.metadata.get("notes") or [] return self._find_resource_note(notes, file_path, fallback_path) async def _ensure_resource_frontmatter(self, path: str, file_path: str) -> None: metadata = {_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, fallback_path: 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 and path != fallback_path: 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 _emit_result_hook(self, *, changes: list[dict], results: list[dict]) -> None: """Notify embedding hosts about the final auto-resource response. The hook is intentionally optional so standalone ReMe and old configs keep the existing behavior. """ if self.app_context is None or self.context is None: return metadata = getattr(self.app_context, "metadata", None) if not isinstance(metadata, dict): return response_metadata = getattr(self.context.response, "metadata", None) if isinstance(response_metadata, dict) and response_metadata.get("modified") is False: self.logger.info(f"[{self.name}] result hook skipped; no resource note change modified=False") return hook = metadata.get("qwenpaw_memory_result_hook") if hook is None: return try: modified = response_metadata.get("modified") if isinstance(response_metadata, dict) else None self.logger.info(f"[{self.name}] result hook emit modified={modified}") value = hook( job_name="auto_resource", response=self.context.response, kwargs={"changes": changes}, metadata={"results": results}, ) if inspect.isawaitable(value): await value except Exception: self.logger.exception(f"[{self.name}] result hook failed") async def _handle_delete(self, file_path: str, date_str: str, note_stem: str) -> None: daily_dir = self.config_value("daily_dir") fallback_path = f"{daily_dir}/{date_str}/{note_stem}.md" try: note = await self._list_resource_note(date_str, file_path, fallback_path) except RuntimeError as exc: self.context.response.success = False self.context.response.answer = str(exc) self.logger.info(f"[{self.name}] delete list failed file_path={file_path} answer={str(exc)!r}") return note_rel = str(note["path"]) if note else fallback_path 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}") await self.file_store.delete([note_rel]) self.logger.info(f"[{self.name}] catalog delete done note={note_rel}") self.logger.info(f"[{self.name}] refresh index start date={date_str} daily_dir={daily_dir}") index_payload = await refresh_day_index(self.file_store, date_str, daily_dir) self.logger.info(f"[{self.name}] refresh index done date={date_str}") self.context.response.success = True self.context.response.answer = f"Deleted resource note: {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, "index": index_payload, }, ) async def _handle_upsert( self, file_path: str, date_str: str, note_stem: str, added: bool, ) -> None: self.logger.info( f"[{self.name}] upsert start file_path={file_path} date={date_str} " f"note_stem={note_stem} added={added}", ) daily_dir = self.config_value("daily_dir") fallback_path = f"{daily_dir}/{date_str}/{note_stem}.md" try: note = await self._list_resource_note(date_str, file_path, fallback_path) except RuntimeError as exc: self.context.response.success = False self.context.response.answer = str(exc) self.logger.info(f"[{self.name}] list failed file_path={file_path} answer={str(exc)!r}") return note_path = str(note["path"]) if note else fallback_path note_created = note is None before_note_path = note_path before_note_bytes = self._note_bytes(note_path) self.logger.info(f"[{self.name}] daily note lookup path={note_path} created={note_created}") # Read resource file content abs_path = self.workspace_path / file_path if not abs_path.is_file(): self.context.response.success = False self.context.response.answer = f"Resource file not found: {file_path}" self.logger.warning(f"[{self.name}] resource missing file_path={file_path}") return skip_read = False try: size_bytes = abs_path.stat().st_size except OSError as exc: self.context.response.success = False self.context.response.answer = f"Failed to inspect resource file: {file_path}: {exc}" self.context.response.metadata.update( { "path": file_path, "action": "failed", "error": str(exc), "modified": False, }, ) self.logger.warning(f"[{self.name}] resource stat failed file_path={file_path} error={exc}") skip_read = True if not skip_read: max_file_bytes = self.max_file_bytes() if size_bytes > max_file_bytes: self.context.response.success = True self.context.response.answer = ( f"Skipped oversized resource file: {file_path} ({size_bytes} > {max_file_bytes} bytes)" ) self.context.response.metadata.update( { "path": file_path, "action": "skipped", "reason": "file_too_large", "oversized": True, "size_bytes": size_bytes, "max_file_bytes": max_file_bytes, "modified": False, }, ) self.logger.warning( f"[{self.name}] skip oversized resource file_path={file_path} " f"size_bytes={size_bytes} max_file_bytes={max_file_bytes}", ) skip_read = True if skip_read: return self.logger.info(f"[{self.name}] read resource start file_path={file_path}") async with aiofiles.open(abs_path, encoding="utf-8", errors="replace") as f: file_content = await f.read() self.logger.info(f"[{self.name}] read resource done file_path={file_path} chars={len(file_content)}") template_key = "user_message_create" if note_created else "user_message_update" user_message = self.prompt_format( template_key, workspace_dir=str(self.workspace_path), note_path=note_path, note_stem=note_stem, file_path=file_path, source_resource=self._source_resource_link(file_path), file_content=file_content, date=date_str, ) agent_session_id = _compute_agent_session_id(file_path) self.logger.info( f"[{self.name}] agent start file_path={file_path} note_path={note_path} " f"agent_session_id={agent_session_id}", ) result = await self.agent_wrapper.reply( user_message, system_prompt=self.prompt_format("system_prompt"), job_tools=self.create_tools if note_created else self.update_tools, session_id=agent_session_id, ) self.logger.info(f"[{self.name}] agent done file_path={file_path} has_result={bool(result.get('result'))}") if note_created: try: note = await self._list_resource_note(date_str, file_path, fallback_path) except RuntimeError as exc: self.context.response.success = False self.context.response.answer = str(exc) self.context.response.metadata.update({"path": None, "created": note_created, "modified": False}) self.logger.info(f"[{self.name}] post-create list failed file_path={file_path} answer={str(exc)!r}") return if note is None: self.context.response.success = True self.context.response.answer = agent_reply_result_text(result) self.context.response.metadata.update({"path": None, "created": False, "modified": False}) self.logger.info(f"[{self.name}] done without note file_path={file_path} modified=False") return note_path = str(note["path"]) try: await self._ensure_resource_frontmatter(note_path, file_path) note_path = await self._rename_from_frontmatter_name( note_path, date_str, file_path, note_stem, fallback_path, allow_rename=note_created, ) except RuntimeError as exc: self.context.response.success = False self.context.response.answer = str(exc) self.context.response.metadata.update( { "path": note_path, "created": note_created, "modified": self._note_modified(before_note_path, before_note_bytes, note_path), }, ) self.logger.info(f"[{self.name}] post-agent failed path={note_path} answer={str(exc)!r}") return modified = self._note_modified(before_note_path, before_note_bytes, note_path) self.logger.info(f"[{self.name}] refresh index start date={date_str} daily_dir={daily_dir}") index_payload = await refresh_day_index(self.file_store, date_str, daily_dir) self.logger.info(f"[{self.name}] refresh index done date={date_str}") self.context.response.success = True self.context.response.answer = agent_reply_result_text(result) self.context.response.metadata.update( { "path": note_path, "created": note_created, "modified": modified, "session_id": note_stem, "source_resource": self._source_resource_link(file_path), "agent_session_id": agent_session_id, "action": "added" if added else "modified", "index": index_payload, }, ) self.logger.info(f"[{self.name}] done {note_path} modified={modified}") async def _handle_change(self, file_path: str, raw_change) -> dict: assert self.context is not 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 = {} file_path = self.to_workspace_relative(file_path) if file_path and Path(file_path).is_absolute() else file_path 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} resource_dir = self.config_value("resource_dir") loose_filename = _loose_resource_filename(file_path, resource_dir) if loose_filename: date_str, filename = self._today(), loose_filename self.logger.info(f"[{self.name}] loose resource file_path={file_path} date={date_str}") 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, ) return { "success": self.context.response.success, "path": file_path, "change": change.name, "answer": self.context.response.answer, "metadata": dict(self.context.response.metadata), } async def execute(self): 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)}") results.append( await self._handle_change(item.get("path") or item.get("file_path", ""), item.get("change", "")), ) 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 ) await self._emit_result_hook(changes=changes, results=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