feat(auto_resource): interpret image resources into daily notes (#500)

* feat: add auto_image step for image resource caption notes

* feat: wire image resources into the resource watch loop

* refactor: split auto resource processors behind router

* refactor: align resource processor module names

* refactor: preserve auto resource compatibility

* refactor: clarify auto resource routing structure

* fix: address auto resource review concerns

* test: scope auto resource fixtures

* docs: align auto resource processor wording

* test: cover image resize failures

* fix: harden image resource lifecycle

* fix: preserve resource image detail and linked daily ownership

* style(file-graph): stabilize multiline docstring formatting
This commit is contained in:
WQS 2026-09-07 12:00:38 +08:00 committed by GitHub
parent 0eba6ea831
commit 5c17874f73
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
28 changed files with 3699 additions and 552 deletions

View file

@ -342,7 +342,7 @@ These guides cover the main user workflows and the runtime contracts implemented
| [Services and Deployment](docs/en/services.md) | Use HTTP, SSE, MCP, and Studio while respecting the default security boundary. |
| [Memory as File](docs/en/memory_as_file.md) | Understand workspace layers, frontmatter, wikilinks, chunks, and the file-as-source-of-truth model. |
| [Auto Memory](docs/en/auto_memory.md) | Preserve source conversations and distill reusable daily memory cards. |
| [Auto Resource](docs/en/auto_resource.md) | Import supported text resources and turn them into source-linked daily cards. |
| [Auto Resource](docs/en/auto_resource.md) | Import supported text and image resources as source-linked daily cards. |
| [Auto Dream](docs/en/auto_dream.md) and [Auto Link](docs/en/auto_link.md) | Consolidate daily notes into evolving digest nodes and readable wikilink relationships. |
| [Memory Search](docs/en/memory_search.md) | Use BM25, optional vectors, RRF fusion, line-range recall, and progressive link expansion. |
| [Proactive](docs/en/proactive.md) | Read interest topics safely and integrate them into a host agent's decision flow. |

View file

@ -332,7 +332,7 @@ ReMe 通过 Agent 多轮搜索与读取的方式,评测多会话和超长上
| [服务与部署](docs/zh/services.md) | 使用 HTTP、SSE、MCP 和 Studio并理解默认安全边界。 |
| [Memory as File](docs/zh/memory_as_file.md) | 理解 workspace 分层、frontmatter、wikilink、chunk 和文件事实来源模型。 |
| [Auto Memory](docs/zh/auto_memory.md) | 保留过滤后的对话来源记录,并提炼可复用的 daily 记忆卡片。 |
| [Auto Resource](docs/zh/auto_resource.md) | 导入支持的文本资料,转换为可追溯来源的 daily 卡片。 |
| [Auto Resource](docs/zh/auto_resource.md) | 导入支持的文本与图像资料,转换为可追溯来源的 daily 卡片。 |
| [Auto Dream](docs/zh/auto_dream.md) 与 [Auto Link](docs/zh/auto_link.md) | 将 daily 记忆整理为持续演化的 digest 节点和可读 wikilink 关系。 |
| [记忆检索](docs/zh/memory_search.md) | 使用 BM25、可选向量、RRF 融合、行号范围召回和渐进式链接扩展。 |
| [Proactive](docs/zh/proactive.md) | 安全读取兴趣主题,并将其接入宿主 Agent 的决策流程。 |

View file

@ -36,7 +36,8 @@ In short, it turns "a file was archived" into "the resource is usable."
Auto Resource uses `resource/` as the entry point for source material. Date directories are recommended, and their date
determines which daily memory layer receives the interpreted card. A file directly under `resource/` is also supported
and uses today in the application timezone.
and uses today in the application timezone when it is first processed. On later days, an exact `source_resource` match
keeps updates and deletion tied to that original daily card instead of creating a new card or leaving an orphan.
Example directory:
@ -49,13 +50,39 @@ workspace/
meeting-notes.csv
```
The current Beta version is best suited to text-based resources such as `md`, `txt`, `json`, `jsonl`, `csv`, `yaml`, and
`html`.
Text resources such as `md`, `txt`, `json`, `jsonl`, `csv`, `yaml`, and `html` are the primary fit. Image resources
(`png`, `jpg`, `jpeg`, `webp`, `gif`, `bmp`, `tiff`, `heic`) produce caption cards as described in
[Image Resources](#image-resources).
Internally, one `AutoResourceStep` receives each change batch and sends every item to the first configured processor
whose class-level matcher accepts it. `AutoImageResourceStep` handles image suffixes and `AutoTextResourceStep` is the
final fallback. A new modality can therefore add a registered processor, its prompt, and one `dispatch_steps` entry
without changing the router.
## Image Resources
Image files are interpreted the same way: a vision model writes a caption card that links back to the original image.
The card body starts with an `![[resource/...]]` embed link and the frontmatter carries `kind: image` and `media_type`,
so text search reaches image content through the caption.
The vision model is the `vision` instance of `as_llm` when configured, and otherwise falls back to the `default`
instance — a multimodal default model needs no extra configuration. Images wider or taller than 2048px are downscaled,
and provider-unfriendly formats are re-encoded, in memory for the request only; the original file under
`resource/` is never modified. Before a full decode, image dimensions are checked against a default limit of 40,000,000
pixels; images over the limit and Pillow decompression-bomb warnings fail only that resource. EXIF orientation is
applied to the in-memory request copy before resizing or conversion. Oversized JPEGs first use decoder-level
downsampling, followed by a final thumbnail pass when needed. The VLM request MIME and the card's frontmatter
`media_type` use the format Pillow detects from the image bytes, rather than trusting the filename extension. When an
image changes, its card is rewritten in place; when the image is deleted, the card is removed with it.
Image preprocessing uses Pillow from the `core` extra. HEIC resources additionally require the optional
`image-heif` extra: `pip install "reme-ai[image-heif]"`. Other supported image formats do not load or require the HEIF
plugin.
## Resource Cards
Each resource file produces one daily resource card. The system initially uses the resource file's stem as a temporary
path. After the agent writes the card, the file is renamed according to its frontmatter `name`:
path. After the matching processor writes the card, the file is renamed according to its frontmatter `name`:
```text
resource/2026-06-20/market-report.md
@ -69,9 +96,9 @@ The resource card links to the original file through frontmatter:
source_resource: "[[resource/2026-06-20/market-report.md]]"
```
When a resource changes, Auto Resource finds and updates the corresponding card through `source_resource`. When a
resource is deleted, its daily note is also removed. The older `daily/YYYY-MM-DD/<resource_stem>.md` naming convention
remains supported as a fallback.
When a resource changes, Auto Resource finds and updates the corresponding card through an exact `source_resource`
match. When a resource is deleted, only the explicitly linked daily note is removed. A same-stem note without that
provenance marker is treated as user-owned and left untouched; new resource cards use a collision-free path instead.
## Daily Index
@ -93,7 +120,7 @@ resource, open its corresponding resource card.
The interpreted daily note is optimized for readability; the original resource is retained for trust and verification.
Auto Resource does not move the original file. It remains at its original path under `resource/`. Text resources can
Auto Resource does not move the original file. It remains at its original path under `resource/`. Resources can
therefore enter the daily memory flow while their source files stay in their original location.
## What Happens Next

View file

@ -192,8 +192,8 @@ reme auto_memory \
```
After placing external material under `resource/YYYY-MM-DD/` or directly under `resource/`, the default background task
watches
`md/txt/json/jsonl/csv/yaml/html`. You can also trigger processing manually:
watches text resources (`md/txt/json/jsonl/csv/yaml/html`) and image resources
(`png/jpg/jpeg/webp/gif/bmp/tiff/heic`). You can also trigger processing manually:
```bash
reme auto_resource changes='[{"path":"resource/2026-06-20/report.md","change":"added"}]'

View file

@ -113,7 +113,7 @@ It is like having a recorder who is always present—not one that mechanically t
Not all valuable information comes from conversations. Research materials, project documents, meeting notes, archived web pages, and structured data may all become part of a personal knowledge base.
Auto Resource provides a general entry point for external materials. After a resource enters `resource/`, ReMe preserves the original and organizes its topics, key facts, and actionable information into daily cards with `source_resource` links. It currently supports text-based resources including Markdown, plain text, JSON, JSONL, CSV, YAML, and HTML.
Auto Resource provides a general entry point for external materials. After a resource enters `resource/`, ReMe preserves the original and organizes its topics, key facts, and actionable information into daily cards with `source_resource` links. It supports text resources including Markdown, plain text, JSON, JSONL, CSV, YAML, and HTML, plus image resources that a vision model turns into caption cards.
In other words, Auto Memory builds personal knowledge from conversations, while Auto Resource builds it from non-conversational materials. Both streams flow into the same daily memory layer, where ReMe indexes, consolidates, and retrieves them together.

View file

@ -66,9 +66,9 @@ The corresponding flow is:
- `auto_memory` saves a filtered source conversation record to `session/dialog/<session_id>.jsonl`, then asks the agent to write
important facts to a topic-named `daily/<date>/<generated_name>.md`. The note keeps `session_id` and
`source_conversation` in frontmatter for stable lookup and provenance.
- `resource_watch_loop` watches text-file changes under `resource/` and triggers `auto_resource_step` to write a daily note
with `source_resource`. The agent suggests a content-based filename, which the system sanitizes and de-duplicates; it is
not guaranteed to match the resource filename.
- `resource_watch_loop` watches supported text and image changes under `resource/` and triggers `auto_resource_step` to
write a daily note with `source_resource`. Text resources use the agent, while images use a vision model. The generated
content-based filename is sanitized and de-duplicated; it is not guaranteed to match the resource filename.
- Auto Memory, Auto Resource, and Auto Dream refresh `daily/<date>.md` after writing.
### Day 1 evening: Auto Dream writes to Digest

View file

@ -33,7 +33,8 @@ resource/[YYYY-MM-DD/]<resource_file>
## 原始资料入口
Auto Resource 以 `resource/` 作为原始资料入口。推荐按日期放置,目录日期会决定它进入哪一天的 daily 记忆层;也支持直接放在
`resource/` 根目录,此时使用应用时区中的今天。
`resource/` 根目录,首次处理时使用应用时区中的今天。之后即使跨天更新或删除,也会通过精确匹配
`source_resource` 继续操作原 daily 卡片,不会重复新建卡片或留下孤立链接。
示例目录:
@ -46,11 +47,30 @@ workspace/
meeting-notes.csv
```
当前 Beta 版本更适合处理文本类资源,例如 `md``txt``json``jsonl``csv``yaml``html`
当前 Beta 版本以文本类资源为主,例如 `md``txt``json``jsonl``csv``yaml``html`;图像资源(`png``jpg``jpeg``webp``gif``bmp``tiff``heic`)会生成 caption 卡片,见下文[图像资源](#图像资源)一节。
内部由统一的 `AutoResourceStep` 接收每批变更,并将每一项交给配置中第一个匹配它的 processor。
`AutoImageResourceStep` 声明图像后缀匹配规则,`AutoTextResourceStep` 作为最后的 fallback。后续新增模态时
只需注册新的 processor、提供独立 prompt 并在 `dispatch_steps` 中增加一项,无需修改 router。
## 图像资源
图像文件的解读方式相同:视觉模型写入一张 caption 卡片并链接原图。卡片正文以 `![[resource/...]]` 嵌入链接开头frontmatter 携带 `kind: image``media_type`,文本检索因此可以通过 caption 命中图像内容。
视觉模型优先使用配置中的 `as_llm` `vision` 实例,未配置时回退到 `default` 实例——默认模型具备视觉能力时无需额外配置。宽或高超过 2048px 的图像会降采样,格式不被模型接受的图像会转码;这些处理只发生在请求前的内存副本中,`resource/` 下的原图文件不会被修改。图像变更时卡片原地重写;图像删除时卡片随之删除。
在完整解码前,系统会检查图像尺寸,默认上限为 40,000,000 像素;超限图像或 Pillow
decompression-bomb 警告只会导致当前资源失败。缩放或转码前,会按 EXIF orientation 校正仅用于请求的内存副本。
尺寸过大的 JPEG 会先使用 decoder-level downsampling并在需要时再完成最终缩放。
VLM 请求的 MIME 和卡片 frontmatter 中的 `media_type` 都使用 Pillow 根据实际图像字节识别的格式,
而不是直接信任文件扩展名。
图像预处理使用 `core` extra 中的 Pillow。HEIC 资源还需要可选的 `image-heif` extra
`pip install "reme-ai[image-heif]"`。其他受支持图像格式不会加载或依赖 HEIF 插件。
## 资源卡片
每个资源文件会生成一张 daily 资源卡片。创建时先使用资源文件 stem 作为临时路径Agent 写入后,系统会根据 frontmatter `name`
每个资源文件会生成一张 daily 资源卡片。创建时先使用资源文件 stem 作为临时路径,对应 processor 写入后,系统会根据 frontmatter `name`
重命名文件:
```text
@ -65,8 +85,8 @@ daily/2026-06-20/市场报告要点.md
source_resource: "[[resource/2026-06-20/market-report.md]]"
```
如果资源文件更新Auto Resource 会通过 `source_resource` 找到对应卡片并更新;如果资源文件删除,对应的 daily note 也会被清理。旧版本按
stem 生成的 `daily/YYYY-MM-DD/<resource_stem>.md` 仍作为 fallback 兼容
如果资源文件更新Auto Resource 会通过精确匹配的 `source_resource` 找到对应卡片并更新;如果资源文件删除,也只会清理显式关联的
daily note。缺少该来源标记的同 stem 笔记会被视为用户笔记并保留,新资源卡片则会使用无冲突路径
## 当天索引
@ -86,7 +106,7 @@ daily/
解读后的 daily note 负责“好读”,原始资源负责“可信”。
Auto Resource 不会把原始文件挪走:它仍然留在 `resource/` 下的原路径。这样,文本资料会进入 daily 记忆流,原始文件也始终保留在它来时的位置。
Auto Resource 不会把原始文件挪走:它仍然留在 `resource/` 下的原路径。这样,文本与图像资料会进入 daily 记忆流,原始文件也始终保留在它来时的位置。
## 后续流向

View file

@ -185,7 +185,8 @@ reme auto_memory \
memory_hint="记录用户偏好"
```
外部资料放入 `resource/YYYY-MM-DD/` 或直接放在 `resource/` 下后,默认后台会监听 `md/txt/json/jsonl/csv/yaml/html`
外部资料放入 `resource/YYYY-MM-DD/` 或直接放在 `resource/` 下后,默认后台会监听文本资源
(`md/txt/json/jsonl/csv/yaml/html`) 和图像资源 (`png/jpg/jpeg/webp/gif/bmp/tiff/heic`)。
也可以手动触发:
```bash

View file

@ -119,7 +119,7 @@ daily/2026-08-07.md 当天索引,负责总览
并不是所有有价值的信息都来自对话。研究资料、项目文档、会议纪要、网页存档和结构化数据,同样可能成为个人知识库的一部分。
Auto Resource 提供了一条更通用的外部资料入口。资料进入 `resource/`ReMe 保留原文,再把主题、关键事实和可行动信息整理为带有
`source_resource` 链接的 daily 卡片。当前可以处理 Markdown、纯文本、JSON、JSONL、CSV、YAML 和 HTML 等文本类资料。
`source_resource` 链接的 daily 卡片。它支持 Markdown、纯文本、JSON、JSONL、CSV、YAML 和 HTML 等文本类资料,也支持由视觉模型生成 caption 卡片的图像资源
这意味着Auto Memory 负责从对话建立个人知识Auto Resource 负责从非对话资料建立个人知识。两条输入最终进入同一个 daily
记忆层,再由 ReMe 统一索引、整合和检索。

View file

@ -61,8 +61,8 @@ daily/
- `auto_memory` 保存对话来源消息到 `session/dialog/<session_id>.jsonl`,再让 Agent 把重要事实写入按主题命名的
`daily/<date>/<generated_name>.md`;卡片 frontmatter 保留 `session_id``source_conversation` 用于稳定定位和追溯。
- `resource_watch_loop` 监听 `resource/` 文本文件变化,并触发 `auto_resource_step` 写带 `source_resource` 的 daily note文件名由
Agent 根据内容建议,再由系统清洗并处理冲突,不保证与资源同名。
- `resource_watch_loop` 监听 `resource/` 下支持的文本与图像变化,并触发 `auto_resource_step` 写带
`source_resource` 的 daily note文本资源由 Agent 处理,图像由视觉模型处理。根据内容生成的文件名会被系统清洗并处理冲突,不保证与资源同名。
- Auto Memory、Auto Resource 和 Auto Dream 都会在写入后刷新 `daily/<date>.md` 当天索引页。
### Day 1 晚上Auto Dream 进入 Digest

View file

@ -49,6 +49,7 @@ web = [
]
core = [
"reme-ai[as]",
"pillow>=10.0.0",
"claude-agent-sdk>=0.2.126",
"dingtalk-stream>=0.24.3",
"openai-codex>=0.144.4",
@ -62,6 +63,10 @@ core = [
"polars>=1.43.0",
"reme_studio",
]
image-heif = [
"pillow>=10.0.0",
"pillow-heif>=0.13.0",
]
dev = [
"packaging>=24.2",
"pre-commit>=4.6.1",
@ -71,6 +76,7 @@ dev = [
full = [
"reme-ai[core]",
"reme-ai[dev]",
"reme-ai[image-heif]",
]
[project.urls]

View file

@ -217,7 +217,8 @@ class Neo4jFileGraph(BaseFileGraph):
async def delete_nodes(self, paths: list[str]) -> None:
"""Demote real → virtual to preserve inbound visibility; fully
remove the (now-virtual) node only if no edge points at it."""
remove the (now-virtual) node only if no edge points at it.
"""
if not paths:
return
async with self._session() as session:

View file

@ -22,7 +22,7 @@ jobs:
resource_watch_loop:
backend: background
watch_dirs: [resource_dir]
watch_suffixes: [md, txt, json, jsonl, csv, yaml, html]
watch_suffixes: [md, txt, json, jsonl, csv, yaml, html, png, jpg, jpeg, webp, gif, bmp, tiff, heic]
steps:
- backend: init_changes_step
monitor_type: file_catalog
@ -31,11 +31,17 @@ jobs:
- backend: update_catalog_step
file_catalog: resource
- backend: auto_resource_step
dispatch_steps:
- auto_image_resource_step
- auto_text_resource_step
- backend: watch_changes_step
dispatch_steps:
- backend: update_catalog_step
file_catalog: resource
- backend: auto_resource_step
dispatch_steps:
- auto_image_resource_step
- auto_text_resource_step
digest_watch_loop:
backend: background
@ -187,6 +193,9 @@ jobs:
- changes
steps:
- backend: auto_resource_step
dispatch_steps:
- auto_image_resource_step
- auto_text_resource_step
proactive:
backend: base
@ -755,6 +764,17 @@ components:
max_tokens: 65536
thinking_enable: false
# Optional dedicated vision model for image resources (auto_image_resource_step).
# Falls back to the "default" instance above when absent; uncomment to
# decouple the vision model from the main LLM.
# vision:
# backend: ${VLM_BACKEND:-openai}
# model: ${VLM_MODEL_NAME:-}
# stream: false
# credential:
# api_key: ${VLM_API_KEY:-}
# base_url: ${VLM_BASE_URL:-}
agent_wrapper:
default:
backend: agentscope

View file

@ -1,17 +1,21 @@
"""Evolve steps."""
from ._evolve import now
from .auto_image_resource import AutoImageResourceStep
from .auto_memory import AutoMemoryStep
from .auto_memory_cc import AutoMemoryCCStep
from .auto_resource import AutoResourceStep
from .auto_text_resource import AutoTextResourceStep
from .compressor import CompressorStep
from .dream import DreamExtractStep, DreamFinishStep, DreamIntegrateStep, DreamTopicsStep, ProactiveStep
__all__ = [
"now",
"AutoImageResourceStep",
"AutoMemoryStep",
"AutoMemoryCCStep",
"AutoResourceStep",
"AutoTextResourceStep",
"CompressorStep",
"DreamExtractStep",
"DreamFinishStep",

View file

@ -0,0 +1,521 @@
"""Image resource processor for the unified auto-resource router."""
import base64
import io
import json
import re
import warnings
from pathlib import Path, PurePosixPath
import aiofiles
from agentscope.message import Base64Source, DataBlock, TextBlock, UserMsg
from agentscope.model import ChatModelBase
from pydantic import BaseModel, Field
from ..file_io._path import IMAGE_SUFFIXES
from .base_auto_resource import _SOURCE_RESOURCE_KEY, _sanitize_note_name, BaseAutoResourceStep
from ...components import R
from ...enumeration import ComponentEnum
DEFAULT_MAX_IMAGE_INPUT_BYTES = 50 * 1024 * 1024
DEFAULT_MAX_IMAGE_PIXELS = 40_000_000
MAX_IMAGE_REQUEST_DIMENSION = 2048
_JPEG_QUALITY = 85
# Decoded formats outside this set are re-encoded to provider-friendly
# PNG/JPEG for VLM requests; the stored resource file is never modified.
_PASSTHROUGH_IMAGE_MIMES = frozenset({"image/png", "image/jpeg", "image/webp", "image/gif"})
_HEIF_BRANDS = frozenset(
{b"heic", b"heif", b"heix", b"heim", b"heis", b"hevc", b"hevx", b"hevm", b"hevs", b"mif1", b"msf1"},
)
_MAX_FTYP_SCAN_BYTES = 4096
_JSON_FENCE_RE = re.compile(r"^\s*```(?:json)?\s*(.*?)\s*```\s*$", re.DOTALL)
class _CaptionOutput(BaseModel):
"""Structured caption contract enforced on the vision model."""
name: str = Field(
description="short kebab-case topic stem based on visible content; filename is only a weak naming hint",
)
description: str = Field(description="one-sentence summary of visible image content that stands on its own")
caption: str = Field(
description="complete description / verbatim transcription of meaningful content visible in the image",
)
def _load_pillow():
"""Load the core image dependency only when image processing runs."""
try:
from PIL import Image, ImageOps # pylint: disable=import-outside-toplevel
except ImportError as exc:
raise RuntimeError("Image captioning requires Pillow; install reme-ai[core]") from exc
return Image, ImageOps
def _looks_like_heif(data: bytes) -> bool:
"""Return whether an ISO-BMFF header declares a HEIC/HEIF brand."""
if len(data) < 12 or data[4:8] != b"ftyp":
return False
box_size = int.from_bytes(data[:4], "big")
if box_size < 12:
return False
end = min(box_size, len(data), _MAX_FTYP_SCAN_BYTES)
if data[8:12] in _HEIF_BRANDS:
return True
return any(data[index : index + 4] in _HEIF_BRANDS for index in range(16, end - 3, 4))
def _register_heif_opener() -> None:
"""Load and register HEIC support only for bytes that declare HEIF."""
try:
from pillow_heif import register_heif_opener # pylint: disable=import-outside-toplevel
except ImportError as exc:
raise RuntimeError("HEIC image captioning requires pillow-heif; install reme-ai[image-heif]") from exc
try:
register_heif_opener()
except Exception as exc: # pylint: disable=broad-except
raise RuntimeError(f"Failed to initialize HEIC image support: {exc}") from exc
def _normalize_image_bytes(
data: bytes,
suffix: str,
*,
max_image_pixels: int = DEFAULT_MAX_IMAGE_PIXELS,
) -> tuple[bytes | None, str, str]:
"""Validate and optionally normalize image bytes for a VLM request.
The returned tuple is ``(normalized_bytes, request_mime, source_mime)``.
``normalized_bytes`` is ``None`` only when the original bytes can be sent
unchanged. MIME values come from the decoded image rather than its suffix.
Missing dependencies, unsafe pixel counts, and decode/convert failures are
explicit. The stored resource file is never modified.
"""
try:
pixel_limit = int(max_image_pixels)
except (TypeError, ValueError) as exc:
raise ValueError(f"max_image_pixels must be a positive integer: {max_image_pixels!r}") from exc
if pixel_limit <= 0:
raise ValueError(f"max_image_pixels must be a positive integer: {max_image_pixels!r}")
image_module, image_ops = _load_pillow()
if _looks_like_heif(data):
_register_heif_opener()
try:
with warnings.catch_warnings():
warnings.simplefilter("error", image_module.DecompressionBombWarning)
image = image_module.open(io.BytesIO(data))
except (image_module.DecompressionBombWarning, image_module.DecompressionBombError) as exc:
raise RuntimeError(
f"Image rejected by Pillow decompression-bomb protection ({suffix or 'unknown suffix'})",
) from exc
except Exception as exc: # pylint: disable=broad-except
raise RuntimeError(f"Failed to decode image ({suffix or 'unknown suffix'}): {exc}") from exc
with image:
width, height = image.size
pixel_count = width * height
if pixel_count > pixel_limit:
raise RuntimeError(
f"Image exceeds max_image_pixels before decode: " f"{width}x{height}={pixel_count} > {pixel_limit}",
)
source_mime = str(image.get_format_mimetype() or "").strip().lower()
if not source_mime.startswith("image/"):
raise RuntimeError(
f"Cannot determine decoded image MIME type ({suffix or 'unknown suffix'}, format={image.format!r})",
)
needs_resize = width > MAX_IMAGE_REQUEST_DIMENSION or height > MAX_IMAGE_REQUEST_DIMENSION
if source_mime == "image/jpeg" and needs_resize:
max_dimension = max(width, height)
decoder_size = (
max(1, (width * MAX_IMAGE_REQUEST_DIMENSION + max_dimension - 1) // max_dimension),
max(1, (height * MAX_IMAGE_REQUEST_DIMENSION + max_dimension - 1) // max_dimension),
)
try:
# JPEG supports power-of-two decoder scaling. ``draft`` picks
# the smallest decoded frame that still covers decoder_size,
# reducing peak memory before the final LANCZOS thumbnail.
image.draft(None, decoder_size)
except Exception as exc: # pylint: disable=broad-except
raise RuntimeError(f"Failed to prepare JPEG decoder downsampling ({width}x{height}): {exc}") from exc
try:
with warnings.catch_warnings():
warnings.simplefilter("error", image_module.DecompressionBombWarning)
image.load()
orientation = int(image.getexif().get(274, 1) or 1)
image_ops.exif_transpose(image, in_place=True)
except (image_module.DecompressionBombWarning, image_module.DecompressionBombError) as exc:
raise RuntimeError(
f"Image rejected by Pillow decompression-bomb protection ({suffix or 'unknown suffix'})",
) from exc
except Exception as exc: # pylint: disable=broad-except
raise RuntimeError(f"Failed to decode image ({suffix or 'unknown suffix'}): {exc}") from exc
needs_convert = source_mime not in _PASSTHROUGH_IMAGE_MIMES
needs_orientation = orientation in range(2, 9)
if not needs_resize and not needs_convert and not needs_orientation:
return None, source_mime, source_mime
resize_frame = None
try:
frame = image
if needs_resize:
# Pillow forces NEAREST for palette and bilevel images, even
# when LANCZOS is requested. Expand these modes within the
# checked pixel budget so resizing retains fine strokes and
# palette transparency. Other modes resize before conversion.
if image.mode in ("P", "1"):
resize_frame = image.convert("RGBA" if image.mode == "P" else "L")
frame = resize_frame
# Pillow 10 cannot apply LANCZOS directly to 16-bit integer
# modes; NEAREST keeps that path bounded without a full-size
# RGB conversion first.
resize_filter = image_module.Resampling.LANCZOS
if frame.mode.startswith("I;16"):
resize_filter = image_module.Resampling.NEAREST
frame.thumbnail((MAX_IMAGE_REQUEST_DIMENSION, MAX_IMAGE_REQUEST_DIMENSION), resize_filter)
has_alpha = frame.mode in ("RGBA", "LA", "P")
frame = frame.convert("RGBA" if has_alpha else "RGB")
try:
buffer = io.BytesIO()
if frame.mode == "RGBA":
frame.save(buffer, format="PNG")
return buffer.getvalue(), "image/png", source_mime
frame.save(buffer, format="JPEG", quality=_JPEG_QUALITY)
return buffer.getvalue(), "image/jpeg", source_mime
finally:
frame.close()
except Exception as exc: # pylint: disable=broad-except
raise RuntimeError(f"Failed to convert/resize image ({suffix or 'unknown suffix'}): {exc}") from exc
finally:
if resize_frame is not None:
resize_frame.close()
def _build_image_request_payload(
data: bytes,
suffix: str,
*,
max_image_pixels: int = DEFAULT_MAX_IMAGE_PIXELS,
) -> dict:
"""Return ``{"data_b64", "mime", "source_mime", "converted"}`` for a VLM request.
``mime`` is the format actually sent (after in-memory downscale/re-encode);
``source_mime`` is the decoded format of the stored resource file and is
what notes record. Both are based on actual bytes, not the filename suffix.
"""
normalized_bytes, mime, source_mime = _normalize_image_bytes(
data,
suffix,
max_image_pixels=max_image_pixels,
)
request_bytes = data if normalized_bytes is None else normalized_bytes
return {
"data_b64": base64.b64encode(request_bytes).decode("ascii"),
"mime": mime,
"source_mime": source_mime,
"converted": normalized_bytes is not None,
}
async def _response_text(result) -> str:
"""Extract text blocks from a streaming or non-streaming ChatResponse."""
if hasattr(type(result), "__aiter__"):
last = None
async for chunk in result:
last = chunk
result = last
if result is None:
return ""
parts: list[str] = []
for block in result.content or []:
if isinstance(block, dict):
if block.get("type") == "text":
parts.append(str(block.get("text") or ""))
elif getattr(block, "type", None) == "text":
parts.append(str(getattr(block, "text", "") or ""))
return "".join(parts).strip()
def _normalize_caption_fields(parsed: dict) -> dict:
"""Normalize parsed caption fields, cross-filling a missing ``caption``
from a present ``description`` so raw JSON never reaches the note body."""
caption = str(parsed.get("caption") or "").strip()
description = str(parsed.get("description") or "").strip()
if not caption and description:
caption = description
return {
"name": str(parsed.get("name") or "").strip(),
"description": description,
"caption": caption,
}
def _parse_caption_json(text: str) -> dict:
"""Parse a plain-call caption response leniently.
Used as the fallback when the schema-forced structured call fails: fenced
JSON and embedded ``{...}`` slices are tried before degrading the whole
response text to the caption.
"""
cleaned = text.strip()
fence = _JSON_FENCE_RE.match(cleaned)
if fence:
cleaned = fence.group(1)
parsed_json = False
for candidate in (cleaned, cleaned[cleaned.find("{") : cleaned.rfind("}") + 1]):
if not candidate:
continue
try:
parsed = json.loads(candidate)
except (json.JSONDecodeError, ValueError):
continue
parsed_json = True
if isinstance(parsed, dict):
normalized = _normalize_caption_fields(parsed)
if normalized["caption"] or normalized["description"]:
return normalized
if parsed_json:
return {"name": "", "description": "", "caption": ""}
return {"name": "", "description": "", "caption": cleaned.strip()}
@R.register("auto_image_resource_step")
class AutoImageResourceStep(BaseAutoResourceStep):
"""Interpret image resource files into daily notes via a direct VLM call.
Unlike text resources (agent + file tools), the image interpretation is a
single vision-model call. Images larger than the request budget or in
provider-unfriendly formats are downscaled/re-encoded in memory for the
request only; files under ``resource/`` are never modified. Note lookup,
renaming, deletion linkage, and day-index refresh reuse the shared
BaseAutoResourceStep lifecycle; only the interpretation differs.
"""
resource_suffixes = IMAGE_SUFFIXES
router_inherit_keys = BaseAutoResourceStep.router_inherit_keys | frozenset(
{"as_llm", "max_image_bytes", "max_image_pixels"},
)
def _max_image_bytes(self) -> int:
"""Return the image read limit from Step or Job context."""
value = self.kwargs.get("max_image_bytes")
if value is None and self.context is not None:
value = self.context.get("max_image_bytes")
return int(value) if value is not None else DEFAULT_MAX_IMAGE_INPUT_BYTES
def _max_image_pixels(self) -> int:
"""Return the deployment-controlled pre-decode pixel limit."""
value = self.kwargs.get("max_image_pixels", DEFAULT_MAX_IMAGE_PIXELS)
try:
limit = int(value)
except (TypeError, ValueError) as exc:
raise ValueError(f"max_image_pixels must be a positive integer: {value!r}") from exc
if limit <= 0:
raise ValueError(f"max_image_pixels must be a positive integer: {value!r}")
return limit
def _vision_model(self) -> ChatModelBase | None:
"""Resolve explicit ``as_llm`` through Ref, otherwise prefer vision/default."""
context_model = self.context.get("as_llm") if self.context is not None else None
if "as_llm" in self.kwargs or isinstance(context_model, ChatModelBase):
return self.as_llm
if self.app_context is None:
return None
models = self.app_context.components.get(ComponentEnum.AS_LLM, {})
for name in ("vision", "default"):
if name in models:
self.kwargs["as_llm"] = name
return self.as_llm
return None
async def _caption_with_retry(self, model: ChatModelBase, user_message: UserMsg) -> dict:
"""Return the caption fields from the vision model.
Primary path is the schema-forced structured output (the SDK enforces
the ``name``/``description``/``caption`` contract and retries transport
errors). When that fails or yields no usable field, retry once with a
plain call parsed leniently.
"""
try:
structured = await model.generate_structured_output(
messages=[user_message],
structured_model=_CaptionOutput,
)
content = structured.content if isinstance(structured.content, dict) else {}
normalized = _normalize_caption_fields(dict(content))
if normalized["caption"] or normalized["description"]:
self.logger.info(f"[{self.name}] structured caption ok name={normalized['name']}")
return normalized
self.logger.warning(f"[{self.name}] structured caption empty; retrying with a plain call")
except Exception as exc: # pylint: disable=broad-except
self.logger.warning(f"[{self.name}] structured caption failed ({exc}); retrying with a plain call")
result = await model([user_message])
parsed = _parse_caption_json(await _response_text(result))
if not parsed["caption"] and not parsed["description"]:
raise RuntimeError("Vision model returned no usable caption")
return parsed
async def _read_image(self, file_path: str, source_path: Path) -> dict | None:
"""Read the image file and build the VLM request payload.
Returns ``None`` when the change must be skipped (stat failure or
oversized file); the skip outcome is already recorded on the response.
"""
max_image_bytes = self._max_image_bytes()
try:
size_bytes = source_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}")
return None
if size_bytes > max_image_bytes:
self._record_oversized_image(file_path, size_bytes, max_image_bytes)
return None
self.logger.info(f"[{self.name}] read image start file_path={file_path}")
async with aiofiles.open(source_path, "rb") as f:
data = await f.read(max_image_bytes + 1)
if len(data) > max_image_bytes:
self._record_oversized_image(file_path, len(data), max_image_bytes)
return None
payload = _build_image_request_payload(
data,
Path(file_path).suffix.lower(),
max_image_pixels=self._max_image_pixels(),
)
self.logger.info(
f"[{self.name}] read image done file_path={file_path} size_bytes={size_bytes} "
f"mime={payload['mime']} source_mime={payload['source_mime']} converted={payload['converted']}",
)
return payload
def _record_oversized_image(self, file_path: str, size_bytes: int, max_image_bytes: int) -> None:
"""Record a stable skip response for an image over the compressed-byte limit."""
assert self.context is not None
self.context.response.success = True
self.context.response.answer = (
f"Skipped oversized image resource file: {file_path} ({size_bytes} > {max_image_bytes} bytes)"
)
self.context.response.metadata.update(
{
"path": file_path,
"action": "skipped",
"reason": "file_too_large",
"oversized": True,
"size_bytes": size_bytes,
"max_image_bytes": max_image_bytes,
"modified": False,
},
)
self.logger.warning(
f"[{self.name}] skip oversized image resource file_path={file_path} "
f"size_bytes={size_bytes} max_image_bytes={max_image_bytes}",
)
async def _handle_upsert(
self,
file_path: str,
date_str: str,
note_stem: str,
added: bool,
source_path: Path,
) -> None:
"""Caption the image and write/refresh its note (image counterpart of the text upsert)."""
note_state = await self._prepare_resource_note(date_str, file_path, note_stem)
note_path = note_state.path
self.logger.info(
f"[{self.name}] upsert start file_path={file_path} date={date_str} " f"note_stem={note_stem} added={added}",
)
model = self._vision_model()
if model is None:
self.context.response.success = True
self.context.response.answer = f"Skipped image resource without a vision model: {file_path}"
self.context.response.metadata.update(
{
"path": file_path,
"action": "skipped",
"reason": "vision_model_not_configured",
"modified": False,
},
)
self.logger.warning(f"[{self.name}] no vision model configured file_path={file_path}")
return
payload = await self._read_image(file_path, source_path)
if payload is None:
return
user_message = UserMsg(
name="user",
content=[
TextBlock(
text=self.prompt_format(
"user_message",
file_path=file_path,
filename=PurePosixPath(file_path).name,
stem=note_stem,
date=date_str,
),
),
DataBlock(
source=Base64Source(data=payload["data_b64"], media_type=payload["mime"]),
name="image",
),
],
)
parsed = await self._caption_with_retry(model, user_message)
name = _sanitize_note_name(str(parsed.get("name") or ""), note_stem)
caption = str(parsed.get("caption") or "").strip()
description = str(parsed.get("description") or "").strip() or caption[:120]
body = f"![[{file_path}]]\n\n## Caption\n\n{caption}\n"
# The write job's ``name`` parameter is the note name; calling the job
# directly (instead of run_job) keeps it clear of run_job's
# positional-only job-selector argument.
write_job = self.get_job("write")
if write_job is None:
raise RuntimeError("Job write not found")
write_response = await write_job(
path=note_path,
name=name,
description=description,
content=body,
metadata={
_SOURCE_RESOURCE_KEY: self._source_resource_link(file_path),
"kind": "image",
"media_type": payload["source_mime"],
},
)
if not write_response.success:
raise RuntimeError(f"write failed: {write_response.answer}")
note_path = await self._finalize_resource_note(
note_state,
date_str,
file_path,
note_stem,
added,
)
if note_path is None:
raise RuntimeError(f"Image caption note was not written: {file_path}")
self.context.response.success = True
self.context.response.answer = f"Captioned image resource {file_path} -> {note_path}"
self.context.response.metadata.update(
{
"media_type": payload["source_mime"],
},
)
self.logger.info(f"[{self.name}] done {note_path} modified={self.context.response.metadata['modified']}")

View file

@ -0,0 +1,65 @@
# AutoImageResourceStep prompts.
user_message: |
Describe the attached image from the user's resource library for a memory knowledge base.
Resource image path: {file_path}
Filename: {filename}
Filename stem: {stem}
Date: {date}
The filename and stem are weak hints for naming or disambiguation only. Do not
treat words in them as visible facts. If a filename conflicts with the image,
trust the visible image content. The description and caption must be grounded
in visible content, not inferred from the filename.
## What to Record
- **Visible facts**: people, objects, places, actions, and relationships that
may matter in future conversations.
- **Text in the image**: transcribe meaningful visible text verbatim (slides,
whiteboards, screenshots, signs, labels); keep the original language; do not
translate or correct it.
- **Numbers and dates**: quantities, measurements, versions, prices, timestamps.
- **Image type and purpose**: photo / screenshot / diagram / whiteboard /
document scan, and what it appears to be for.
## What to Ignore
- Pure visual style, composition, lighting, or aesthetic qualities with no
informational value.
- Speculation about anything not visible in the image.
## Completeness
The caption must let a person who cannot see the image understand its content.
For text-heavy images, verbatim transcription takes priority over summary; use
lists or line breaks to mirror the layout when helpful.
## Output
Return a JSON object with the fields `name` (short kebab-case topic stem for
the note filename; never include dates), `description` (one-sentence summary
that conveys the key information on its own), and `caption` (complete
description / transcription). Return only the JSON object.
user_message_zh: |
为记忆知识库描述用户资源库中的这张图像。
资源图像路径:{file_path}
文件名:{filename}
文件名 stem:{stem}
日期:{date}
文件名和 stem 只能作为命名或消歧时的弱提示,不能当作图中可见事实。如果文件名与图像内容冲突,以图像中的可见内容为准。
description 和 caption 必须基于图像中的可见内容,不得从文件名推断事实。
## 记录什么
- **可见事实**:人物、物体、地点、动作及相互关系——未来对话中可能重要的信息。
- **图中的文字**:逐字转录有意义的可见文字(幻灯片、白板、截图、招牌、标签);保留原语言,不翻译、不纠错。
- **数字与日期**:数量、度量、版本号、价格、时间戳。
- **图像类型与用途**:照片/截图/图表/白板/文档扫描,以及它看起来是做什么用的。
## 忽略什么
- 纯视觉风格、构图、光照等无信息量的美学属性。
- 对图中不存在内容的猜测。
## 完整性
caption 必须让看不到图的人理解其内容。文本密集的图,逐字转录优先于概括;可借用列表/换行还原版式。
## 输出
返回只含以下字段的 JSON 对象:`name`(笔记文件名的简短 kebab-case 主题词;不要含任何日期)、`description`(一句话总结,单独读即可传达关键信息)、`caption`(完整描述/转录)。只返回 JSON 对象本身。

View file

@ -1,246 +1,116 @@
"""auto_resource — interpret resource files into source-linked daily notes via an agent."""
"""Unified processor router for automatic resource interpretation."""
import hashlib
import copy
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
from ...enumeration import ComponentEnum
from ..base_step import BaseStep
from .base_auto_resource import BaseAutoResourceStep, _results_answer
_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
_ProcessorSpec = str | dict
_IndexedChange = tuple[int, dict]
_ProcessorRoute = tuple[dict, type[BaseAutoResourceStep], list[_IndexedChange]]
@R.register("auto_resource_step")
class AutoResourceStep(BaseStep):
"""Interpret resource files into daily notes via an Agent."""
"""Route each resource change to one configured processor and aggregate once."""
def __init__(self, **kwargs):
router_options = dict(kwargs)
super().__init__(**kwargs)
self.create_tools: list[str] = ["write"]
self.update_tools: list[str] = ["read", "edit", "frontmatter_update", "write"]
self._router_options = router_options
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 _processor_spec(self, spec: _ProcessorSpec, step_cls: type[BaseAutoResourceStep]) -> dict:
"""Merge router options declared by the processor into its explicit Step spec."""
params = {"backend": spec} if isinstance(spec, str) else dict(spec)
inherited = {
key: self._router_options[key]
for key in step_cls.router_inherit_keys
if key in self._router_options and self._router_options[key] is not None
}
return {**inherited, **params}
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 _processor_routes(self) -> list[_ProcessorRoute]:
"""Resolve configured processors and allocate their per-invocation batches."""
raw_specs = self.dispatch_step_specs
if not raw_specs:
registry = self.app_context.registry if self.app_context is not None else R
raw_specs = [
backend
for backend, step_cls in registry.get_all(ComponentEnum.STEP).items()
if isinstance(step_cls, type)
and issubclass(step_cls, BaseAutoResourceStep)
and step_cls.resource_fallback
]
if len(raw_specs) != 1:
candidates = ", ".join(sorted(raw_specs)) or "none"
raise RuntimeError(
"AutoResourceStep without dispatch_steps requires exactly one registered "
f"fallback resource processor; found: {candidates}",
)
self.logger.warning(
f"[{self.name}] dispatch_steps omitted; using registered fallback processor={raw_specs[0]}",
)
def _daily_note_path(self, day: str, name: str) -> str:
return f"{self.config_value('daily_dir')}/{day}/{name}.md"
routes: list[_ProcessorRoute] = []
fallback_indexes: list[int] = []
for index, raw_spec in enumerate(raw_specs):
step_cls, _ = self._resolve_dispatch_step(raw_spec)
if not isinstance(step_cls, type) or not issubclass(step_cls, BaseAutoResourceStep):
backend = raw_spec if isinstance(raw_spec, str) else raw_spec.get("backend", "")
raise TypeError(f"Resource processor '{backend}' must inherit BaseAutoResourceStep")
spec = self._processor_spec(raw_spec, step_cls)
if step_cls.resource_fallback:
fallback_indexes.append(index)
routes.append((spec, step_cls, []))
if len(fallback_indexes) > 1:
raise ValueError("AutoResourceStep accepts at most one fallback processor")
if fallback_indexes and fallback_indexes[0] != len(routes) - 1:
raise ValueError("AutoResourceStep fallback processor must be last in dispatch_steps")
return routes
async def _dispatch_processor(
self,
spec: dict,
indexed_changes: list[_IndexedChange],
result_slots: list[dict | None],
) -> None:
"""Dispatch one routed sub-batch and snapshot its shared Response immediately."""
if not indexed_changes:
return
changes = [item for _, item in indexed_changes]
responses = await self.dispatch_steps([spec], changes=changes)
processor_response = responses[-1]
processor_results = copy.deepcopy(processor_response.metadata.get("results") or [])
if len(processor_results) != len(indexed_changes):
raise RuntimeError(
f"Resource processor returned {len(processor_results)} result(s) "
f"for {len(indexed_changes)} change(s)",
)
for (index, _), result in zip(indexed_changes, processor_results):
result_slots[index] = result
@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
def _unsupported_result(item: dict, file_path: str) -> dict:
"""Return a stable failure result when no configured processor accepts a resource."""
answer = f"No configured resource processor accepts: {file_path}"
return {
"success": False,
"path": file_path,
"change": str(item.get("change", "")),
"answer": answer,
"metadata": {
"path": file_path,
"action": "failed",
"reason": "unsupported_resource",
"modified": False,
},
}
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.
"""
"""Notify embedding hosts once with the final aggregate response."""
if self.app_context is None or self.context is None:
return
metadata = getattr(self.app_context, "metadata", None)
@ -267,265 +137,6 @@ class AutoResourceStep(BaseStep):
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")
@ -535,25 +146,46 @@ class AutoResourceStep(BaseStep):
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):
routes = self._processor_routes()
fallback = next((route for route in routes if route[1].resource_fallback), None)
specific_routes = [route for route in routes if not route[1].resource_fallback]
result_slots: list[dict | None] = [None] * len(changes)
for index, item in enumerate(changes):
if not isinstance(item, dict):
self.logger.warning(f"[{self.name}] skip invalid change item index={index} type={type(item).__name__}")
self.logger.warning(
f"[{self.name}] skip invalid change item index={index + 1} 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", "")),
file_path = item.get("path") or item.get("file_path", "")
target = next(
(route for route in specific_routes if route[1].matches_change(item)),
fallback,
)
if target is None:
result_slots[index] = self._unsupported_result(item, file_path)
continue
target[2].append((index, item))
route_counts = ", ".join(f"{spec['backend']}={len(batch)}" for spec, _, batch in routes)
self.logger.info(f"[{self.name}] route changes={len(changes)} processors=({route_counts})")
try:
for spec, _, indexed_changes in routes:
await self._dispatch_processor(spec, indexed_changes, result_slots)
finally:
# dispatch_steps merges the sub-batch into the shared context.
# Downstream steps and the result hook must see the original batch.
self.context["changes"] = changes
results = [item for item in result_slots if item is not None]
success_count = sum(1 for item in results if item.get("success"))
self.context.response.success = success_count == len(changes)
self.context.response.success = len(results) == len(changes) and 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.context.response.metadata = {
"processed": len(results),
"results": results,
"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)} "

View file

@ -0,0 +1,149 @@
"""Text resource processor for the unified auto-resource router."""
import uuid
from pathlib import Path
import aiofiles
from ...components import R
from ._evolve import agent_reply_result_text
from .base_auto_resource import BaseAutoResourceStep
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))
@R.register("auto_text_resource_step")
class AutoTextResourceStep(BaseAutoResourceStep):
"""Interpret text resource files into daily notes via an Agent."""
# Preserve the pre-router AutoResourceStep behavior for direct calls and
# custom watcher suffixes; the default watcher still limits normal inputs.
resource_fallback = True
router_inherit_keys = BaseAutoResourceStep.router_inherit_keys | frozenset(
{"agent_wrapper", "max_file_bytes", "prompt_dict"},
)
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.create_tools: list[str] = ["write"]
self.update_tools: list[str] = ["read", "edit", "frontmatter_update", "write"]
async def _handle_upsert(
self,
file_path: str,
date_str: str,
note_stem: str,
added: bool,
source_path: Path,
) -> None:
self.logger.info(
f"[{self.name}] upsert start file_path={file_path} date={date_str} " f"note_stem={note_stem} added={added}",
)
note_state = await self._prepare_resource_note(date_str, file_path, note_stem)
note_path = note_state.path
note_created = note_state.created
self.logger.info(f"[{self.name}] daily note lookup path={note_path} created={note_created}")
# Read resource file content
if not source_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 = source_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(source_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'))}")
note_path = await self._finalize_resource_note(
note_state,
date_str,
file_path,
note_stem,
added,
)
if note_path is None:
self.context.response.success = True
self.context.response.answer = agent_reply_result_text(result)
self.logger.info(f"[{self.name}] done without note file_path={file_path} modified=False")
return
self.context.response.success = True
self.context.response.answer = agent_reply_result_text(result)
self.context.response.metadata.update(
{
"agent_session_id": agent_session_id,
},
)
self.logger.info(f"[{self.name}] done {note_path} modified={self.context.response.metadata['modified']}")

View file

@ -1,3 +1,4 @@
# AutoTextResourceStep prompts.
system_prompt: |
You are an automatic resource interpretation system. Your job is to read a resource file and record a structured summary into a daily note at the specified path. Think about what information in this file would be most useful for future retrieval and understanding.

View file

@ -0,0 +1,626 @@
"""Shared lifecycle and helpers for automatic resource processors."""
import hashlib
import re
from abc import abstractmethod
from collections.abc import Mapping
from dataclasses import dataclass
from pathlib import Path, PurePosixPath
from typing import Any
import frontmatter
from watchfiles import Change
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 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]+')
@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 _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_fallback = False
resource_suffixes: frozenset[str] = frozenset()
router_inherit_keys = frozenset({"file_store", "language"})
@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:
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
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:
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)
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:
"""Find the single daily-card owner for a root-level resource."""
matches: list[tuple[str, str]] = []
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"])))
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 = await self._list_resource_note(day, file_path)
if note is not None:
note_path = str(note["path"])
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 _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) -> 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,
*,
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,
) -> 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)
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,
},
)
return note_path
async def _handle_delete(self, file_path: str, date_str: str, note_stem: str) -> None:
note = await self._list_resource_note(date_str, file_path)
if note 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_rel = str(note["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}")
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."""
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 = {}
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),
}
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
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

View file

@ -28,6 +28,9 @@ IMAGE_MIME_BY_EXT: dict[str, str] = {
".heic": "image/heic",
}
IMAGE_SUFFIXES = frozenset(IMAGE_MIME_BY_EXT)
_INVALID_CHARS = re.compile(r'[<>:"/\\|?*\x00-\x1f]')
_RESERVED_NAMES = {
"CON",

View file

@ -92,6 +92,7 @@ def match_file(file_path: str, rules: list[WatchRule]) -> bool:
def _match_rule(p: Path, rule: WatchRule) -> bool:
"""Check if a single path matches a rule's suffix constraint."""
if rule.suffixes and not any(p.name.endswith("." + s.strip(".")) for s in rule.suffixes):
filename = p.name.casefold()
if rule.suffixes and not any(filename.endswith("." + suffix.strip(".").casefold()) for suffix in rule.suffixes):
return False
return True

View file

@ -25,7 +25,8 @@ sys.path.insert(0, str(INTEGRATION_DIR))
# pylint: disable=wrong-import-position
from _workspace_fixture import workspace_env # noqa: E402
from reme.steps.evolve.auto_resource import _compute_agent_session_id, _compute_note_stem # noqa: E402
from reme.steps.evolve.base_auto_resource import _compute_note_stem # noqa: E402
from reme.steps.evolve.auto_text_resource import _compute_agent_session_id # noqa: E402
RESOURCE_FILENAME = "project-roadmap.md"
RESOURCE_CONTENT_V1 = """\

View file

@ -0,0 +1,5 @@
"""Pytest fixtures loaded only by the auto-resource test modules."""
from .auto_resource_test_support import auto_resource_env
__all__ = ["auto_resource_env"]

View file

@ -0,0 +1,270 @@
"""Shared test harness for auto-resource processor and router tests."""
import io
import json
from dataclasses import dataclass
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import MagicMock
import pytest_asyncio
from agentscope.model import ChatModelBase
from PIL import Image
from reme.components import R
from reme.components.agent_wrapper import BaseAgentWrapper
from reme.components.file_store import LocalFileStore
from reme.components.runtime_context import RuntimeContext
from reme.steps.evolve.auto_image_resource import AutoImageResourceStep
from reme.steps.evolve.auto_resource import AutoResourceStep
from reme.steps.evolve.base_auto_resource import BaseAutoResourceStep
from reme.steps.file_io import DailyListStep, FrontmatterUpdateStep, MoveStep, WriteStep
class FakeAgentWrapper(BaseAgentWrapper):
"""Capture text-processor calls without invoking a real model."""
def __init__(self):
super().__init__()
self.inputs = ""
async def reply(self, inputs, **_kwargs) -> dict:
"""Record and accept one text-processor request."""
self.inputs = inputs
return {"result": "ok"}
class FlakyAgentWrapper(BaseAgentWrapper):
"""Fail one text item, then succeed."""
def __init__(self):
super().__init__()
self.calls = 0
async def reply(self, _inputs, **_kwargs) -> dict:
"""Fail the first request and accept subsequent ones."""
self.calls += 1
if self.calls == 1:
raise RuntimeError("text provider unavailable")
return {"result": "recovered"}
class FakeVisionModel(ChatModelBase):
"""Capture VLM calls and return canned plain text."""
def __init__(self, text: str):
self.text = text
self.calls: list = []
async def generate_structured_output(self, messages, structured_model, **kwargs): # pylint: disable=unused-argument
"""Force callers through the plain fallback."""
raise NotImplementedError("structured path not faked")
async def __call__(self, messages, **kwargs):
"""Record a call and return the canned plain text."""
self.calls.append(messages)
return SimpleNamespace(content=[{"type": "text", "text": self.text}])
class FlakyVisionModel(ChatModelBase):
"""Fail the first plain call, then succeed."""
def __init__(self, text: str):
self.text = text
self.calls = 0
async def generate_structured_output(self, messages, structured_model, **kwargs): # pylint: disable=unused-argument
"""Force callers through the plain fallback."""
raise NotImplementedError("structured path not faked")
async def __call__(self, messages, **kwargs):
"""Fail once, then return the canned plain text."""
self.calls += 1
if self.calls == 1:
raise RuntimeError("vision backend unavailable")
return SimpleNamespace(content=[{"type": "text", "text": self.text}])
class StructuredVisionModel(ChatModelBase):
"""Serve structured output and count fallback plain calls."""
def __init__(self, content: dict | None = None, error: Exception | None = None, plain_text: str = "plain"):
self.content = content
self.error = error
self.plain_text = plain_text
self.structured_calls: list = []
self.plain_calls: list = []
async def generate_structured_output(self, messages, structured_model, **kwargs): # pylint: disable=unused-argument
"""Return or fail with the configured structured response."""
self.structured_calls.append(messages)
if self.error is not None:
raise self.error
return SimpleNamespace(content=dict(self.content or {}))
async def __call__(self, messages, **kwargs): # pylint: disable=unused-argument
"""Record and return the configured plain fallback."""
self.plain_calls.append(messages)
return SimpleNamespace(content=[{"type": "text", "text": self.plain_text}])
class FakeAudioResourceStep(BaseAutoResourceStep):
"""Minimal third modality used to verify the router extension contract."""
resource_suffixes = frozenset({".wav"})
async def _handle_upsert(
self,
file_path: str,
date_str: str,
note_stem: str,
added: bool,
source_path: Path,
) -> None:
del source_path
self.context.response.success = True
self.context.response.answer = f"Processed audio resource: {file_path}"
self.context.response.metadata.update(
{
"path": f"daily/{date_str}/{note_stem}.md",
"action": "added" if added else "modified",
"processor": "audio",
"modified": True,
},
)
class _StepJob:
"""Tiny job adapter for tests that need ``BaseStep.run_job``."""
def __init__(self, step_cls, app_context, file_store):
self.step_cls = step_cls
self.app_context = app_context
self.file_store = file_store
async def __call__(self, **kwargs):
step = self.step_cls(app_context=self.app_context, file_store=self.file_store)
result = await step(**kwargs)
return result or step.context.response
def make_app_context(workspace: Path):
"""Create the minimal application context used by resource tests."""
context = MagicMock()
context.app_config.workspace_dir = str(workspace)
context.app_config.daily_dir = "daily"
context.app_config.digest_dir = "digest"
context.app_config.resource_dir = "resource"
context.app_config.session_dir = "session"
context.app_config.timezone = None
return context
def _install_file_jobs(app_context, file_store) -> None:
app_context.jobs = {
"daily_list": _StepJob(DailyListStep, app_context, file_store),
"frontmatter_update": _StepJob(FrontmatterUpdateStep, app_context, file_store),
"move": _StepJob(MoveStep, app_context, file_store),
"write": _StepJob(WriteStep, app_context, file_store),
}
def image_bytes(image_format: str = "PNG", size=(8, 8), color=(200, 30, 30)) -> bytes:
"""Synthesize a small image in a Pillow-supported format."""
if image_format == "HEIF":
from pillow_heif import register_heif_opener
register_heif_opener()
image = Image.new("RGB", size, color)
buffer = io.BytesIO()
image.save(buffer, format=image_format)
return buffer.getvalue()
def png_bytes(width: int = 8, height: int = 8, color=(200, 30, 30)) -> bytes:
"""Compatibility shorthand for PNG-focused assertions."""
return image_bytes("PNG", (width, height), color)
def write_binary(path: Path, data: bytes) -> Path:
"""Write test bytes, creating parent directories."""
path.parent.mkdir(parents=True, exist_ok=True)
path.write_bytes(data)
return path
def write_note(path: Path, source_resource: str, body: str = "old caption") -> Path:
"""Write a minimal source-owned image note."""
content = (
f"---\nname: {path.stem}\ndescription: old\n"
f'source_resource: "{source_resource}"\nkind: image\n'
f"media_type: image/png\n---\n![[{source_resource[2:-2]}]]\n\n## Caption\n\n{body}\n"
)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content, encoding="utf-8")
return path
def caption_json(name: str, description: str, caption: str) -> str:
"""Build a plain-call caption payload."""
return json.dumps({"name": name, "description": description, "caption": caption})
def image_processor(app_context, file_store, model, *, routed: bool, **kwargs):
"""Build either the image processor or the public unified-router path."""
if not routed:
return AutoImageResourceStep(app_context=app_context, file_store=file_store, as_llm=model, **kwargs)
app_context.registry = R
return AutoResourceStep(
app_context=app_context,
**kwargs,
dispatch_steps=[
{"backend": "auto_image_resource_step", "file_store": file_store, "as_llm": model},
{
"backend": "auto_text_resource_step",
"file_store": file_store,
"agent_wrapper": FakeAgentWrapper(),
},
],
)
@dataclass
class AutoResourceTestEnv:
"""Started, isolated workspace shared by one test invocation."""
workspace: Path
app_context: object
file_store: LocalFileStore
def write_binary(self, relative_path: str, data: bytes) -> Path:
"""Write bytes relative to this workspace."""
return write_binary(self.workspace / relative_path, data)
def write_note(self, relative_path: str, source_resource: str, body: str = "old caption") -> Path:
"""Write a source-owned note relative to this workspace."""
return write_note(self.workspace / relative_path, source_resource, body)
def processor(self, model, *, routed: bool = False, **kwargs):
"""Build the direct processor or unified router for this workspace."""
return image_processor(self.app_context, self.file_store, model, routed=routed, **kwargs)
async def run(self, step, changes, **context_kwargs):
"""Run one processor invocation with a fresh runtime context."""
return await step(RuntimeContext(changes=changes, **context_kwargs))
@pytest_asyncio.fixture
async def auto_resource_env(tmp_path, monkeypatch):
"""Yield a started resource-test workspace and always close its file store."""
workspace = tmp_path / "workspace"
workspace.mkdir()
monkeypatch.chdir(workspace)
app_context = make_app_context(workspace)
file_store = LocalFileStore(name="test_store", embedding_store="")
await file_store.start()
_install_file_jobs(app_context, file_store)
try:
yield AutoResourceTestEnv(workspace, app_context, file_store)
finally:
await file_store.close()

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,410 @@
"""Regression tests for the safety findings from the Auto Resource PR review."""
from pathlib import Path
from unittest.mock import AsyncMock, patch
import frontmatter
import pytest
from reme.components import R
from reme.steps.evolve.auto_resource import AutoResourceStep
from reme.steps.evolve.auto_text_resource import AutoTextResourceStep
from reme.steps.evolve.base_auto_resource import BaseAutoResourceStep
from .auto_resource_test_support import (
FakeAgentWrapper,
FakeVisionModel,
StructuredVisionModel,
caption_json,
image_bytes,
write_binary,
write_note,
)
pytest_plugins = ("unit.auto_resource_test_plugin",)
pytestmark = pytest.mark.asyncio
def _link_daily_directory(workspace: Path, day: str, target: Path) -> None:
"""Expose a fixture directory through the supported daily layout."""
link = workspace / "daily" / day
link.parent.mkdir(parents=True, exist_ok=True)
try:
link.symlink_to(target, target_is_directory=True)
except OSError as exc:
pytest.skip(f"symlinks unavailable: {exc}")
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
async def test_image_rejects_paths_outside_the_resource_tree(routed, auto_resource_env, tmp_path):
"""Traversal and external paths fail before image reads."""
env = auto_resource_env
outside = write_binary(tmp_path / "outside.png", image_bytes())
nonresource = env.write_binary("private.png", image_bytes())
model = FakeVisionModel(caption_json("unsafe", "Unsafe", "Must not be read."))
response = await env.run(
env.processor(model, routed=routed),
[
{"change": "added", "path": "resource/2026-01-01/../../../outside.png"},
{"change": "added", "path": str(outside)},
{"change": "added", "path": str(nonresource)},
],
)
results = response.metadata["results"]
assert response.success is False
assert len(results) == 3
assert all(item["metadata"]["action"] == "failed" for item in results)
assert all(item["metadata"]["modified"] is False for item in results)
assert "cannot contain '.' or '..'" in results[0]["metadata"]["error"]
assert "must stay inside the workspace" in results[1]["metadata"]["error"]
assert "configured resource directory" in results[2]["metadata"]["error"]
assert not model.calls
assert outside.read_bytes() == image_bytes()
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
async def test_image_rejects_resource_symlink_outside_workspace(routed, auto_resource_env, tmp_path):
"""An escaping resource symlink fails without weakening other containment tests."""
env = auto_resource_env
outside = write_binary(tmp_path / "outside.png", image_bytes())
external_link = env.workspace / "resource/2026-01-01/external.png"
external_link.parent.mkdir(parents=True, exist_ok=True)
try:
external_link.symlink_to(outside)
except OSError as exc:
pytest.skip(f"symlinks unavailable: {exc}")
model = FakeVisionModel(caption_json("unsafe", "Unsafe", "Must not be read."))
response = await env.run(
env.processor(model, routed=routed),
[{"change": "added", "path": str(external_link)}],
)
result = response.metadata["results"][0]
assert response.success is False
assert result["metadata"]["action"] == "failed"
assert result["metadata"]["modified"] is False
assert "must stay inside the workspace" in result["metadata"]["error"]
assert not model.calls
assert outside.read_bytes() == image_bytes()
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
async def test_image_internal_symlink_keeps_logical_provenance(routed, auto_resource_env):
"""An internal symlink is read safely while ownership follows the watched alias."""
env = auto_resource_env
target = env.write_binary("resource/2026-01-01/original.png", image_bytes())
link = target.with_name("alias.png")
try:
link.symlink_to(target.name)
except OSError as exc:
pytest.skip(f"symlinks unavailable: {exc}")
model = FakeVisionModel(caption_json("linked-image", "Linked", "An internal linked image."))
step = env.processor(model, routed=routed)
response = await env.run(step, [{"change": "added", "path": str(link)}])
note_path = env.workspace / "daily/2026-01-01/linked-image.md"
post = frontmatter.loads(note_path.read_text(encoding="utf-8"))
assert response.success is True
assert post.metadata["source_resource"] == "[[resource/2026-01-01/alias.png]]"
assert "![[resource/2026-01-01/alias.png]]" in post.content
assert len(model.calls) == 1
link.unlink()
deleted = await env.run(step, [{"change": "deleted", "path": str(link)}])
assert deleted.success is True
assert deleted.metadata["results"][0]["metadata"]["action"] == "deleted"
assert not note_path.exists()
assert target.is_file()
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@pytest.mark.parametrize("existing_owner", ["different-source", "no-source"])
@pytest.mark.parametrize("change", ["modified", "deleted"])
async def test_image_preserves_unowned_same_stem_note(routed, existing_owner, change, auto_resource_env):
"""Upsert and delete never claim a same-stem note without exact ownership."""
env = auto_resource_env
source = env.workspace / "resource/2026-01-01/img.png"
if change == "modified":
write_binary(source, image_bytes())
same_stem = env.workspace / "daily/2026-01-01/img.md"
if existing_owner == "different-source":
write_note(same_stem, "[[resource/2026-01-01/other.png]]", body="unrelated image note")
else:
same_stem.parent.mkdir(parents=True, exist_ok=True)
same_stem.write_text(
"---\nname: img\ndescription: user-owned note\n---\nuser-owned bytes\n",
encoding="utf-8",
)
before = same_stem.read_bytes()
model = FakeVisionModel(caption_json("generated-caption", "Generated", "Generated caption."))
response = await env.run(env.processor(model, routed=routed), [{"change": change, "path": str(source)}])
assert response.success is True
assert same_stem.read_bytes() == before
if change == "deleted":
result = response.metadata["results"][0]["metadata"]
assert result["action"] == "skipped"
assert result["reason"] == "resource_note_not_found"
assert result["modified"] is False
else:
generated = env.workspace / "daily/2026-01-01/generated-caption.md"
post = frontmatter.loads(generated.read_text(encoding="utf-8"))
assert post.metadata["source_resource"] == "[[resource/2026-01-01/img.png]]"
assert "Generated caption." in post.content
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@pytest.mark.parametrize("plain_text", [" ", "```json\n\n```"], ids=["whitespace", "empty-json-fence"])
async def test_blank_plain_caption_does_not_create_or_overwrite_note(routed, plain_text, auto_resource_env):
"""An empty structured result plus blank plain fallback leaves notes untouched."""
env = auto_resource_env
new_source = env.write_binary("resource/2026-01-01/blank-new.png", image_bytes())
old_source = env.write_binary("resource/2026-01-01/blank-old.png", image_bytes())
old_note = env.write_note(
"daily/2026-01-01/preserved.md",
"[[resource/2026-01-01/blank-old.png]]",
body="caption that must survive",
)
before = old_note.read_bytes()
model = StructuredVisionModel(content={}, plain_text=plain_text)
step = env.processor(model, routed=routed)
added = await env.run(step, [{"change": "added", "path": str(new_source)}])
modified = await env.run(step, [{"change": "modified", "path": str(old_source)}])
for response in (added, modified):
result = response.metadata["results"][0]["metadata"]
assert response.success is False
assert result["action"] == "failed"
assert result["modified"] is False
assert "no usable caption" in result["error"]
assert not (env.workspace / "daily/2026-01-01/blank-new.md").exists()
assert old_note.read_bytes() == before
assert len(model.structured_calls) == len(model.plain_calls) == 2
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@pytest.mark.parametrize("linked_day", [False, True], ids=["directory", "internal-symlink"])
async def test_loose_root_image_keeps_original_daily_card_across_days(routed, linked_day, auto_resource_env):
"""Later updates and deletion keep a loose resource's first daily-card ownership."""
env = auto_resource_env
if linked_day:
archive = env.workspace / "archive-day"
archive.mkdir()
_link_daily_directory(env.workspace, "2026-01-01", archive)
source = env.write_binary("resource/photo.png", image_bytes(color=(200, 30, 30)))
initial_model = FakeVisionModel(caption_json("original-card", "Original", "first-day caption"))
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-01"):
added = await env.run(
env.processor(initial_model, routed=routed),
[{"change": "added", "path": str(source)}],
)
owned_note = env.workspace / "daily/2026-01-01/original-card.md"
add_result = added.metadata["results"][0]["metadata"]
assert added.success is True
assert add_result["path"] == "daily/2026-01-01/original-card.md"
assert add_result["action"] == "added"
assert add_result["index"]["date"] == "2026-01-01"
assert "first-day caption" in owned_note.read_text(encoding="utf-8")
unrelated_note = env.write_note(
"daily/2026-01-02/photo.md",
"[[resource/other.png]]",
body="unrelated note that must survive",
)
unrelated_before = unrelated_note.read_bytes()
first_model = FakeVisionModel(caption_json("renamed-on-day-two", "Updated", "second-day caption"))
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
first_update = await env.run(
env.processor(first_model, routed=routed),
[{"change": "modified", "path": str(source)}],
)
first_result = first_update.metadata["results"][0]["metadata"]
assert first_update.success is True
assert first_result["path"] == "daily/2026-01-01/original-card.md"
assert first_result["action"] == "modified"
assert first_result["created"] is False
assert first_result["index"]["date"] == "2026-01-01"
assert "second-day caption" in owned_note.read_text(encoding="utf-8")
assert not (env.workspace / "daily/2026-01-02/renamed-on-day-two.md").exists()
assert unrelated_note.read_bytes() == unrelated_before
assert not (env.workspace / "daily/2026-01-02.md").exists()
source.write_bytes(image_bytes(color=(20, 90, 200)))
second_model = FakeVisionModel(caption_json("renamed-on-day-three", "Updated again", "third-day caption"))
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-03"):
second_update = await env.run(
env.processor(second_model, routed=routed),
[{"change": "modified", "path": str(source)}],
)
second_result = second_update.metadata["results"][0]["metadata"]
assert second_update.success is True
assert second_result["path"] == "daily/2026-01-01/original-card.md"
assert second_result["action"] == "modified"
assert second_result["created"] is False
assert second_result["index"]["date"] == "2026-01-01"
assert "third-day caption" in owned_note.read_text(encoding="utf-8")
assert not (env.workspace / "daily/2026-01-03/renamed-on-day-three.md").exists()
assert unrelated_note.read_bytes() == unrelated_before
assert not (env.workspace / "daily/2026-01-03.md").exists()
source.unlink()
delete_model = FakeVisionModel(caption_json("unused", "Unused", "Must not be requested."))
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-04"):
deleted = await env.run(
env.processor(delete_model, routed=routed),
[{"change": "deleted", "path": str(source)}],
)
delete_result = deleted.metadata["results"][0]["metadata"]
assert deleted.success is True
assert delete_result["path"] == "daily/2026-01-01/original-card.md"
assert delete_result["action"] == "deleted"
assert delete_result["modified"] is True
assert delete_result["index"]["date"] == "2026-01-01"
assert delete_result["index"]["notes"] == []
assert not owned_note.exists()
assert unrelated_note.read_bytes() == unrelated_before
assert "(none)" in (env.workspace / "daily/2026-01-01.md").read_text(encoding="utf-8")
assert not (env.workspace / "daily/2026-01-04.md").exists()
assert len(initial_model.calls) == len(first_model.calls) == len(second_model.calls) == 1
assert not delete_model.calls
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
@pytest.mark.parametrize("linked_day", [False, True], ids=["directory", "internal-symlink"])
async def test_loose_root_image_duplicate_daily_owners_fail_closed(routed, linked_day, auto_resource_env):
"""Ambiguous exact ownership is reported without reading the image model or changing notes."""
env = auto_resource_env
if linked_day:
archive = env.workspace / "archive-day"
archive.mkdir()
_link_daily_directory(env.workspace, "2026-01-01", archive)
source = env.write_binary("resource/duplicate.png", image_bytes())
first_note = env.write_note("daily/2026-01-01/first.md", "[[resource/duplicate.png]]", body="first owner")
second_note = env.write_note("daily/2026-01-02/second.md", "[[resource/duplicate.png]]", body="second owner")
before = {first_note: first_note.read_bytes(), second_note: second_note.read_bytes()}
model = FakeVisionModel(caption_json("replacement", "Replacement", "Must not be generated."))
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-03"):
response = await env.run(
env.processor(model, routed=routed),
[{"change": "modified", "path": str(source)}],
)
result = response.metadata["results"][0]["metadata"]
assert response.success is False
assert result["action"] == "failed"
assert result["modified"] is False
assert "Multiple daily resource notes claim resource/duplicate.png" in result["error"]
assert "daily/2026-01-01/first.md" in result["error"]
assert "daily/2026-01-02/second.md" in result["error"]
assert not model.calls
assert all(path.read_bytes() == contents for path, contents in before.items())
assert not (env.workspace / "daily/2026-01-01.md").exists()
assert not (env.workspace / "daily/2026-01-02.md").exists()
assert not (env.workspace / "daily/2026-01-03.md").exists()
@pytest.mark.parametrize("routed", [False, True], ids=["image", "unified-router"])
async def test_loose_root_lookup_ignores_unsafe_daily_links(routed, auto_resource_env, tmp_path):
"""Outside, missing, and cyclic directories cannot expose notes or block a safe owner."""
env = auto_resource_env
outside = write_note(tmp_path / "outside-day/claim.md", "[[resource/photo.png]]", body="private outside note")
outside_before = outside.read_bytes()
_link_daily_directory(env.workspace, "2025-12-01", outside.parent)
_link_daily_directory(env.workspace, "2025-12-02", env.workspace / "missing-day")
_link_daily_directory(env.workspace, "2025-12-03", env.workspace / "daily/2025-12-03")
source = env.write_binary("resource/photo.png", image_bytes())
owned_note = env.write_note("daily/2026-01-01/original.md", "[[resource/photo.png]]")
model = FakeVisionModel(caption_json("original", "Updated", "Updated inside workspace."))
original_read_text = Path.read_text
outside_reads = []
def guarded_read_text(path, *args, **kwargs):
if path.resolve() == outside.resolve():
outside_reads.append(path)
raise AssertionError("cross-date lookup read a note outside the workspace")
return original_read_text(path, *args, **kwargs)
with patch.object(Path, "read_text", guarded_read_text):
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
updated = await env.run(env.processor(model, routed=routed), [{"change": "modified", "path": str(source)}])
assert updated.success is True
assert updated.metadata["results"][0]["metadata"]["path"] == "daily/2026-01-01/original.md"
assert "Updated inside workspace." in owned_note.read_text(encoding="utf-8")
source.unlink()
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-03"):
deleted = await env.run(env.processor(model, routed=routed), [{"change": "deleted", "path": str(source)}])
assert deleted.success is True
assert deleted.metadata["results"][0]["metadata"]["action"] == "deleted"
assert not owned_note.exists()
assert not outside_reads
assert outside.read_bytes() == outside_before
assert len(model.calls) == 1
assert not (env.workspace / "daily/2026-01-02").exists()
assert not (env.workspace / "daily/2026-01-03").exists()
@pytest.mark.parametrize("routed", [False, True], ids=["text", "unified-router"])
async def test_loose_root_text_updates_and_deletes_original_daily_card(routed, auto_resource_env):
"""Text processing also uses exact cross-date ownership for its prompt and lifecycle."""
env = auto_resource_env
source = env.write_binary("resource/report.txt", b"Updated report text.")
note_rel = "daily/2026-01-01/original.md"
owned_note = env.write_note(note_rel, "[[resource/report.txt]]", body="Original report text.")
unrelated = env.write_note("daily/2026-01-02/report.md", "[[resource/other.txt]]", body="Keep unrelated report.")
unrelated_before = unrelated.read_bytes()
wrapper = FakeAgentWrapper()
async def update_note(inputs, **kwargs):
assert "Date: 2026-01-01" in inputs
assert f"Target note path: {note_rel}" in inputs
assert "Updated report text." in inputs
assert "read" in kwargs["job_tools"]
response = await env.app_context.jobs["write"](
path=note_rel,
name="suggested-rename",
description="Updated report",
content="Updated report text.",
metadata={"source_resource": "[[resource/report.txt]]"},
)
assert response.success is True
return {"result": "Updated original report."}
env.app_context.registry = R
step_cls = AutoResourceStep if routed else AutoTextResourceStep
step = step_cls(app_context=env.app_context, file_store=env.file_store, agent_wrapper=wrapper, language="en")
with patch.object(wrapper, "reply", new=AsyncMock(side_effect=update_note)) as reply:
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-02"):
updated = await env.run(step, [{"change": "modified", "path": str(source)}])
assert updated.success is True
result = updated.metadata["results"][0]["metadata"]
assert result["path"] == note_rel
assert result["created"] is False
assert result["action"] == "modified"
assert result["index"]["date"] == "2026-01-01"
assert "Updated report text." in owned_note.read_text(encoding="utf-8")
source.unlink()
with patch.object(BaseAutoResourceStep, "_today", return_value="2026-01-03"):
deleted = await env.run(step, [{"change": "deleted", "path": str(source)}])
reply.assert_awaited_once()
assert deleted.success is True
assert deleted.metadata["results"][0]["metadata"]["path"] == note_rel
assert deleted.metadata["results"][0]["metadata"]["action"] == "deleted"
assert deleted.metadata["results"][0]["metadata"]["index"]["date"] == "2026-01-01"
assert not owned_note.exists()
assert unrelated.read_bytes() == unrelated_before
assert list((env.workspace / "daily/2026-01-02").glob("*.md")) == [unrelated]
assert not (env.workspace / "daily/2026-01-03").exists()

View file

@ -31,7 +31,9 @@ from reme.components.file_store import LocalFileStore
from reme.components.runtime_context import RuntimeContext
from reme.enumeration import ComponentEnum
from reme.steps.evolve.auto_memory import AutoMemoryStep
from reme.steps.evolve.auto_resource import AutoResourceStep, _compute_note_stem
from reme.steps.evolve.base_auto_resource import _compute_note_stem
from reme.steps.evolve.auto_resource import AutoResourceStep
from reme.steps.evolve.auto_text_resource import AutoTextResourceStep
from reme.steps.file_io.daily_list import DailyListStep
from reme.steps.file_io.frontmatter_update import FrontmatterUpdateStep
from reme.steps.file_io.move import MoveStep
@ -187,6 +189,14 @@ def test_match_file_suffix():
print("✓ test_match_file_suffix passed")
def test_match_file_suffix_is_case_insensitive():
"""Configured suffixes match uppercase file extensions."""
rules = [WatchRule(path=Path("/workspace/resource"), suffixes=["jpg", ".png"])]
assert match_file("/workspace/resource/photo.JPG", rules)
assert match_file("/workspace/resource/sub/diagram.PNG", rules)
assert not match_file("/workspace/resource/photo.GIF", rules)
def test_match_file_no_suffix_filter():
"""Empty suffixes list means all files match."""
rules = [WatchRule(path=Path("/workspace/resource"), suffixes=[])]
@ -220,6 +230,18 @@ def test_collect_existing_filters():
print("✓ test_collect_existing_filters passed")
def test_collect_existing_matches_uppercase_suffixes():
"""The initial scan includes files whose extension casing differs from the rule."""
with tempfile.TemporaryDirectory() as tmpdir:
resource = Path(tmpdir) / "resource"
uppercase = write_file(resource / "photo.JPG")
write_file(resource / "ignore.GIF")
result = collect_existing([WatchRule(path=resource, suffixes=["jpg"])], recursive=True)
assert set(result) == {str(uppercase.absolute())}
# ---------------------------------------------------------------------------
# InitChangesStep
# ---------------------------------------------------------------------------
@ -1171,6 +1193,21 @@ def test_watch_changes_filter_matches_rules():
print("✓ test_watch_changes_filter_matches_rules passed")
def test_watch_changes_filter_matches_uppercase_suffixes():
"""The live watcher accepts uppercase extensions configured in lowercase."""
with tempfile.TemporaryDirectory() as tmpdir:
workspace = Path(tmpdir)
(workspace / "resource").mkdir()
app_ctx = _make_app_context(workspace)
step = WatchChangesStep(app_context=app_ctx)
step.context = RuntimeContext(watch_dirs=["resource_dir"], watch_suffixes=["jpg"])
step._rules = step._get_watch_rules()
assert step._filter(Change.added, str(workspace / "resource/photo.JPG"))
assert not step._filter(Change.added, str(workspace / "resource/photo.PNG"))
def test_watch_changes_dispatch_steps_list():
"""dispatch_steps config is stored by BaseStep."""
step = WatchChangesStep(dispatch_steps=["update_catalog_step", "auto_resource_step"])
@ -1198,7 +1235,7 @@ def test_auto_resource_batch_deleted_changes():
"---\nname: test\nsource_resource: '[[resource/2026-01-01/file.md]]'\n---\nbody\n",
)
step = AutoResourceStep(app_context=app_ctx, file_store=fs)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs)
ctx = RuntimeContext(
changes=[
{"change": "deleted", "path": str(cwd / "resource" / "2026-01-01" / filename)},
@ -1231,7 +1268,7 @@ def test_auto_resource_skips_oversized_file_before_reading():
_install_file_jobs(app_ctx, fs)
try:
source = write_file(cwd / "resource" / "2026-01-01" / "large.txt", "too large")
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(
changes=[{"change": "added", "path": str(source)}],
@ -1266,7 +1303,7 @@ def test_auto_resource_batch_keeps_result_metadata_isolated():
large = write_file(cwd / "resource" / "2026-01-01" / "large.txt", "too large")
small = write_file(cwd / "resource" / "2026-01-01" / "small.txt", "ok")
second_large = write_file(cwd / "resource" / "2026-01-01" / "second-large.txt", "also large")
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(
changes=[
@ -1318,7 +1355,7 @@ def test_auto_resource_handles_file_removed_before_stat():
raise FileNotFoundError("file disappeared")
return original_stat(path, *args, **kwargs)
step = AutoResourceStep(app_context=app_ctx, file_store=fs)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs)
with patch.object(Path, "stat", disappearing_stat):
resp = await step(
RuntimeContext(changes=[{"change": "added", "path": str(source)}]),
@ -1335,6 +1372,155 @@ def test_auto_resource_handles_file_removed_before_stat():
asyncio.run(run())
def test_auto_resource_rejects_paths_outside_resource_scope_before_agent_call():
"""Traversal, outside absolute paths, and escaping symlinks fail closed."""
async def run():
with (
tempfile.TemporaryDirectory() as tmpdir,
tempfile.TemporaryDirectory() as outside_dir,
temp_chdir(tmpdir),
):
workspace = Path.cwd()
outside = write_file(Path(outside_dir) / "outside.txt", "secret")
workspace_outside = write_file(workspace / "daily" / "outside.txt", "workspace secret")
link = workspace / "resource" / "2026-01-01" / "escape.txt"
link.parent.mkdir(parents=True, exist_ok=True)
link.symlink_to(outside)
wrapper = _FakeAgentWrapper()
step = AutoTextResourceStep(app_context=_make_app_context(workspace), agent_wrapper=wrapper)
resp = await step(
RuntimeContext(
changes=[
{"change": "added", "path": "resource/2026-01-01/../../../outside.txt"},
{"change": "modified", "path": str(outside)},
{"change": "added", "path": str(workspace_outside)},
{"change": "added", "path": str(link)},
],
),
)
assert resp.success is False
assert wrapper.inputs == ""
assert len(resp.metadata["results"]) == 4
assert all(result["metadata"]["action"] == "failed" for result in resp.metadata["results"])
assert all(result["metadata"]["modified"] is False for result in resp.metadata["results"])
assert outside.read_text(encoding="utf-8") == "secret"
assert workspace_outside.read_text(encoding="utf-8") == "workspace secret"
asyncio.run(run())
def test_auto_resource_rejects_traversal_delete_without_touching_note():
"""A malicious deleted path cannot reach or remove a daily note."""
async def run():
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
workspace = Path.cwd()
note = write_file(
workspace / "daily" / "2026-01-01" / "outside.md",
"---\nname: outside\n"
"source_resource: '[[resource/2026-01-01/../../../outside.txt]]'\n---\nkeep me\n",
)
step = AutoTextResourceStep(app_context=_make_app_context(workspace))
resp = await step(
RuntimeContext(
changes=[
{"change": "deleted", "path": "resource/2026-01-01/../../../outside.txt"},
],
),
)
result = resp.metadata["results"][0]
assert resp.success is False
assert result["metadata"]["action"] == "failed"
assert result["metadata"]["modified"] is False
assert note.read_text(encoding="utf-8").endswith("keep me\n")
asyncio.run(run())
def test_auto_resource_internal_symlink_keeps_logical_source_identity():
"""A safe internal symlink is read by target while provenance keeps the link path."""
async def run():
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
workspace = Path.cwd()
target = write_file(workspace / "resource" / "2026-01-01" / "target.txt", "visible")
link = workspace / "resource" / "2026-01-01" / "link.txt"
link.symlink_to(target)
captured = {}
step = AutoTextResourceStep(app_context=_make_app_context(workspace))
async def fake_upsert(file_path, date_str, note_stem, added, source_path):
captured.update(
{
"file_path": file_path,
"date_str": date_str,
"note_stem": note_stem,
"added": added,
"source_path": source_path,
},
)
step.context.response.success = True
step.context.response.answer = "ok"
step._handle_upsert = fake_upsert
resp = await step(RuntimeContext(changes=[{"change": "added", "path": str(link)}]))
assert resp.success is True
assert captured == {
"file_path": "resource/2026-01-01/link.txt",
"date_str": "2026-01-01",
"note_stem": "link",
"added": True,
"source_path": target.resolve(),
}
asyncio.run(run())
def test_auto_resource_accepts_absolute_resource_dir_inside_workspace():
"""An absolute in-workspace resource_dir keeps a workspace-relative source identity."""
async def run():
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
workspace = Path.cwd()
resource_dir = workspace / "assets"
source = write_file(resource_dir / "2026-01-01" / "report.txt", "visible")
step = AutoTextResourceStep(app_context=_make_app_context(workspace, resource_dir=str(resource_dir)))
captured = {}
async def fake_upsert(file_path, date_str, note_stem, added, source_path):
captured.update(
{
"file_path": file_path,
"date_str": date_str,
"note_stem": note_stem,
"added": added,
"source_path": source_path,
},
)
step.context.response.success = True
step.context.response.answer = "ok"
step._handle_upsert = fake_upsert
response = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}]))
assert response.success is True
assert captured == {
"file_path": "assets/2026-01-01/report.txt",
"date_str": "2026-01-01",
"note_stem": "report",
"added": True,
"source_path": source.resolve(),
}
asyncio.run(run())
def test_auto_resource_accepts_loose_root_resource():
"""Root-level resource files use today's date without moving the source."""
@ -1346,11 +1532,17 @@ def test_auto_resource_accepts_loose_root_resource():
today = datetime.datetime.now().strftime("%Y-%m-%d")
captured = {}
step = AutoResourceStep(app_context=app_ctx)
step = AutoTextResourceStep(app_context=app_ctx)
async def fake_upsert(file_path, date_str, note_stem, created):
async def fake_upsert(file_path, date_str, note_stem, created, source_path):
captured.update(
{"file_path": file_path, "date_str": date_str, "note_stem": note_stem, "created": created},
{
"file_path": file_path,
"date_str": date_str,
"note_stem": note_stem,
"created": created,
"source_path": source_path,
},
)
step.context.response.success = True
step.context.response.answer = "ok"
@ -1367,6 +1559,7 @@ def test_auto_resource_accepts_loose_root_resource():
"date_str": today,
"note_stem": "report",
"created": True,
"source_path": source.resolve(),
}
print("✓ test_auto_resource_accepts_loose_root_resource passed")
@ -1385,11 +1578,17 @@ def test_auto_resource_loose_root_resource_keeps_existing_dated_resource():
source = write_file(workspace / "resource" / "report.txt", "new")
captured = {}
step = AutoResourceStep(app_context=app_ctx)
step = AutoTextResourceStep(app_context=app_ctx)
async def fake_upsert(file_path, date_str, note_stem, created):
async def fake_upsert(file_path, date_str, note_stem, created, source_path):
captured.update(
{"file_path": file_path, "date_str": date_str, "note_stem": note_stem, "created": created},
{
"file_path": file_path,
"date_str": date_str,
"note_stem": note_stem,
"created": created,
"source_path": source_path,
},
)
step.context.response.success = True
step.context.response.answer = "ok"
@ -1406,6 +1605,7 @@ def test_auto_resource_loose_root_resource_keeps_existing_dated_resource():
"date_str": today,
"note_stem": "report",
"created": True,
"source_path": source.resolve(),
}
print("✓ test_auto_resource_loose_root_resource_keeps_existing_dated_resource passed")
@ -1429,7 +1629,7 @@ def test_auto_resource_modified_missing_note_uses_create_tools():
cwd / "daily" / "2026-01-01" / "report.md",
"---\nname: resource-summary\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n",
)
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]),
)
@ -1449,6 +1649,98 @@ def test_auto_resource_modified_missing_note_uses_create_tools():
asyncio.run(run())
def test_auto_resource_create_preserves_unowned_same_stem_note():
"""A no-source same-stem note is user-owned and never used as the staging path."""
async def run():
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
workspace = Path.cwd()
app_ctx = _make_app_context(workspace)
fs = LocalFileStore(name="test_store", embedding_store="")
wrapper = _FakeAgentWrapper()
await fs.start()
_install_file_jobs(app_ctx, fs)
try:
source = write_file(workspace / "resource" / "2026-01-01" / "report.txt", "resource body")
user_note = write_file(
workspace / "daily" / "2026-01-01" / "report.md",
"---\nname: report\ndescription: private user note\n---\nkeep this body\n",
)
original = user_note.read_bytes()
def write_allocated_target(inputs, _kwargs):
target_line = next(
line for line in str(inputs).splitlines() if line.startswith("Target note path: ")
)
target_path = target_line.removeprefix("Target note path: ").strip()
assert target_path != "daily/2026-01-01/report.md"
write_file(
workspace / target_path,
"---\nname: generated-topic\ndescription: resource summary\n"
"source_resource: '[[resource/2026-01-01/report.txt]]'\n---\nsummary\n",
)
wrapper.on_reply = write_allocated_target
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}]))
result_meta = resp.metadata["results"][0]["metadata"]
assert resp.success is True
assert result_meta["created"] is True
assert result_meta["path"] == "daily/2026-01-01/generated-topic.md"
assert user_note.read_bytes() == original
assert (workspace / result_meta["path"]).is_file()
finally:
await fs.close()
asyncio.run(run())
def test_auto_resource_reports_modified_when_post_write_lookup_fails():
"""A text note written before post-processing failure remains reported as modified."""
async def run():
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
workspace = Path.cwd()
app_ctx = _make_app_context(workspace)
file_store = LocalFileStore(name="test_store", embedding_store="")
wrapper = _FakeAgentWrapper()
await file_store.start()
_install_file_jobs(app_ctx, file_store)
try:
source = write_file(workspace / "resource/2026-01-01/report.txt", "resource body")
wrapper.on_reply = lambda *_: write_file(
workspace / "daily/2026-01-01/report.md",
"---\nname: report\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nsummary\n",
)
step = AutoTextResourceStep(app_context=app_ctx, file_store=file_store, agent_wrapper=wrapper)
list_resource_note = step._list_resource_note
calls = 0
async def fail_second_lookup(day, file_path):
nonlocal calls
calls += 1
if calls == 2:
raise RuntimeError("daily_list failed after agent write")
return await list_resource_note(day, file_path)
step._list_resource_note = fail_second_lookup
response = await step(RuntimeContext(changes=[{"change": "added", "path": str(source)}]))
result = response.metadata["results"][0]
assert response.success is False
assert result["metadata"]["action"] == "failed"
assert result["metadata"]["path"] == "daily/2026-01-01/report.md"
assert result["metadata"]["created"] is True
assert result["metadata"]["modified"] is True
assert "daily_list failed after agent write" in result["metadata"]["error"]
assert (workspace / "daily/2026-01-01/report.md").is_file()
finally:
await file_store.close()
asyncio.run(run())
def test_auto_resource_sanitizes_invalid_generated_name():
"""Invalid LLM-suggested names are sanitized before renaming."""
@ -1466,7 +1758,7 @@ def test_auto_resource_sanitizes_invalid_generated_name():
cwd / "daily" / "2026-01-01" / "report.md",
"---\nname: bad/name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n",
)
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(changes=[{"change": "added", "path": "resource/2026-01-01/report.txt"}]),
)
@ -1503,7 +1795,7 @@ def test_auto_resource_uniquifies_conflicting_generated_name():
cwd / "daily" / "2026-01-01" / "report.md",
"---\nname: resource-summary\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n",
)
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(changes=[{"change": "added", "path": "resource/2026-01-01/report.txt"}]),
)
@ -1540,7 +1832,7 @@ def test_auto_resource_update_finds_renamed_note_by_source_resource():
cwd / "daily" / "2026-01-01" / "generated-name.md",
"---\nname: generated-name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nold body\n",
)
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]),
)
@ -1583,7 +1875,7 @@ def test_auto_resource_update_keeps_existing_renamed_path():
)
wrapper.on_reply = rewrite_frontmatter
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]),
)
@ -1619,7 +1911,7 @@ def test_auto_resource_reports_unmodified_when_agent_skips_existing_note():
cwd / "daily" / "2026-01-01" / "report.md",
"---\nname: report\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n",
)
step = AutoResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs, agent_wrapper=wrapper)
resp = await step(
RuntimeContext(changes=[{"change": "modified", "path": "resource/2026-01-01/report.txt"}]),
)
@ -1635,8 +1927,8 @@ def test_auto_resource_reports_unmodified_when_agent_skips_existing_note():
asyncio.run(run())
def test_auto_resource_deletes_loose_root_resource_note_for_today():
"""Deleting a loose root resource deletes today's same-stem note."""
def test_auto_resource_preserves_unowned_loose_root_same_stem_note():
"""Deleting a loose resource never claims an unowned same-stem user note."""
async def run():
with tempfile.TemporaryDirectory() as tmpdir, temp_chdir(tmpdir):
@ -1648,18 +1940,21 @@ def test_auto_resource_deletes_loose_root_resource_note_for_today():
try:
today = datetime.datetime.now().strftime("%Y-%m-%d")
note_path = write_file(workspace / "daily" / today / "report.md", "---\nname: report\n---\nbody\n")
step = AutoResourceStep(app_context=app_ctx, file_store=fs)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs)
resp = await step(RuntimeContext(changes=[{"change": "deleted", "path": "resource/report.txt"}]))
assert resp.success is True
assert resp.metadata["results"][0]["path"] == "resource/report.txt"
assert resp.metadata["results"][0]["metadata"]["modified"] is True
assert resp.metadata["modified"] is True
assert not note_path.exists()
result_meta = resp.metadata["results"][0]["metadata"]
assert result_meta["action"] == "skipped"
assert result_meta["reason"] == "resource_note_not_found"
assert result_meta["modified"] is False
assert resp.metadata["modified"] is False
assert note_path.read_text(encoding="utf-8") == "---\nname: report\n---\nbody\n"
finally:
await fs.close()
print("✓ test_auto_resource_deletes_loose_root_resource_note_for_today passed")
print("✓ test_auto_resource_preserves_unowned_loose_root_same_stem_note passed")
asyncio.run(run())
@ -1679,7 +1974,7 @@ def test_auto_resource_deletes_renamed_note_by_source_resource():
workspace / "daily" / "2026-01-01" / "generated-name.md",
"---\nname: generated-name\nsource_resource: '[[resource/2026-01-01/report.txt]]'\n---\nbody\n",
)
step = AutoResourceStep(app_context=app_ctx, file_store=fs)
step = AutoTextResourceStep(app_context=app_ctx, file_store=fs)
resp = await step(
RuntimeContext(changes=[{"change": "deleted", "path": "resource/2026-01-01/report.txt"}]),
@ -1937,7 +2232,7 @@ if __name__ == "__main__":
test_auto_resource_update_keeps_existing_renamed_path()
test_auto_memory_uses_message_day_for_historical_create()
test_auto_memory_rejects_invalid_explicit_date_before_saving_session()
test_auto_resource_deletes_loose_root_resource_note_for_today()
test_auto_resource_preserves_unowned_loose_root_same_stem_note()
test_auto_resource_deletes_renamed_note_by_source_resource()
test_auto_resource_result_hook_is_optional_and_isolated()
# LogChangesStep