feat: add cron scheduling support and enhance Claude Code integration (#278)
Some checks failed
Tests ReMe / Unit Tests - py3.11 (push) Has been cancelled
Tests ReMe / Unit Tests - py3.12 (push) Has been cancelled
Tests ReMe / Unit Tests - py3.13 (push) Has been cancelled
Pre-commit / run (ubuntu-latest) (push) Has been cancelled

* feat: add cron scheduling support and enhance Claude Code integration

- Introduce CronStep for periodic job execution with support for cron
  expressions, daily schedules, and fixed intervals
- Add automatic session management to Claude Code agent wrapper with
  cache-friendly defaults for system prompts and setting sources
- Implement fork session support with proper validation
- Enhance auto-dream functionality to dispatch per-file jobs instead
  of direct method calls for better backend agnosticism
- Add session ID tracking to auto-resource operations
- Remove deprecated download step component
- Update auto-dream job naming from auto-dream to auto_dream
- Add croniter dependency and update package data to include markdown
  files

* feat: add CronJob component and rename cron step to cron job
This commit is contained in:
Sen Huang 2026-06-10 20:40:18 +08:00 committed by GitHub
parent c3fb825af0
commit f458566e2c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
22 changed files with 1594 additions and 1596 deletions

View file

@ -30,12 +30,33 @@ class CcAgentWrapper(BaseAgentWrapper):
for k, v in self.kwargs.items():
kwargs.setdefault(k, v)
session_id: str = kwargs.pop("session_id", "")
fork_session: bool = kwargs.pop("fork_session", False)
sp = kwargs.get("system_prompt")
if isinstance(sp, str):
kwargs["system_prompt"] = {
"type": "preset",
"preset": "claude_code",
"append": sp,
"exclude_dynamic_sections": True,
}
kwargs.setdefault("setting_sources", [])
opts = ClaudeAgentOptions()
skip_keys = {"tools", "output_schema"}
for k, v in kwargs.items():
if k not in skip_keys and hasattr(opts, k):
setattr(opts, k, v)
if session_id:
opts.resume = session_id
if fork_session:
opts.fork_session = True
elif fork_session:
# fork_session with no session_id to fork from is meaningless.
raise ValueError("fork_session=True requires a non-empty session_id")
tools: list["BaseJob"] = kwargs.get("tools", [])
if tools:
sdk_tools = [self._make_tool(job) for job in tools]
@ -57,5 +78,7 @@ class CcAgentWrapper(BaseAgentWrapper):
if last_msg is None:
raise ValueError("No message received from Claude Code.")
result = last_msg.structured_output if output_schema and last_msg.structured_output else last_msg
return last_msg.session_id or "", result
if output_schema:
structured = last_msg.structured_output or {}
return last_msg.session_id or "", {"message": last_msg, "structured_output": structured}
return last_msg.session_id or "", last_msg

View file

@ -2,10 +2,12 @@
from .background_job import BackgroundJob
from .base_job import BaseJob
from .cron_job import CronJob
from .stream_job import StreamJob
__all__ = [
"BackgroundJob",
"BaseJob",
"CronJob",
"StreamJob",
]

View file

@ -0,0 +1,212 @@
"""``cron`` — a supervised background job that periodically dispatches
downstream job(s) and/or step(s) on a schedule.
A ``CronJob`` is a :class:`BackgroundJob` whose body is a scheduler loop
rather than a fixed list of steps: it sleeps until the next fire time,
dispatches its configured downstream job(s) and/or step(s), and loops.
It exits when ``self._stop_event`` is set, so the surrounding
``Application`` shutdown / supervisor can stop it cleanly.
Declared directly under ``jobs:`` in YAML (no step wrapper)::
auto_dream_cron:
backend: cron
dispatch_job: auto_dream
cron: "0 3 * * *" # daily at 03:00; "0 */6 * * *" = every 6h
run_on_start: false
Two dispatch modes pick one (or both, executed in order: jobs first,
then steps):
* ``dispatch_job`` / ``dispatch_jobs`` invoke a **registered job** by
name through ``self.app_context.jobs``. Goes through the job's own
parameter validation and context construction, identical to a CLI /
MCP invocation. Prefer this when the periodic task already has a job
wrapping it (e.g. ``auto_dream``).
* ``dispatch_step`` / ``dispatch_steps`` instantiate a **registered
step** by name and invoke it directly. Use for steps that don't have a
job wrapper.
Three schedule modes (mutually exclusive exactly one must be set):
* ``cron: "M H DoM Mon DoW"`` standard 5-field cron expression in
``app_config.timezone``. Most flexible; use for non-daily cadence
(``"0 */6 * * *"`` = every 6 hours, ``"0 3 * * 1-5"`` = 3am on
weekdays, ``"*/15 * * * *"`` = every 15 minutes).
* ``daily_at: "HH:MM"`` fire once per day at this wall-clock time, in
``app_config.timezone``. Convenience shorthand for ``"M H * * *"``.
* ``interval_seconds: int`` fire every N seconds since launch. Useful
for tests and for time-zone-independent sub-minute cadence.
Per-dispatch exceptions are caught and logged a failed downstream job
or step never kills the cron loop (so the ``BackgroundJob`` supervisor is
reserved for genuine scheduler-loop crashes). ``run_on_start: true``
triggers one immediate dispatch on launch (default ``false``, to avoid a
burst when ``Application.start`` brings several cron jobs up at once).
Cron expressions are evaluated via ``croniter`` and validated eagerly at
construction time, so a typo fails at app start rather than at 3 a.m.
"""
import datetime
import zoneinfo
from croniter import croniter
from .background_job import BackgroundJob
from ..component_registry import R
from ...enumeration import ComponentEnum
from ...schema import Response
@R.register("cron")
class CronJob(BackgroundJob):
"""Dispatch downstream step(s) and/or job(s) on a schedule until ``stop_event`` fires."""
def __init__(
self,
dispatch_step: str = "",
dispatch_steps: list[str] | None = None,
dispatch_job: str = "",
dispatch_jobs: list[str] | None = None,
cron: str = "",
daily_at: str = "",
interval_seconds: int = 0,
run_on_start: bool = False,
**kwargs,
):
super().__init__(**kwargs)
self.dispatch_steps: list[str] = dispatch_steps or ([dispatch_step] if dispatch_step else [])
self.dispatch_jobs: list[str] = dispatch_jobs or ([dispatch_job] if dispatch_job else [])
self.cron: str = cron
self.daily_at: str = daily_at
self.interval_seconds: int = interval_seconds
self.run_on_start: bool = run_on_start
if not self.dispatch_steps and not self.dispatch_jobs:
raise ValueError(
"cron job requires at least one of 'dispatch_step'/'dispatch_steps' "
"or 'dispatch_job'/'dispatch_jobs'",
)
schedules_set = sum(bool(x) for x in (self.cron, self.daily_at, self.interval_seconds))
if schedules_set != 1:
raise ValueError(
"cron job requires exactly one of "
"'cron' (5-field expression), 'daily_at' (HH:MM), "
"or 'interval_seconds' (int)",
)
if self.daily_at:
# Validate HH:MM format eagerly so misconfig fails at start, not at 3am.
h, m = self._parse_hh_mm(self.daily_at)
self._fire_hour, self._fire_minute = h, m
elif self.cron:
# Fail at start, not at the next scheduled tick.
if not croniter.is_valid(self.cron):
raise ValueError(f"cron expression invalid, got {self.cron!r}")
@staticmethod
def _parse_hh_mm(value: str) -> tuple[int, int]:
try:
h_str, m_str = value.split(":", 1)
h, m = int(h_str), int(m_str)
except (ValueError, AttributeError) as exc:
raise ValueError(f"daily_at must be 'HH:MM', got {value!r}") from exc
if not (0 <= h < 24 and 0 <= m < 60):
raise ValueError(f"daily_at out of range, got {value!r}")
return h, m
def _tz(self) -> datetime.tzinfo | None:
if not self.app_context:
return None
tz_name = self.app_context.app_config.timezone
if not tz_name:
return None
try:
return zoneinfo.ZoneInfo(tz_name)
except zoneinfo.ZoneInfoNotFoundError:
self.logger.warning(f"[{self.name}] unknown timezone {tz_name!r}; using local time")
return None
def _next_fire_delay(self) -> float:
"""Seconds from now until the next fire — never negative, never zero."""
if self.interval_seconds:
return float(self.interval_seconds)
tz = self._tz()
now = datetime.datetime.now(tz)
if self.cron:
# croniter requires an explicit base time; tz comes through on `now`.
nxt = croniter(self.cron, now).get_next(datetime.datetime)
return (nxt - now).total_seconds()
# daily_at
target = now.replace(hour=self._fire_hour, minute=self._fire_minute, second=0, microsecond=0)
if target <= now:
target = target + datetime.timedelta(days=1)
return (target - now).total_seconds()
async def _fire(self, dispatch_classes: list[type]) -> None:
"""Dispatch each downstream job (via the registry) and step (class-level), in that
order; swallow exceptions per-dispatch so a single failure doesn't break the loop."""
# Jobs first — they're the higher-level invocation path (matches CLI / MCP).
for name in self.dispatch_jobs:
try:
job = self.app_context.jobs.get(name) if self.app_context else None
if job is None:
raise RuntimeError(f"Job {name!r} not found")
await job()
self.logger.info(f"[{self.name}] dispatched job {name!r}")
except Exception as exc:
self.logger.exception(f"[{self.name}] dispatch job {name!r} raised: {exc}")
for cls in dispatch_classes:
try:
s = cls(app_context=self.app_context)
await s()
self.logger.info(f"[{self.name}] dispatched step {cls.__name__}")
except Exception as exc:
self.logger.exception(f"[{self.name}] dispatch step {cls.__name__} raised: {exc}")
async def __call__(self, **kwargs) -> Response:
"""Scheduler body: loop dispatching downstream jobs/steps until ``stop_event`` fires.
Per-dispatch exceptions are swallowed in ``_fire``, so this body only propagates
genuine scheduler-loop failures to the ``BackgroundJob`` supervisor for restart.
"""
assert self._stop_event is not None
stop_event = self._stop_event
dispatch_classes: list[type] = []
for name in self.dispatch_steps:
cls = R.get(ComponentEnum.STEP, name)
if cls is None:
raise RuntimeError(f"Unregistered step '{name}'")
dispatch_classes.append(cls)
if self.cron:
mode = f"cron={self.cron!r}"
elif self.daily_at:
mode = f"daily_at={self.daily_at}"
else:
mode = f"interval_seconds={self.interval_seconds}"
self.logger.info(
f"[{self.name}] cron loop start "
f"dispatch_jobs={self.dispatch_jobs} dispatch_steps={self.dispatch_steps} "
f"{mode} run_on_start={self.run_on_start}",
)
if self.run_on_start and not stop_event.is_set():
await self._fire(dispatch_classes)
while not stop_event.is_set():
delay = self._next_fire_delay()
self.logger.info(f"[{self.name}] next fire in {delay:.0f}s")
await self._wait_or_stop(delay)
if stop_event.is_set():
break
await self._fire(dispatch_classes)
response = Response()
response.success = True
response.answer = f"cron job jobs={self.dispatch_jobs!r} steps={self.dispatch_steps!r} stopped"
return response

View file

@ -48,6 +48,12 @@ jobs:
- backend: watch_changes_step
dispatch_step: log_changes_step
# auto_dream_cron:
# backend: cron
# dispatch_job: auto_dream
# cron: "0 3 * * *" # daily at 03:00; "0 */6 * * *" = every 6h
# run_on_start: false
version:
backend: base
description: "return reme package version"
@ -436,7 +442,7 @@ jobs:
steps:
- backend: dream_step
auto-dream:
auto_dream:
backend: base
description: "Auto-dream: scan today's day-index <daily_dir>/<today>.md and session notes under <daily_dir>/<today>/*.md — run dream on each (Phase 1 extract+classify, Phase 2 per-bucket integrate)."
parameters:
@ -471,10 +477,6 @@ jobs:
memory_hint:
type: string
description: "optional hint"
timezone:
type: string
description: "IANA timezone, e.g. Asia/Shanghai"
default: "Asia/Shanghai"
required:
- messages
steps:

View file

@ -21,6 +21,7 @@ classifiers = [
dependencies = [
"aiofiles>=24.1.0",
"croniter>=2.0",
"fastapi>=0.135.1",
"fastmcp>=3.1.0",
"httpx>=0.28.1",

View file

@ -37,9 +37,6 @@ from .index.update_index import UpdateIndexStep
from .index.foreach_dispatch import ForeachDispatchStep
from .index.log_changes import LogChangesStep
from .index.watch_changes import WatchChangesStep
from .transfer.download import DownloadStep
from .transfer.ingest import IngestStep
from .transfer.upload import UploadStep
__all__ = [
"BaseStep",
@ -90,8 +87,4 @@ __all__ = [
# evolve (dream)
"AutoDreamStep",
"DreamStep",
# transfer
"DownloadStep",
"IngestStep",
"UploadStep",
]

View file

@ -1,4 +1,4 @@
"""AutoDreamStep — daily-tick wrapper around :class:`DreamStep`.
"""AutoDreamStep — daily-tick wrapper that dispatches per-file to the ``dream`` job.
Each tick scans today's two surfaces under ``<daily_dir>/``:
@ -17,6 +17,11 @@ prefix so we never disturb entries from other days), then:
* ``indexed`` keys not in ``existing`` **deleted**, drop from catalog
* mtime match **unchanged**, skip
For every to-dream file, the step calls the configured ``dispatch_job``
(default ``"dream"``) via :meth:`BaseStep.run_job`. The dispatch job's
``Response`` carries a ``DreamResult`` payload in ``metadata``;
AutoDream re-hydrates that for its per-file aggregate.
After dreaming, successful (and Phase 1 vacuously-skipped) files
upsert their current ``st_mtime`` so the next tick re-dreams only
what actually changed. Failures leave the catalog untouched and will
@ -32,17 +37,24 @@ work — invoke it from a system cron, ``reme auto-dream date=...``,
or any other catch-up trigger when ``auto_dream_loop`` missed a file
(e.g. process crashed before the watcher fired).
**Backend-agnostic.** Because dispatch goes through the configured
``dream`` job, the per-file dream implementation is decided by the
YAML config whichever step the ``dream`` job's ``backend`` resolves
to. AutoDream itself doesn't care which backend runs underneath.
Inputs (RuntimeContext):
date (str, optional): YYYY-MM-DD to scan. Defaults to today
in the dreamer's timezone.
hint (str, optional): passed through to each per-file dream.
Step kwargs (from yaml ``backend: auto_dream_step``):
dispatch_job (str, default "dream"): name of the job to call
per file. Override only if the deployment renamed the dream
job. The dispatched job must accept ``path`` and ``hint``
kwargs and return a ``Response`` whose ``metadata`` matches
:class:`DreamResult`.
persist (bool, default True): when True, ``file_catalog.dump()``
is called after the batch so progress survives a restart.
The outer loop reuses ``DreamStep``'s prompt mounting via MRO — no
separate YAML; ``dream.yaml`` is found through the parent class.
"""
from pathlib import Path
@ -50,7 +62,8 @@ from pathlib import Path
from pydantic import BaseModel, Field
from ._evolve import now
from .dream import DreamStep, DreamResult
from .dream import DreamResult
from ..base_step import BaseStep
from ...components import R
from ...schema import FileNode
@ -70,14 +83,57 @@ class AutoDreamResult(BaseModel):
@R.register("auto_dream_step")
class AutoDreamStep(DreamStep):
"""Scan ``daily/<today>.md`` + ``daily/<today>/`` and dream each file
whose ``st_mtime`` doesn't already match its ``file_catalog`` entry."""
class AutoDreamStep(BaseStep):
"""Scan ``daily/<today>.md`` + ``daily/<today>/`` and dispatch to the
configured ``dream`` job for each file whose ``st_mtime`` doesn't
already match its ``file_catalog`` entry."""
def __init__(self, persist: bool = True, **kwargs):
def __init__(self, dispatch_job: str = "dream", persist: bool = True, **kwargs):
super().__init__(**kwargs)
self.dispatch_job: str = dispatch_job
self.persist: bool = persist
def _vault_dir(self) -> Path:
"""Vault root as an absolute path (mirrors :meth:`DreamStep._vault_dir`)."""
vr = getattr(self.file_store, "vault_path", None)
return Path(vr).resolve() if vr else Path.cwd().resolve()
async def _dispatch_dream(self, rel_path: str, hint: str) -> DreamResult:
"""Call the configured ``dispatch_job`` once and re-hydrate its
``Response.metadata`` into a :class:`DreamResult`.
On dispatch failure (job raises) returns a ``DreamResult`` with
``error`` populated so the caller's accounting stays uniform.
"""
try:
resp = await self.run_job(self.dispatch_job, path=rel_path, hint=hint)
except Exception as e: # pylint: disable=broad-except
self.logger.error(
f"[{self.name}] dispatch {self.dispatch_job!r} failed on {rel_path}: " f"{type(e).__name__}: {e}",
)
return DreamResult(path=rel_path, error=f"{type(e).__name__}: {e}")
# The dream job's execute() does context.response.metadata.update(result.model_dump()),
# so metadata carries every DreamResult field. extra keys are
# ignored by pydantic v2 default (extra='ignore').
md = dict(resp.metadata or {})
try:
dr = DreamResult.model_validate(md)
except Exception as e: # noqa: BLE001
self.logger.error(
f"[{self.name}] dispatch {self.dispatch_job!r} returned non-DreamResult metadata "
f"for {rel_path}: {type(e).__name__}: {e}",
)
return DreamResult(path=rel_path, error=f"bad dispatch metadata: {type(e).__name__}: {e}")
# If the underlying step set success=False, treat as failure even
# if metadata didn't carry an error string (defensive).
if not resp.success and not dr.error:
dr.error = resp.answer or "dispatch returned success=False"
if not dr.path:
dr.path = rel_path
return dr
async def execute(self):
assert self.context is not None
date_input: str = (self.context.get("date", "") or "").strip()
@ -127,7 +183,7 @@ class AutoDreamStep(DreamStep):
self.logger.info(
f"[{self.name}] auto-dream tick date={today} scanned={len(existing)} "
f"unchanged={len(unchanged)} todo={len(to_dream)} deleted={len(to_delete)} "
f"under {daily_dir}/{today}{{.md,/}}",
f"under {daily_dir}/{today}{{.md,/}} dispatch={self.dispatch_job!r}",
)
# Drop catalog entries for files no longer on disk first. Cheap, no
@ -141,21 +197,13 @@ class AutoDreamStep(DreamStep):
f"[{self.name}] file_catalog.delete failed: {type(e).__name__}: {e}",
)
# Dream + upsert per-file. Single-file granularity means an LLM
# Dispatch + upsert per-file. Single-file granularity means a job
# failure on file N doesn't block files N+1..K from advancing their
# catalog mtime.
# catalog mtime. The dispatch decouples backend choice (AS / CC)
# from this loop — the configured ``dream`` job picks the runner.
upsert_nodes: list[FileNode] = []
for rel_path, mtime in to_dream:
try:
dr = await self.dream_one(rel_path, hint)
except Exception as e: # pylint: disable=broad-except
self.logger.error(
f"[{self.name}] dream_one failed on {rel_path}: {type(e).__name__}: {e}",
)
dr = DreamResult(
path=rel_path,
error=f"{type(e).__name__}: {e}",
)
dr = await self._dispatch_dream(rel_path, hint)
result.per_file.append(dr)
if dr.error:
# Failures leave the catalog untouched — next tick retries.

View file

@ -60,7 +60,7 @@ class AutoResourceStep(BaseStep):
self.context.response.success = True
self.context.response.answer = f"Deleted resource note: {note_rel}"
self.context.response.metadata.update({"path": note_rel, "action": "deleted"})
self.context.response.metadata.update({"path": note_rel, "session_id": session_id, "action": "deleted"})
async def _handle_upsert(self, file_path: str, date_str: str, session_id: str, created: bool) -> None:
create_response = await self.run_job("daily_create", session_id=session_id, date=date_str)
@ -97,12 +97,18 @@ class AutoResourceStep(BaseStep):
user_message,
system_prompt=self.prompt_format("system_prompt"),
tools=tools,
session_id=session_id,
)
self.context.response.success = True
self.context.response.answer = (msg.get_text_content() or "").strip()
self.context.response.metadata.update(
{"path": note_path, "created": note_created, "action": "added" if created else "modified"},
{
"path": note_path,
"created": note_created,
"session_id": session_id,
"action": "added" if created else "modified",
},
)
self.logger.info(f"[{self.name}] done {note_path}")

View file

@ -115,7 +115,8 @@ class WatchChangesStep(BaseStep):
]
if changes:
self.logger.info(f"Detected {len(changes)} change(s)")
extra = {k: v for k, v in self.context.data.items() if k not in ("stop_event",)}
# TODO @jinli
extra = {k: v for k, v in self.context.data.items() if k not in ("stop_event", "changes")}
for cls in dispatch_classes:
s = cls(app_context=self.app_context)
await s(changes=changes, **extra)

View file

@ -1,287 +0,0 @@
"""Fixture for the dreamer integration tests.
Seeds a vault with:
- 4 pre-existing digest/ nodes spread across the three buckets
(procedure / personal / wiki) these are the **recall targets**;
the new material partially overlaps them so Phase 2 must find
them via search + read and decide UPDATE.
- 4 small daily/ stubs the digest nodes already link to they exist
only so the seeded digest bodies don't dangle.
- 1 NEW daily note (the file the dreamer will be invoked on).
It exercises CREATE and UPDATE across the three buckets:
* wiki : UPDATE digest/wiki/jwt.md (24h rotation cadence
refines short-credential-compliance framing) +
CREATE digest/wiki/kid-versioning.md +
CREATE digest/wiki/soc2-30day-finding.md
(the OAuth2 restatement section is intentionally
a non-abstraction Phase 1 should NOT emit a
sub-unit for it; tests Phase 1's gate-keeping)
* procedure : UPDATE digest/procedure/key-rotation.md (24h
cadence + kid-versioning supersede the 30-day
JWKS-cache flow)
* personal : UPDATE digest/personal/no-trailing-summary.md
(extend "no trailing summary" to also forbid
"next steps" lists) + CREATE
digest/personal/small-pr.md
Total budget per integration run: 1 Phase 1 + up-to-6 Phase 2 = up
to 7 ReAct sessions, each with several tool turns (search +
traverse frontmatter_read + read write / edit).
Idempotent: re-running does NOT overwrite existing files. To re-seed
from scratch, delete the vault and rerun.
Usage as a script:
python tests4/integration/_dreamer_fixture.py /tmp/my-vault
Usage as a module:
from _dreamer_fixture import clean_vault, seed_vault, INPUT_PATH
clean_vault(Path("/tmp/my-vault"))
seed_vault(Path("/tmp/my-vault"))
"""
import shutil
import sys
from pathlib import Path
INPUT_PATH = "daily/2026-05-28/auth-refactor/notes.md"
_FILES: dict[str, str] = {
# ----- pre-existing digest nodes (recall targets) -----
"digest/wiki/jwt.md": """\
---
name: jwt
description: JSON Web Token signed authentication token format
---
# JWT
JSON Web Token (RFC 7519). A compact, signed (JWS) or encrypted (JWE)
token used to assert identity and claims between parties.
## Structure
- Header `alg`, `typ`, `kid`
- Payload claims: `iss`, `sub`, `aud`, `exp`, `iat`
- Signature
## Related
Often issued by [[digest/wiki/oauth2.md]] flows.
derived_from:: [[daily/2026-05-15/auth-design/notes.md]]
""",
"digest/wiki/oauth2.md": """\
---
name: oauth2
description: OAuth 2.0 delegated authorization framework
---
# OAuth 2.0
RFC 6749. A delegated authorization framework: a resource owner grants
a client limited access to a protected resource via an access token
issued by an authorization server.
## Grant types
- Authorization code (with PKCE for public clients)
- Client credentials
- Refresh token
derived_from:: [[daily/2026-05-10/oauth-intro/notes.md]]
""",
"digest/procedure/key-rotation.md": """\
---
name: key-rotation
description: Rotating signing keys for JWT issuance
---
# Key rotation (current — pre 2026-05-28 refactor)
Procedure for rotating the signing key used by [[digest/wiki/jwt.md]]
issuance.
## Steps
1. Generate new keypair offline.
2. Publish the public key to the JWKS endpoint with a fresh `kid`.
3. Wait 24h for clients to refresh their JWKS cache.
4. Cut over the signer to the new private key.
5. Mark the old `kid` as deprecated; remove after 30 days.
## Cadence
Default rotation cadence is **30 days**. Driven by historical practice;
no formal compliance requirement has tightened this so far.
derived_from:: [[daily/2026-05-20/rotation-plan/notes.md]]
""",
"digest/personal/no-trailing-summary.md": """\
---
name: no-trailing-summary
description: 不要在回复末尾加总结段落
---
# 不要在回复末尾加总结段落
用户能看 diff,不需要在回复末尾重述刚做的事
**Why**: diff 已经把"改了什么"摆在用户面前;再口述一遍是噪音
**How to apply**: 任意编码 / 编辑任务回复结束时,直接停在最后一条
有信息量的话上,不要再补一段"以上就是本次的修改..."
derived_from:: [[daily/2026-05-01/style-feedback/notes.md]]
""",
# ----- daily provenance stubs (so the digest links don't dangle) -----
"daily/2026-05-01/style-feedback/notes.md": """\
---
name: notes
description: style feedback to Claude on 2026-05-01
---
# Style feedback (2026-05-01)
每次任务结束都重述了一遍刚做的事不需要,我能看 diff以后直接停
""",
"daily/2026-05-10/oauth-intro/notes.md": """\
---
name: notes
description: OAuth 2.0 intro session
---
# OAuth 2.0 intro
简介 grant types: authorization code (with PKCE), client credentials,
refresh token重点放在 PKCE 是给 public clients 用的
""",
"daily/2026-05-15/auth-design/notes.md": """\
---
name: notes
description: initial auth design discussion
---
# Auth design
讨论 JWT 的结构 (header / payload / signature) 和我们项目里的 claim
约定 (iss, sub, aud, exp, iat)
""",
"daily/2026-05-20/rotation-plan/notes.md": """\
---
name: notes
description: key rotation plan v1
---
# Key rotation plan v1
定下当前的 5 步轮换流程:offline 生成 keypair 发布到 JWKS ( kid)
24h cache 切签发 30 天后清旧 kid周期定 30
""",
# ----- the NEW daily note dreamer will be invoked on -----
INPUT_PATH: """\
---
name: notes
description: auth refactor working notes 2026-05-28
---
# Auth refactor — 2026-05-28
## 决定:JWT 轮换周期改为 24 小时
今天确定把 JWT 签名密钥的轮换周期从 30 天压到 **24 小时**原因是
SOC2 合规审计批评:30 天的会话 token 太长,不满足"短期凭证"原则
新流程不再依赖 JWKS cache 24h 等待,改成走 Redis 里的 `kid`
版本号实时下发客户端在 token 验证失败时主动拉新 JWKS,而不是定
时轮询
(这条同时更新 JWT 概念笔记和 key-rotation 流程笔记)
## 新概念:kid 版本号机制
`kid` (key ID) JWT header 里的字段我们把它当成版本号来用:
Redis key `auth:jwks:current_kid` 保存当前活跃 kid;Auth Service
在签发 token 时读这个 key,客户端验证失败时也读这个 key 再拉对应
public key这样无须等 cache TTL
## 顺带复习:OAuth 2.0 是什么
(为了帮新同学接住上下文,这里把 OAuth 2.0 简单重述一下,不引入
新事实)OAuth 2.0 (RFC 6749) 是一个委托授权框架:资源所有者
允许 client 通过 authorization server 颁发的 access token 来有
限度地访问受保护资源常见 grant types: authorization code
(public client PKCE)client credentialsrefresh token
这一段没有任何新内容,纯粹是给后面 JWT 24h 轮换决定铺垫读者
的背景知识
## 观察:SOC2 审计在 30 天周期上的具体批评
审计员引用 SOC2 CC6.1 控制点:"会话凭证应有合理的短期有效期"
30 天对应于人类工作周期,但对自动化客户端 token 来说过长审计
要求 24h 或更短,且必须能在事件响应时立即吊销 (kid 切换可满足)
## 偏好:小 PR 优先
后续这个 refactor PR ,每个 PR 控制在 < 300 原因是 review
负担太大时容易被拍脑袋通过,这违背了 SOC2 审计中变更管理的精神
## 偏好:回复结尾再补充
之前说过不要总结段落 (我能看 diff),今天再补充一点:也不要"接下来
的步骤"列表,除非我明确问 next steps。直接回答问题然后停。
""",
}
_CLEAN_DIRS = ("daily", "digest", "reme_metadata")
def clean_vault(vault: Path) -> list[str]:
"""Remove fixture-managed subdirs (`daily/`, `digest/`, `reme_metadata/`)
under `vault` so the next `seed_vault` starts from a clean slate.
Returns the relative paths that were actually removed."""
removed: list[str] = []
for rel in _CLEAN_DIRS:
target = vault / rel
if target.exists():
shutil.rmtree(target)
removed.append(rel)
return removed
def seed_vault(vault: Path) -> list[str]:
"""Write any missing fixture files under `vault`. Return relative paths
that were actually written (skipped existing ones)."""
seeded: list[str] = []
for rel, body in _FILES.items():
target = vault / rel
if target.exists():
continue
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(body, encoding="utf-8")
seeded.append(rel)
return seeded
# pylint: disable=missing-function-docstring
def main() -> None:
if len(sys.argv) < 2:
print(f"usage: {sys.argv[0]} <vault_dir>", file=sys.stderr)
sys.exit(2)
vault = Path(sys.argv[1]).resolve()
vault.mkdir(parents=True, exist_ok=True)
removed = clean_vault(vault)
if removed:
print(f"cleaned {len(removed)} dir(s) under {vault}: {', '.join(removed)}")
seeded = seed_vault(vault)
if seeded:
print(f"seeded {len(seeded)} file(s) under {vault}:")
for f in seeded:
print(f" + {f}")
else:
print(f"vault {vault} already seeded — no changes")
print(f"\nDream this file:\n {vault}/{INPUT_PATH}")
if __name__ == "__main__":
main()

View file

@ -0,0 +1,652 @@
"""Shared integration-test fixture: build a vault test environment.
Every integration test in this directory used to repeat the same
boilerplate temp dir, ``chdir``, ``load_env()``, ``_make_app()``,
``_today()``, ``_AgentMemoryRecorder``, ``_wait_for_glob`` /
``_wait_for_populated`` / ``_wait_for_server``, ad-hoc seed helpers
(daily notes, resource files, the dreamer's pre-existing digest nodes).
This module unifies all of that behind one entry point:
from _vault_fixture import vault_env
async def run():
with vault_env() as env:
app = await env.make_app()
try:
# env.vault_dir, env.today, env.place_resource(...),
# env.seed_daily_note(...), env.seed_dream_vault(),
# env.wait_for_populated(...), env.record_agents(...) ...
...
finally:
await env.close_all()
``env.make_app()`` defaults to the standard config; pass
``config="cc"`` for the CC SDK wiring, or arbitrary kwargs to deep-merge
into ``resolve_app_config``. The vault path is fixed to
``<tmp_workspace>/.reme`` so seed helpers can write files before the
app is started.
The dreamer-specific seed (4 pre-existing digest nodes spread across
the three buckets + 4 daily provenance stubs + a new daily note that
exercises CREATE and UPDATE in each bucket) is preserved as
``env.seed_dream_vault()`` / ``DREAM_INPUT_PATH``.
Usage as a script (for the dreamer manual run):
python tests4/integration/_vault_fixture.py /tmp/my-vault
"""
from __future__ import annotations
import asyncio
import contextlib
import json
import os
import shutil
import socket
import sys
import tempfile
import time
from datetime import date as _date
from pathlib import Path
from typing import Any, Iterator
REPO_ROOT = Path(__file__).resolve().parents[2]
INTEGRATION_DIR = Path(__file__).resolve().parent
# ─────────────────────────────────────────────────────────────────────────────
# Dream preset — pre-existing digest nodes + a new daily note that exercises
# CREATE and UPDATE across the three buckets (procedure / personal / wiki).
# ─────────────────────────────────────────────────────────────────────────────
DREAM_INPUT_PATH = "daily/2026-05-28/auth-refactor/notes.md"
_DREAM_FILES: dict[str, str] = {
# ----- pre-existing digest nodes (recall targets) -----
"digest/wiki/jwt.md": """\
---
name: jwt
description: JSON Web Token signed authentication token format
---
# JWT
JSON Web Token (RFC 7519). A compact, signed (JWS) or encrypted (JWE)
token used to assert identity and claims between parties.
## Structure
- Header `alg`, `typ`, `kid`
- Payload claims: `iss`, `sub`, `aud`, `exp`, `iat`
- Signature
## Related
Often issued by [[digest/wiki/oauth2.md]] flows.
derived_from:: [[daily/2026-05-15/auth-design/notes.md]]
""",
"digest/wiki/oauth2.md": """\
---
name: oauth2
description: OAuth 2.0 delegated authorization framework
---
# OAuth 2.0
RFC 6749. A delegated authorization framework: a resource owner grants
a client limited access to a protected resource via an access token
issued by an authorization server.
## Grant types
- Authorization code (with PKCE for public clients)
- Client credentials
- Refresh token
derived_from:: [[daily/2026-05-10/oauth-intro/notes.md]]
""",
"digest/procedure/key-rotation.md": """\
---
name: key-rotation
description: Rotating signing keys for JWT issuance
---
# Key rotation (current — pre 2026-05-28 refactor)
Procedure for rotating the signing key used by [[digest/wiki/jwt.md]]
issuance.
## Steps
1. Generate new keypair offline.
2. Publish the public key to the JWKS endpoint with a fresh `kid`.
3. Wait 24h for clients to refresh their JWKS cache.
4. Cut over the signer to the new private key.
5. Mark the old `kid` as deprecated; remove after 30 days.
## Cadence
Default rotation cadence is **30 days**. Driven by historical practice;
no formal compliance requirement has tightened this so far.
derived_from:: [[daily/2026-05-20/rotation-plan/notes.md]]
""",
"digest/personal/no-trailing-summary.md": """\
---
name: no-trailing-summary
description: 不要在回复末尾加总结段落
---
# 不要在回复末尾加总结段落
用户能看 diff,不需要在回复末尾重述刚做的事
**Why**: diff 已经把"改了什么"摆在用户面前;再口述一遍是噪音
**How to apply**: 任意编码 / 编辑任务回复结束时,直接停在最后一条
有信息量的话上,不要再补一段"以上就是本次的修改..."
derived_from:: [[daily/2026-05-01/style-feedback/notes.md]]
""",
# ----- daily provenance stubs (so the digest links don't dangle) -----
"daily/2026-05-01/style-feedback/notes.md": """\
---
name: notes
description: style feedback to Claude on 2026-05-01
---
# Style feedback (2026-05-01)
每次任务结束都重述了一遍刚做的事不需要,我能看 diff以后直接停
""",
"daily/2026-05-10/oauth-intro/notes.md": """\
---
name: notes
description: OAuth 2.0 intro session
---
# OAuth 2.0 intro
简介 grant types: authorization code (with PKCE), client credentials,
refresh token重点放在 PKCE 是给 public clients 用的
""",
"daily/2026-05-15/auth-design/notes.md": """\
---
name: notes
description: initial auth design discussion
---
# Auth design
讨论 JWT 的结构 (header / payload / signature) 和我们项目里的 claim
约定 (iss, sub, aud, exp, iat)
""",
"daily/2026-05-20/rotation-plan/notes.md": """\
---
name: notes
description: key rotation plan v1
---
# Key rotation plan v1
定下当前的 5 步轮换流程:offline 生成 keypair 发布到 JWKS ( kid)
24h cache 切签发 30 天后清旧 kid周期定 30
""",
# ----- the NEW daily note dreamer will be invoked on -----
DREAM_INPUT_PATH: """\
---
name: notes
description: auth refactor working notes 2026-05-28
---
# Auth refactor — 2026-05-28
## 决定:JWT 轮换周期改为 24 小时
今天确定把 JWT 签名密钥的轮换周期从 30 天压到 **24 小时**原因是
SOC2 合规审计批评:30 天的会话 token 太长,不满足"短期凭证"原则
新流程不再依赖 JWKS cache 24h 等待,改成走 Redis 里的 `kid`
版本号实时下发客户端在 token 验证失败时主动拉新 JWKS,而不是定
时轮询
(这条同时更新 JWT 概念笔记和 key-rotation 流程笔记)
## 新概念:kid 版本号机制
`kid` (key ID) JWT header 里的字段我们把它当成版本号来用:
Redis key `auth:jwks:current_kid` 保存当前活跃 kid;Auth Service
在签发 token 时读这个 key,客户端验证失败时也读这个 key 再拉对应
public key这样无须等 cache TTL
## 顺带复习:OAuth 2.0 是什么
(为了帮新同学接住上下文,这里把 OAuth 2.0 简单重述一下,不引入
新事实)OAuth 2.0 (RFC 6749) 是一个委托授权框架:资源所有者
允许 client 通过 authorization server 颁发的 access token 来有
限度地访问受保护资源常见 grant types: authorization code
(public client PKCE)client credentialsrefresh token
这一段没有任何新内容,纯粹是给后面 JWT 24h 轮换决定铺垫读者
的背景知识
## 观察:SOC2 审计在 30 天周期上的具体批评
审计员引用 SOC2 CC6.1 控制点:"会话凭证应有合理的短期有效期"
30 天对应于人类工作周期,但对自动化客户端 token 来说过长审计
要求 24h 或更短,且必须能在事件响应时立即吊销 (kid 切换可满足)
## 偏好:小 PR 优先
后续这个 refactor PR ,每个 PR 控制在 < 300 原因是 review
负担太大时容易被拍脑袋通过,这违背了 SOC2 审计中变更管理的精神
## 偏好:回复结尾再补充
之前说过不要总结段落 (我能看 diff),今天再补充一点:也不要"接下来
的步骤"列表,除非我明确问 next steps。直接回答问题然后停。
""",
}
_CLEAN_DIRS = ("daily", "digest", "resource", "reme_metadata")
# ─────────────────────────────────────────────────────────────────────────────
# Small primitives
# ─────────────────────────────────────────────────────────────────────────────
def today() -> str:
"""ISO date today, e.g. ``2026-06-08`` — same shape every test expected."""
return _date.today().isoformat()
@contextlib.contextmanager
def temp_chdir(path) -> Iterator[Path]:
"""``chdir`` to ``path`` for the block; restore the original cwd on exit."""
old = os.getcwd()
os.chdir(path)
try:
yield Path(path)
finally:
os.chdir(old)
def port_free(host: str, port: int) -> bool:
"""True iff (host, port) is currently bindable. Used by webhook tests
to fail fast instead of racing the connector start-up against a
listener that's already squatting the port."""
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
try:
sock.bind((host, port))
except OSError:
return False
return True
# ─────────────────────────────────────────────────────────────────────────────
# Agent transcript capture — monkey-patches ``Agent.__init__`` to grab every
# agent created within its ``with`` block, then dumps each agent's context
# to ``<dump_dir>/<prefix>_<idx>_<name>.jsonl`` on ``dump()``.
# ─────────────────────────────────────────────────────────────────────────────
class AgentMemoryRecorder:
"""Record every ``agentscope.agent.Agent`` instance created within the block.
Used to surface the ReAct trace of Phase 1 / Phase 2 dreams, the
daily-write fork, the resource-interpret agent, etc. what tools
were called in what order, what candidates were recalled, what the
LLM decided. Dumps land under ``<vault>/agent_logs/`` by default
(created by :meth:`VaultEnv.record_agents`) so they're auto-cleaned
with the throwaway vault. Pass an explicit ``dump_dir`` to persist
them somewhere else for post-mortem inspection.
"""
def __init__(self, dump_dir: Path, prefix: str = "agent"):
self.dump_dir = dump_dir
self.prefix = prefix
self.agents: list[Any] = []
self._orig_init = None
self.dumped_paths: list[Path] = []
def __enter__(self):
from agentscope.agent import Agent # local import — heavy module
self._orig_init = Agent.__init__
agents = self.agents
orig = self._orig_init
def _capturing_init(agent_self, *args, **kwargs):
orig(agent_self, *args, **kwargs)
agents.append(agent_self)
Agent.__init__ = _capturing_init
return self
def __exit__(self, *exc):
from agentscope.agent import Agent
if self._orig_init is not None:
Agent.__init__ = self._orig_init
async def dump(self) -> list[Path]:
"""Serialize captured agent transcripts to ``<dump_dir>/`` and return paths."""
self.dump_dir.mkdir(parents=True, exist_ok=True)
for stale in self.dump_dir.glob(f"{self.prefix}_*.jsonl"):
stale.unlink()
for idx, agent in enumerate(self.agents, 1):
messages = agent.state.context
name = getattr(agent, "name", "agent") or "agent"
out_path = self.dump_dir / f"{self.prefix}_{idx:02d}_{name}.jsonl"
with out_path.open("w", encoding="utf-8") as f:
for msg in messages:
f.write(json.dumps(msg.model_dump(), ensure_ascii=False, default=str) + "\n")
self.dumped_paths.append(out_path)
return self.dumped_paths
# ─────────────────────────────────────────────────────────────────────────────
# VaultEnv — the value the ``vault_env()`` context manager yields. Holds the
# resolved vault path, app construction, seeding, wait helpers, recorder
# factory, and tracks any apps the test started so they get closed.
# ─────────────────────────────────────────────────────────────────────────────
class VaultEnv:
"""A vault test environment — temp workspace + helpers."""
def __init__(self, workspace: Path, vault_dir: Path):
self.workspace = workspace
self.vault_dir = vault_dir
self.today = today()
self._apps: list[Any] = []
# ----- app construction -----------------------------------------------
async def make_app(self, *, config: str | None = None, **overrides) -> Any:
"""Build and start an ``Application`` (default config) and track it for cleanup.
Pass ``config="cc"`` for the CC SDK wiring; arbitrary kwargs are
deep-merged into ``resolve_app_config`` (e.g. ``jobs={...}`` to
inject a background connector).
"""
from reme4 import Application
from reme4.config import resolve_app_config
kwargs: dict[str, Any] = {
"log_to_console": False,
"log_to_file": False,
"enable_logo": False,
"vault_dir": str(self.vault_dir),
}
if config:
kwargs["config"] = config
kwargs.update(overrides)
cfg = resolve_app_config(**kwargs)
app = Application(**cfg)
await app.start()
self._apps.append(app)
return app
async def make_reme(self, **overrides) -> Any:
"""Same as ``make_app`` but returns a ``ReMe`` instance (alias of ``Application``)."""
from reme4 import ReMe
from reme4.config import resolve_app_config
kwargs: dict[str, Any] = {"vault_dir": str(self.vault_dir)}
kwargs.update(overrides)
cfg = resolve_app_config(**kwargs)
app = ReMe(**cfg)
await app.start()
self._apps.append(app)
return app
async def close_all(self) -> None:
"""Close every app started via this env. Idempotent."""
for app in self._apps:
await app.close()
self._apps.clear()
# ----- seeding --------------------------------------------------------
def clean(self) -> list[str]:
"""Remove fixture-managed subdirs (``daily/``, ``digest/``, ``resource/``,
``reme_metadata/``) under the vault so the next seed starts clean.
Returns relative paths that were actually removed."""
removed: list[str] = []
for rel in _CLEAN_DIRS:
target = self.vault_dir / rel
if target.exists():
shutil.rmtree(target)
removed.append(rel)
return removed
def seed_dream_vault(self) -> list[str]:
"""Write the dreamer preset: pre-existing digest nodes + provenance
stubs + the new daily note dreamer will be invoked on
(``DREAM_INPUT_PATH``). Idempotent skips files that already exist.
Returns relative paths that were actually written."""
seeded: list[str] = []
for rel, body in _DREAM_FILES.items():
target = self.vault_dir / rel
if target.exists():
continue
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(body, encoding="utf-8")
seeded.append(rel)
return seeded
def place_resource(
self,
filename: str,
content: str,
*,
date: str | None = None,
) -> str:
"""Drop a file under ``resource/<date>/<filename>``. Returns the
vault-relative path so the caller can pass it straight to
``auto_resource``."""
d = date or self.today
resource_dir = self.vault_dir / "resource" / d
resource_dir.mkdir(parents=True, exist_ok=True)
path = resource_dir / filename
path.write_text(content, encoding="utf-8")
return f"resource/{d}/{filename}"
def seed_daily_note(
self,
stem: str,
body: str,
*,
date: str | None = None,
) -> Path:
"""Write a note at ``daily/<date>/<stem>.md`` and return the absolute path."""
d = date or self.today
day_dir = self.vault_dir / "daily" / d
day_dir.mkdir(parents=True, exist_ok=True)
path = day_dir / f"{stem}.md"
path.write_text(body, encoding="utf-8")
return path
# ----- introspection --------------------------------------------------
def daily_notes(self, *, date: str | None = None) -> list[Path]:
"""All ``.md`` files under ``daily/<date>/`` (sorted)."""
d = date or self.today
day_dir = self.vault_dir / "daily" / d
if not day_dir.is_dir():
return []
return sorted(day_dir.glob("*.md"))
def digest_files(self) -> list[Path]:
"""All ``.md`` files anywhere under ``digest/`` (sorted)."""
digest_root = self.vault_dir / "digest"
if not digest_root.is_dir():
return []
return sorted(digest_root.rglob("*.md"))
def session_state_files(self, prefix: str = "session_state_") -> list[Path]:
"""All session-state jsonl files (the agent wrapper writes these under
``resource/`` whenever a ``session_id`` is provided)."""
resource_dir = self.vault_dir / "resource"
if not resource_dir.exists():
return []
return sorted(resource_dir.rglob(f"{prefix}*.jsonl"))
# ----- async wait helpers --------------------------------------------
async def wait_for_glob(
self,
parent: Path,
pattern: str,
timeout: float,
poll: float = 0.5,
) -> Path:
"""Poll ``parent.glob(pattern)`` until the first match appears or timeout."""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if parent.is_dir():
matches = sorted(parent.glob(pattern))
if matches:
return matches[0]
await asyncio.sleep(poll)
listing = [p.name for p in parent.iterdir()] if parent.is_dir() else []
raise TimeoutError(
f"timeout after {timeout}s waiting for {parent}/{pattern}; dir listing: {listing}",
)
async def wait_for_populated(
self,
parent: Path,
pattern: str,
min_bytes: int,
timeout: float,
poll: float = 1.0,
) -> Path:
"""Like ``wait_for_glob`` but only returns once the file exceeds ``min_bytes``.
``daily_create`` writes a ~50-byte frontmatter-only stub before the
agent fills in the body, so a simple ``glob`` check returns too early.
"""
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if parent.is_dir():
for p in sorted(parent.glob(pattern)):
try:
if p.stat().st_size >= min_bytes:
return p
except OSError:
pass
await asyncio.sleep(poll)
listing = [(p.name, p.stat().st_size) for p in parent.iterdir()] if parent.is_dir() else []
raise TimeoutError(
f"timeout after {timeout}s waiting for {parent}/{pattern} with size>={min_bytes}; " f"current: {listing}",
)
async def wait_for_server(self, url: str, timeout: float = 10.0) -> None:
"""Poll ``url`` until the server answers (any 2xx/4xx counts). Used by
the webhook test so the POST isn't racing uvicorn's bind."""
import httpx # local — keep optional
deadline = time.monotonic() + timeout
async with httpx.AsyncClient(timeout=1.0) as client:
while time.monotonic() < deadline:
try:
resp = await client.request("HEAD", url)
if resp.status_code in (200, 202, 404, 405):
return
except (httpx.ConnectError, httpx.ReadError):
pass
await asyncio.sleep(0.2)
raise TimeoutError(f"server didn't come up at {url} within {timeout}s")
# ----- agent recorder factory ----------------------------------------
def record_agents(
self,
prefix: str = "agent",
dump_dir: Path | None = None,
) -> AgentMemoryRecorder:
"""Context manager that captures every Agent created in its block.
Dumps default to ``<vault>/agent_logs/`` so they're cleaned up with
the throwaway vault. Pass ``dump_dir`` to persist them elsewhere
(e.g. for post-mortem inspection of a failing run).
"""
return AgentMemoryRecorder(
dump_dir=dump_dir or self.vault_dir / "agent_logs",
prefix=prefix,
)
# ─────────────────────────────────────────────────────────────────────────────
# Public entry point
# ─────────────────────────────────────────────────────────────────────────────
@contextlib.contextmanager
def vault_env(
*,
chdir: bool = True,
vault_name: str = ".reme",
load_env_file: bool = True,
) -> Iterator[VaultEnv]:
"""Yield a fresh ``VaultEnv`` rooted at a temp workspace.
- Creates ``<tmp>/<vault_name>`` eagerly so seed helpers work before
``make_app()``.
- chdirs into the temp workspace (so relative vault_dir resolves
correctly and any helper that writes under cwd lands inside the
throwaway tree).
- Loads ``.env`` once (idempotent safe to call repeatedly).
The workspace is cleaned up automatically when the block exits. The
caller is still responsible for ``await env.close_all()`` to release
any apps it started.
"""
if load_env_file:
from reme4.utils import load_env
load_env()
with tempfile.TemporaryDirectory() as tmp_dir:
workspace = Path(tmp_dir).resolve()
vault = workspace / vault_name
vault.mkdir(parents=True, exist_ok=True)
env = VaultEnv(workspace=workspace, vault_dir=vault)
if chdir:
with temp_chdir(workspace):
yield env
else:
yield env
# ─────────────────────────────────────────────────────────────────────────────
# Script entry — replicate the original dreamer manual-seed CLI.
# ─────────────────────────────────────────────────────────────────────────────
# pylint: disable=missing-function-docstring
def main() -> None:
if len(sys.argv) < 2:
print(f"usage: {sys.argv[0]} <vault_dir>", file=sys.stderr)
sys.exit(2)
vault = Path(sys.argv[1]).resolve()
vault.mkdir(parents=True, exist_ok=True)
env = VaultEnv(workspace=vault.parent, vault_dir=vault)
removed = env.clean()
if removed:
print(f"cleaned {len(removed)} dir(s) under {vault}: {', '.join(removed)}")
seeded = env.seed_dream_vault()
if seeded:
print(f"seeded {len(seeded)} file(s) under {vault}:")
for f in seeded:
print(f" + {f}")
else:
print(f"vault {vault} already seeded — no changes")
print(f"\nDream this file:\n {vault}/{DREAM_INPUT_PATH}")
if __name__ == "__main__":
main()

View file

@ -5,52 +5,23 @@ environment or a .env file at the repo root. Hits the real LLM API.
"""
import asyncio
import os
import tempfile
import sys
from pathlib import Path
from reme4 import Application
from reme4.config import resolve_app_config
from reme4.enumeration import ComponentEnum
from reme4.utils import load_env
INTEGRATION_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(INTEGRATION_DIR))
load_env()
# pylint: disable=wrong-import-position
from _vault_fixture import vault_env # noqa: E402
class _temp_chdir:
def __init__(self, path):
self.path = path
self._old = None
def __enter__(self):
self._old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self._old)
async def _make_app() -> Application:
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
app = Application(**cfg)
await app.start()
return app
def _find_session_files(vault_root: Path, prefix: str = "session_reme_") -> list[Path]:
resource_dir = vault_root / "resource"
if not resource_dir.exists():
return []
return sorted(resource_dir.rglob(f"{prefix}*.jsonl"))
from reme4.enumeration import ComponentEnum # noqa: E402
async def _run_session_persistence() -> None:
"""Two consecutive replies with the same session_id should share context."""
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
wrapper = app.context.components[ComponentEnum.AGENT_WRAPPER]["default"]
sid = "test-persist-session"
@ -66,7 +37,7 @@ async def _run_session_persistence() -> None:
assert text_1, "Empty first reply"
# Verify session file was created
files_after_1 = _find_session_files(vault_root)
files_after_1 = env.session_state_files()
print(f"[session_persist] session files after reply 1: {files_after_1}")
assert len(files_after_1) == 1, f"Expected 1 session file, got {len(files_after_1)}"
assert sid in files_after_1[0].name
@ -83,15 +54,14 @@ async def _run_session_persistence() -> None:
print("✓ test_session_persistence passed")
finally:
await app.close()
await env.close_all()
async def _run_fork_session() -> None:
"""fork_session=True should create a new session file with a new session_id."""
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
wrapper = app.context.components[ComponentEnum.AGENT_WRAPPER]["default"]
sid = "test-fork-origin"
@ -102,7 +72,7 @@ async def _run_fork_session() -> None:
session_id=sid,
system_prompt="You are a helpful assistant. Keep answers short.",
)
files_before_fork = _find_session_files(vault_root)
files_before_fork = env.session_state_files()
assert len(files_before_fork) == 1
# Fork the session
@ -117,7 +87,7 @@ async def _run_fork_session() -> None:
assert "42" in text_fork, f"Forked session should recall '42', got: {text_fork!r}"
# Verify: original file still exists + new forked file created
files_after_fork = _find_session_files(vault_root)
files_after_fork = env.session_state_files()
print(f"[fork_session] session files after fork: {[f.name for f in files_after_fork]}")
assert (
len(files_after_fork) == 2
@ -126,20 +96,19 @@ async def _run_fork_session() -> None:
# Forked session_id should differ from the original
assert forked_sid != sid, f"Forked session_id should differ from original, got {forked_sid!r}"
original_file = vault_root / "resource" / files_before_fork[0].relative_to(vault_root / "resource")
original_file = files_before_fork[0]
assert original_file.exists(), "Original session file should still exist after fork"
print("✓ test_fork_session passed")
finally:
await app.close()
await env.close_all()
async def _run_no_session_id() -> None:
"""When session_id is empty, no session file should be created."""
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
wrapper = app.context.components[ComponentEnum.AGENT_WRAPPER]["default"]
_, msg = await wrapper.reply(
@ -150,12 +119,12 @@ async def _run_no_session_id() -> None:
print(f"\n[no_session] reply: {text!r}")
assert text, "Empty reply"
files = _find_session_files(vault_root)
files = env.session_state_files()
assert len(files) == 0, f"No session files should be created without session_id, found {files}"
print("✓ test_no_session_id passed")
finally:
await app.close()
await env.close_all()
async def _run_all() -> None:

View file

@ -14,21 +14,14 @@ environment or a .env file at the repo root. Hits the real LLM API.
"""
import asyncio
import json
import os
import tempfile
from datetime import date as _date
import sys
from pathlib import Path
from agentscope.agent import Agent
INTEGRATION_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(INTEGRATION_DIR))
from reme4 import Application
from reme4.config import resolve_app_config
from reme4.utils import load_env
load_env()
DUMP_DIR = Path(__file__).resolve().parent / "agent_logs"
# pylint: disable=wrong-import-position
from _vault_fixture import vault_env # noqa: E402
SEED_STEM = "auth-middleware-rewrite"
SEED_BODY = """---
@ -56,40 +49,6 @@ description: JWT auth middleware rewrite driven by legal/compliance requirements
"""
def _today() -> str:
return _date.today().isoformat()
class _temp_chdir:
def __init__(self, path):
self.path = path
self._old = None
def __enter__(self):
self._old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self._old)
async def _make_app() -> Application:
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
app = Application(**cfg)
await app.start()
return app
def _seed_note(vault_root: Path, today: str) -> Path:
"""Write the existing auth-middleware-rewrite note under today's daily folder."""
day_dir = vault_root / "daily" / today
day_dir.mkdir(parents=True, exist_ok=True)
path = day_dir / f"{SEED_STEM}.md"
path.write_text(SEED_BODY, encoding="utf-8")
return path
def _auth_messages() -> list[dict]:
"""Messages continuing the auth middleware thread."""
return [
@ -167,71 +126,25 @@ def _read_text(p: Path) -> str:
return p.read_text(encoding="utf-8")
class _AgentMemoryRecorder:
"""Monkey-patches Agent.__init__ to capture every agent created inside
the ``with`` block, then dumps each agent's memory to a jsonl file in
DUMP_DIR on exit.
"""
def __init__(self, dump_dir: Path, prefix: str = "agent_memory"):
"""init"""
self.dump_dir = dump_dir
self.prefix = prefix
self.agents: list[Agent] = []
self._orig_init = None
self.dumped_paths: list[Path] = []
def __enter__(self):
"""Monkey-patch Agent.__init__."""
self._orig_init = Agent.__init__
agents = self.agents
orig = self._orig_init
def _capturing_init(agent_self, *args, **kwargs):
orig(agent_self, *args, **kwargs)
agents.append(agent_self)
Agent.__init__ = _capturing_init
return self
def __exit__(self, *exc):
"""Restore the original __init__."""
Agent.__init__ = self._orig_init
async def dump(self) -> list[Path]:
"""Dump all agent context histories."""
self.dump_dir.mkdir(parents=True, exist_ok=True)
for stale in self.dump_dir.glob(f"{self.prefix}_*.jsonl"):
stale.unlink()
for idx, agent in enumerate(self.agents, 1):
messages = agent.state.context
name = getattr(agent, "name", "agent") or "agent"
out_path = self.dump_dir / f"{self.prefix}_{idx:02d}_{name}.jsonl"
with out_path.open("w", encoding="utf-8") as f:
for msg in messages:
f.write(json.dumps(msg.model_dump(), ensure_ascii=False, default=str) + "\n")
self.dumped_paths.append(out_path)
return self.dumped_paths
def test_auto_memory_create():
"""CREATE a new note from scratch with a fresh session_id."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
today = _today()
today = env.today
print("\n" + "=" * 70)
print("[setup] vault_root =", vault_root)
print("[setup] vault_root =", env.vault_dir)
print("[setup] today =", today)
print("=" * 70)
pytorch_session_id = "pytorch-distributed-training"
with _AgentMemoryRecorder(DUMP_DIR, prefix="agent_create") as recorder:
# daily_create stamps the file as ``session_<session_id>.md``,
# so the path metadata daily_create returns carries the prefix.
expected_stem = f"session_{pytorch_session_id}"
with env.record_agents(prefix="agent_create") as recorder:
response = await app.run_job(
"auto_memory",
messages=_pytorch_messages(),
@ -244,9 +157,9 @@ def test_auto_memory_create():
assert response.success is True, f"CREATE job failed: {response.answer!r}"
meta = response.metadata or {}
assert meta.get("created") is True, f"Expected created=True, got {meta!r}"
assert meta.get("path") == f"daily/{today}/{pytorch_session_id}.md"
assert meta.get("path") == f"daily/{today}/{expected_stem}.md"
pytorch_path = vault_root / meta["path"]
pytorch_path = env.vault_dir / meta["path"]
assert pytorch_path.is_file(), f"created note not found at {pytorch_path}"
pytorch_text = _read_text(pytorch_path)
@ -274,7 +187,7 @@ def test_auto_memory_create():
print("test_auto_memory_create passed")
print("=" * 70)
finally:
await app.close()
await env.close_all()
asyncio.run(run())
@ -283,22 +196,25 @@ def test_auto_memory_update():
"""UPDATE an existing note — old facts must survive, new facts must land."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
today = _today()
seed_path = _seed_note(vault_root, today)
today = env.today
# daily_create resolves session_id=SEED_STEM to file stem
# ``session_<SEED_STEM>`` — seed at that exact stem so the
# UPDATE branch finds the existing note rather than CREATE.
expected_stem = f"session_{SEED_STEM}"
seed_path = env.seed_daily_note(expected_stem, SEED_BODY)
seed_before = _read_text(seed_path)
assert "legal/compliance" in seed_before
print("\n" + "=" * 70)
print("[setup] vault_root =", vault_root)
print("[setup] vault_root =", env.vault_dir)
print("[setup] today =", today)
print("[setup] seed_path =", seed_path)
print("=" * 70)
with _AgentMemoryRecorder(DUMP_DIR, prefix="agent_update") as recorder:
with env.record_agents(prefix="agent_update") as recorder:
response = await app.run_job(
"auto_memory",
messages=_auth_messages(),
@ -311,7 +227,7 @@ def test_auto_memory_update():
assert response.success is True, f"UPDATE job failed: {response.answer!r}"
meta = response.metadata or {}
assert meta.get("created") is False, f"Expected created=False, got {meta!r}"
assert meta.get("path") == f"daily/{today}/{SEED_STEM}.md"
assert meta.get("path") == f"daily/{today}/{expected_stem}.md"
seed_after = _read_text(seed_path)
print("\n" + "=" * 70)
@ -338,7 +254,7 @@ def test_auto_memory_update():
print("test_auto_memory_update passed")
print("=" * 70)
finally:
await app.close()
await env.close_all()
asyncio.run(run())

View file

@ -2,38 +2,36 @@
Drives the ``auto_resource`` step against a real LLM. Three scenarios:
1. **CREATE (added)**: places a resource file in ``resource/{date}/``,
calls ``auto_resource`` with change="added". Expects a new note
``daily/{date}/resource_{hash}.md`` with key facts from the file.
1. **CREATE (added)** / **UPDATE (modified)**: places a resource file in
``resource/{date}/``, calls ``auto_resource`` with the matching change.
The current AS-backed step only captures the agent's read+reason
transcript to ``resource/{date}/session_state_{sid}.jsonl`` daily-note
writing is handled by whichever ``auto_memory_*`` step comes next in
the chain (the CC variant fork-writes from session_id; the AS variant
needs explicit messages and is wired separately). So this test asserts
on the session_state landing + agent fact coverage, not on a daily
note file.
2. **UPDATE (modified)**: seeds an existing resource note, updates the
resource file, calls ``auto_resource`` with change="modified".
Expects the note to reflect the updated content.
3. **DELETE (deleted)**: seeds a resource note, calls ``auto_resource``
with change="deleted". Expects the note file to be removed.
2. **DELETE (deleted)**: seeds a resource note under
``daily/{date}/session_{sid}.md``, calls ``auto_resource`` with
change="deleted". Expects the note file to be removed (the step
stamps ``path`` on its metadata only in this branch).
Requires LLM_API_KEY (and optionally LLM_BASE_URL / LLM_MODEL_NAME) in the
environment or a .env file at the repo root. Hits the real LLM API.
"""
import asyncio
import json
import os
import tempfile
from datetime import date as _date
import sys
from pathlib import Path
from agentscope.agent import Agent
INTEGRATION_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(INTEGRATION_DIR))
from reme4 import Application
from reme4.config import resolve_app_config
from reme4.steps.evolve.auto_resource import _compute_session_id
from reme4.utils import load_env
# pylint: disable=wrong-import-position
from _vault_fixture import vault_env # noqa: E402
load_env()
DUMP_DIR = Path(__file__).resolve().parent / "agent_logs"
from reme4.steps.evolve.auto_resource import _compute_session_id # noqa: E402
RESOURCE_FILENAME = "project-roadmap.md"
RESOURCE_CONTENT_V1 = """\
@ -81,122 +79,38 @@ RESOURCE_CONTENT_V2 = """\
"""
def _today() -> str:
return _date.today().isoformat()
class _temp_chdir:
def __init__(self, path):
self.path = path
self._old = None
def __enter__(self):
self._old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self._old)
async def _make_app() -> Application:
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
app = Application(**cfg)
await app.start()
return app
def _place_resource(vault_root: Path, today: str, filename: str, content: str) -> str:
"""Write a resource file and return its vault-relative path."""
resource_dir = vault_root / "resource" / today
resource_dir.mkdir(parents=True, exist_ok=True)
path = resource_dir / filename
path.write_text(content, encoding="utf-8")
return f"resource/{today}/{filename}"
def _seed_resource_note(vault_root: Path, today: str, session_id: str, body: str) -> Path:
"""Pre-seed a resource note in daily/{date}/."""
day_dir = vault_root / "daily" / today
day_dir.mkdir(parents=True, exist_ok=True)
path = day_dir / f"{session_id}.md"
path.write_text(body, encoding="utf-8")
return path
def _read_text(p: Path) -> str:
return p.read_text(encoding="utf-8")
class _AgentMemoryRecorder:
"""Monkey-patches Agent.__init__ to capture every agent created inside
the ``with`` block, then dumps each agent's memory to a jsonl file.
def test_auto_resource_create():
"""CREATE branch: agent reads the resource file and saves session_state.
Daily-note writing belongs to the auto_memory_* step that follows; this
test only asserts that auto_resource_step ran the agent and persisted
its transcript under ``resource/{date}/session_state_{sid}.jsonl``.
"""
def __init__(self, dump_dir: Path, prefix: str = "agent_memory"):
self.dump_dir = dump_dir
self.prefix = prefix
self.agents: list[Agent] = []
self._orig_init = None
self.dumped_paths: list[Path] = []
def __enter__(self):
self._orig_init = Agent.__init__
agents = self.agents
orig = self._orig_init
def _capturing_init(agent_self, *args, **kwargs):
orig(agent_self, *args, **kwargs)
agents.append(agent_self)
Agent.__init__ = _capturing_init
return self
def __exit__(self, *exc):
Agent.__init__ = self._orig_init
async def dump(self) -> list[Path]:
"""Serialize captured agent transcripts to disk."""
self.dump_dir.mkdir(parents=True, exist_ok=True)
for stale in self.dump_dir.glob(f"{self.prefix}_*.jsonl"):
stale.unlink()
for idx, agent in enumerate(self.agents, 1):
messages = agent.state.context
name = getattr(agent, "name", "agent") or "agent"
out_path = self.dump_dir / f"{self.prefix}_{idx:02d}_{name}.jsonl"
with out_path.open("w", encoding="utf-8") as f:
for msg in messages:
f.write(json.dumps(msg.model_dump(), ensure_ascii=False, default=str) + "\n")
self.dumped_paths.append(out_path)
return self.dumped_paths
def test_auto_resource_create():
"""CREATE a resource note from a new file (change=added)."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
today = _today()
today = env.today
print("\n" + "=" * 70)
print("[setup] vault_root =", vault_root)
print("[setup] vault_root =", env.vault_dir)
print("[setup] today =", today)
print("=" * 70)
file_path = _place_resource(vault_root, today, RESOURCE_FILENAME, RESOURCE_CONTENT_V1)
file_path = env.place_resource(RESOURCE_FILENAME, RESOURCE_CONTENT_V1)
session_id = _compute_session_id(RESOURCE_FILENAME)
# daily_create prepends "session_agent_" to the session_id
expected_note_path = f"daily/{today}/session_agent_{session_id}.md"
expected_session_jsonl = env.vault_dir / "resource" / today / f"session_reme_{session_id}.jsonl"
print(f"[CREATE] file_path = {file_path}")
print(f"[CREATE] session_id = {session_id}")
print(f"[CREATE] expected note= {expected_note_path}")
print(f"[CREATE] file_path = {file_path}")
print(f"[CREATE] session_id = {session_id}")
print(f"[CREATE] expected transcript = {expected_session_jsonl.relative_to(env.vault_dir)}")
with _AgentMemoryRecorder(DUMP_DIR, prefix="agent_resource_create") as recorder:
with env.record_agents(prefix="agent_resource_create") as recorder:
response = await app.run_job(
"auto_resource",
file_path=file_path,
@ -208,67 +122,71 @@ def test_auto_resource_create():
assert response.success is True, f"CREATE job failed: {response.answer!r}"
meta = response.metadata or {}
assert meta.get("path") == expected_note_path, f"Unexpected path: {meta!r}"
assert meta.get("action") == "added"
assert meta.get("action") == "added", f"Unexpected action: {meta!r}"
assert meta.get("session_id") == session_id, f"Unexpected session_id: {meta!r}"
note_path = vault_root / expected_note_path
assert note_path.is_file(), f"Created note not found at {note_path}"
assert expected_session_jsonl.is_file(), (
f"agent session_state not persisted at {expected_session_jsonl}; "
f"session_state files under resource/: "
f"{[p.name for p in env.session_state_files(prefix='session_reme_')]}"
)
note_text = _read_text(note_path)
# Read the agent transcript and check it actually opened
# the resource file (the file_path should show up in a
# tool-call argument) so we know the step did its job.
transcript = _read_text(expected_session_jsonl)
print("\n" + "=" * 70)
print(f"[CREATE] {note_path} ({len(note_text)} bytes)")
print(f"[CREATE] body:\n{note_text}")
print("=" * 70)
print(f"[CREATE] {expected_session_jsonl.name} ({len(transcript)} bytes)")
topic_hits = [
needle
for needle in ("v2.0", "July 15", "Alice", "Bob", "p99", "200ms", "Redis")
if needle in note_text
for needle in ("v2.0", "July 15", "Alice", "Bob", "p99", "200ms", "Redis", file_path)
if needle in transcript
]
print(f"[CREATE] landed topic facts: {topic_hits}")
assert (
len(topic_hits) >= 3
), f"CREATE only captured {topic_hits!r} of expected facts\n--- CREATE ---\n{note_text}"
print(f"[CREATE] facts visible in transcript: {topic_hits}")
assert topic_hits, (
"agent transcript shows no signal it actually read the resource file; "
f"transcript head:\n{transcript[:500]}"
)
print("\n" + "=" * 70)
print("test_auto_resource_create passed")
print("=" * 70)
finally:
await app.close()
await env.close_all()
asyncio.run(run())
def test_auto_resource_update():
"""UPDATE an existing resource note (change=modified)."""
"""UPDATE branch: same contract as CREATE — only session_state changes."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
today = _today()
today = env.today
print("\n" + "=" * 70)
print("[setup] vault_root =", vault_root)
print("[setup] vault_root =", env.vault_dir)
print("[setup] today =", today)
print("=" * 70)
# First create a note via "added"
file_path = _place_resource(vault_root, today, RESOURCE_FILENAME, RESOURCE_CONTENT_V1)
# First run as "added" so the resource file exists and the
# initial transcript lands.
file_path = env.place_resource(RESOURCE_FILENAME, RESOURCE_CONTENT_V1)
session_id = _compute_session_id(RESOURCE_FILENAME)
session_jsonl = env.vault_dir / "resource" / today / f"session_reme_{session_id}.jsonl"
response = await app.run_job("auto_resource", file_path=file_path, change="added")
assert response.success is True, f"Initial create failed: {response.answer!r}"
assert session_jsonl.is_file(), "initial added run did not save session_state"
size_before = session_jsonl.stat().st_size
print(f"[UPDATE] transcript before modify ({size_before} bytes)")
note_abs = vault_root / "daily" / today / f"session_agent_{session_id}.md"
note_before = _read_text(note_abs)
print(f"[UPDATE] note before update ({len(note_before)} bytes)")
# Now update the resource file and call with "modified".
env.place_resource(RESOURCE_FILENAME, RESOURCE_CONTENT_V2)
# Now update the resource file and call with "modified"
_place_resource(vault_root, today, RESOURCE_FILENAME, RESOURCE_CONTENT_V2)
with _AgentMemoryRecorder(DUMP_DIR, prefix="agent_resource_update") as recorder:
with env.record_agents(prefix="agent_resource_update") as recorder:
response = await app.run_job(
"auto_resource",
file_path=file_path,
@ -280,29 +198,29 @@ def test_auto_resource_update():
assert response.success is True, f"UPDATE job failed: {response.answer!r}"
meta = response.metadata or {}
assert meta.get("action") == "modified"
assert meta.get("action") == "modified", f"Unexpected action: {meta!r}"
assert meta.get("session_id") == session_id, f"Unexpected session_id: {meta!r}"
note_after = _read_text(note_abs)
print("\n" + "=" * 70)
print(f"[UPDATE] {note_abs} ({len(note_before)} -> {len(note_after)} bytes)")
print(f"[UPDATE] body after:\n{note_after}")
print("=" * 70)
size_after = session_jsonl.stat().st_size
print(f"[UPDATE] transcript after modify ({size_after} bytes)")
assert size_after > size_before, (
f"transcript did not grow after modified run " f"({size_before} -> {size_after})"
)
transcript = _read_text(session_jsonl)
new_hits = [
needle
for needle in ("July 20", "150ms", "Dave", "rate limiting", "resolved")
if needle in note_after
if needle in transcript
]
print(f"[UPDATE] landed new facts: {new_hits}")
assert (
len(new_hits) >= 2
), f"UPDATE only landed {new_hits!r} of expected new facts\n--- AFTER ---\n{note_after}"
print(f"[UPDATE] V2 facts visible in transcript: {new_hits}")
assert new_hits, "modified run added no V2 content to the transcript; " f"tail:\n{transcript[-800:]}"
print("\n" + "=" * 70)
print("test_auto_resource_update passed")
print("=" * 70)
finally:
await app.close()
await env.close_all()
asyncio.run(run())
@ -311,14 +229,13 @@ def test_auto_resource_delete():
"""DELETE a resource note (change=deleted)."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
vault_root = Path(app.config.vault_dir).absolute()
today = _today()
today = env.today
print("\n" + "=" * 70)
print("[setup] vault_root =", vault_root)
print("[setup] vault_root =", env.vault_dir)
print("[setup] today =", today)
print("=" * 70)
@ -328,7 +245,7 @@ def test_auto_resource_delete():
# Seed the note file (daily_create prepends "session_agent_")
note_filename = f"session_agent_{session_id}"
seed_body = "---\nname: test\ndescription: test note\n---\n\nSome content.\n"
note_path = _seed_resource_note(vault_root, today, note_filename, seed_body)
note_path = env.seed_daily_note(note_filename, seed_body)
assert note_path.is_file()
print(f"[DELETE] seeded note: {note_path}")
@ -348,7 +265,7 @@ def test_auto_resource_delete():
print("test_auto_resource_delete passed")
print("=" * 70)
finally:
await app.close()
await env.close_all()
asyncio.run(run())

View file

@ -1,166 +1,91 @@
"""dreamer in-process integration test.
Loads the default reme4 config, seeds a rich workspace (pre-existing
digest nodes spread across the three buckets + a new daily that
exercises CREATE and UPDATE in each bucket), reindexes so search can
hit the pre-existing nodes, then calls `dream` and prints what
happened.
Loads the default reme4 config, seeds a rich workspace via the shared
``vault_env`` fixture (pre-existing digest nodes spread across the three
buckets + a new daily that exercises CREATE and UPDATE in each bucket),
reindexes so search can hit the pre-existing nodes, then calls ``dream``
and prints what happened.
Phase 1 classifies each sub-unit into one of {procedure, personal,
wiki}; Phase 2 dispatches to the bucket-specific integrate prompt
and writes via the canonical `write` / `edit` tools.
and writes via the canonical ``write`` / ``edit`` tools.
Usage (from anywhere):
VAULT_PATH=tests4/integration/vault python tests4/integration/test_dreamer_inproc.py
VAULT_PATH=tests4/integration/vault python tests4/integration/test_dreamer_inproc.py \\
python tests4/integration/test_dreamer_inproc.py
python tests4/integration/test_dreamer_inproc.py \\
daily/2026-05-28/auth-refactor/notes.md
Defaults:
VAULT_PATH unset tests4/integration/vault
Each run wipes `daily/`, `digest/`, and `reme_metadata/` under the
vault before reseeding, so the dreamer always starts from the same
fixture state. See _dreamer_fixture.py for what gets created and
the expected CREATE / UPDATE landings per bucket.
Each run wipes ``daily/``, ``digest/``, ``resource/``, and
``reme_metadata/`` under a freshly-built throwaway vault before
reseeding, so the dreamer always starts from the same fixture state.
See ``_vault_fixture.py`` (``seed_dream_vault`` / ``DREAM_INPUT_PATH``)
for what gets created and the expected CREATE / UPDATE landings per
bucket.
Required env (from .env or shell):
LLM_API_KEY, LLM_BASE_URL, LLM_MODEL_NAME for the Phase 1/2 agents
"""
import asyncio
import json
import os
import sys
from pathlib import Path
from agentscope.agent import Agent
# Make `reme4` importable regardless of the caller's cwd; and make the
# fixture module importable as a top-level name.
REPO_ROOT = Path(__file__).resolve().parents[2]
# Make ``_vault_fixture`` importable as a top-level module regardless of
# the caller's cwd.
INTEGRATION_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(REPO_ROOT))
sys.path.insert(0, str(INTEGRATION_DIR))
# pylint: disable=wrong-import-position
from _dreamer_fixture import clean_vault, seed_vault, INPUT_PATH # noqa: E402
VAULT = os.environ.get("VAULT_PATH", "tests4/integration/vault")
class _AgentMemoryRecorder:
"""Monkey-patches Agent.__init__ to capture every agent created inside
the ``with`` block, then dumps each agent's context history to a jsonl
file under ``<vault>/agent_logs/`` on dump().
Used to inspect the actual ReAct trace of Phase 1 extract + Phase 2
integrate (per sub-unit) what tools were called in what order, what
candidates were recalled, what the LLM decided.
"""
def __init__(self, vault: Path, prefix: str = "dream"):
self.dump_dir = vault / "agent_logs"
self.prefix = prefix
self.agents: list[Agent] = []
self._orig_init = None
self.dumped_paths: list[Path] = []
def __enter__(self):
self._orig_init = Agent.__init__
agents = self.agents
orig = self._orig_init
def _capturing_init(agent_self, *args, **kwargs):
orig(agent_self, *args, **kwargs)
agents.append(agent_self)
Agent.__init__ = _capturing_init
return self
def __exit__(self, *exc):
Agent.__init__ = self._orig_init
async def dump(self) -> list[Path]:
"""Dump all captured agents' context to <vault>/agent_logs/."""
self.dump_dir.mkdir(parents=True, exist_ok=True)
for stale in self.dump_dir.glob(f"{self.prefix}_*.jsonl"):
stale.unlink()
for idx, agent in enumerate(self.agents, 1):
messages = agent.state.context
name = getattr(agent, "name", "agent") or "agent"
out_path = self.dump_dir / f"{self.prefix}_{idx:02d}_{name}.jsonl"
with out_path.open("w", encoding="utf-8") as f:
for msg in messages:
f.write(json.dumps(msg.model_dump(), ensure_ascii=False, default=str) + "\n")
self.dumped_paths.append(out_path)
return self.dumped_paths
from _vault_fixture import DREAM_INPUT_PATH, vault_env # noqa: E402
async def main() -> None:
"""Main function for testing the ReMeFs CLI."""
from reme4 import ReMe # noqa: E402
from reme4.config import resolve_app_config # noqa: E402
from reme4.utils import load_env # noqa: E402
"""Seed a vault, reindex it, run ``dream`` on the seeded daily note."""
rel_input = sys.argv[1] if len(sys.argv) > 1 else DREAM_INPUT_PATH
os.chdir(REPO_ROOT) # so load_env() picks up the repo's .env
load_env()
with vault_env() as env:
seeded = env.seed_dream_vault()
print(f"--- seeded {len(seeded)} fixture file(s) under {env.vault_dir}")
vault = Path(VAULT).resolve()
vault.mkdir(parents=True, exist_ok=True)
app = await env.make_reme()
print(f"--- vault_dir: {env.vault_dir}")
print(f"--- input: {rel_input}")
removed = clean_vault(vault)
if removed:
print(f"--- cleaned {len(removed)} dir(s) under {vault}: {', '.join(removed)}")
seeded = seed_vault(vault)
print(f"--- seeded {len(seeded)} fixture file(s) under {vault}")
try:
# Reindex first so search can actually find the pre-seeded
# digest/ nodes — otherwise Phase 2 recall returns empty and
# every sub-unit ends up as CREATE (UPDATE path not exercised).
print("\n--- reindexing vault so Phase 2 recall has something to hit")
await app.run_job("reindex")
rel_input = sys.argv[1] if len(sys.argv) > 1 else INPUT_PATH
print(f"\n--- running dream path={rel_input}")
with env.record_agents(prefix="dream") as recorder:
resp = await app.run_job("dream", path=rel_input)
dumped = await recorder.dump()
print(f"\n--- dumped {len(dumped)} agent memory file(s) to {recorder.dump_dir}")
for p in dumped:
print(f" {p.relative_to(recorder.dump_dir)}")
cfg = resolve_app_config(vault_dir=str(vault))
print(f"--- vault_dir: {cfg.get('vault_dir')}")
print(f"--- input: {rel_input}")
print("\n=== Response.success ===")
print(resp.success)
print("\n=== Response.answer ===")
print(resp.answer)
print("\n=== Response.metadata (DreamResult fields) ===")
for k, v in (resp.metadata or {}).items():
if isinstance(v, list) and len(v) > 8:
print(f" {k}: list({len(v)} items) head={v[:3]!r}")
else:
print(f" {k}: {v!r}")
finally:
await env.close_all()
app = ReMe(**cfg)
await app.start()
try:
# Reindex first so search can actually find the pre-seeded
# digest/ nodes — otherwise Phase 2 recall returns empty and
# every sub-unit ends up as CREATE (UPDATE path not exercised).
print("\n--- reindexing vault so Phase 2 recall has something to hit")
await app.run_job("reindex")
print(f"\n--- running dream path={rel_input}")
with _AgentMemoryRecorder(vault, prefix="dream") as recorder:
resp = await app.run_job("dream", path=rel_input)
dumped = await recorder.dump()
print(f"\n--- dumped {len(dumped)} agent memory file(s) to {recorder.dump_dir}")
for p in dumped:
print(f" {p.relative_to(vault)}")
print("\n=== Response.success ===")
print(resp.success)
print("\n=== Response.answer ===")
print(resp.answer)
print("\n=== Response.metadata (DreamResult fields) ===")
for k, v in (resp.metadata or {}).items():
if isinstance(v, list) and len(v) > 8:
print(f" {k}: list({len(v)} items) head={v[:3]!r}")
else:
print(f" {k}: {v!r}")
finally:
await app.close()
print("\n=== digest/ tree after dream ===")
digest_root = vault / "digest"
if not digest_root.exists():
print(" (no digest/ created)")
return
files = sorted(digest_root.rglob("*.md"))
if not files:
print(" (digest/ is empty)")
for p in files:
print(f"\n--- {p.relative_to(vault)} ---")
# print(p.read_text(encoding="utf-8"))
print("\n=== digest/ tree after dream ===")
digest_files = env.digest_files()
if not digest_files:
print(" (no digest files)")
for p in digest_files:
print(f"\n--- {p.relative_to(env.vault_dir)} ---")
# print(p.read_text(encoding="utf-8"))
if __name__ == "__main__":

View file

@ -6,50 +6,28 @@ Hits the real embedding API.
"""
import asyncio
import os
import tempfile
import sys
from pathlib import Path
import numpy as np
from reme4 import Application
from reme4.config import resolve_app_config
from reme4.enumeration import ComponentEnum
from reme4.schema import EmbNode
from reme4.utils import cosine_similarity, load_env
INTEGRATION_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(INTEGRATION_DIR))
load_env()
# pylint: disable=wrong-import-position
from _vault_fixture import vault_env # noqa: E402
class _temp_chdir:
"""chdir to path for the duration of the block; restore on exit."""
def __init__(self, path):
self.path = path
self._old = None
def __enter__(self):
self._old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self._old)
async def _make_app() -> Application:
"""Build and start an Application from the default config."""
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
app = Application(**cfg)
await app.start()
return app
from reme4.enumeration import ComponentEnum # noqa: E402
from reme4.schema import EmbNode # noqa: E402
from reme4.utils import cosine_similarity # noqa: E402
def test_embedding_health_check():
"""health_check() returns True with a working API key."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
store = app.context.components[ComponentEnum.EMBEDDING_STORE]["default"]
result = await store.health_check(timeout=10.0)
@ -57,7 +35,7 @@ def test_embedding_health_check():
assert store.is_healthy is True
print("✓ test_embedding_health_check passed")
finally:
await app.close()
await env.close_all()
asyncio.run(run())
@ -66,8 +44,8 @@ def test_embedding_single_text():
"""Single text produces a valid embedding vector."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
store = app.context.components[ComponentEnum.EMBEDDING_STORE]["default"]
emb = await store.get_embedding("Hello, world!")
@ -78,7 +56,7 @@ def test_embedding_single_text():
print(f"\n [single] len={len(emb)}, first5={emb[:5].tolist()}")
print("✓ test_embedding_single_text passed")
finally:
await app.close()
await env.close_all()
asyncio.run(run())
@ -87,8 +65,8 @@ def test_embedding_multiple_texts():
"""Batch embedding returns correct count and shapes."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
store = app.context.components[ComponentEnum.EMBEDDING_STORE]["default"]
texts = ["cat", "dog", "house"]
@ -101,7 +79,7 @@ def test_embedding_multiple_texts():
print(f"\n [{texts[i]}] len={len(emb)}, first5={emb[:5].tolist()}")
print("✓ test_embedding_multiple_texts passed")
finally:
await app.close()
await env.close_all()
asyncio.run(run())
@ -110,8 +88,8 @@ def test_embedding_cache_hit():
"""Same text returns cached result on second call."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
store = app.context.components[ComponentEnum.EMBEDDING_STORE]["default"]
text = "test caching behavior"
@ -127,7 +105,7 @@ def test_embedding_cache_hit():
print(f"\n [cache] len={len(emb1)}, first5={emb1[:5].tolist()}")
print("✓ test_embedding_cache_hit passed")
finally:
await app.close()
await env.close_all()
asyncio.run(run())
@ -136,8 +114,8 @@ def test_embedding_similarity():
"""Semantically similar texts have higher cosine similarity."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
store = app.context.components[ComponentEnum.EMBEDDING_STORE]["default"]
text_a = "The cat sat on the mat"
@ -157,7 +135,7 @@ def test_embedding_similarity():
assert sim_ab > sim_ac, f"similar pair ({sim_ab:.4f}) not > dissimilar ({sim_ac:.4f})"
print("✓ test_embedding_similarity passed")
finally:
await app.close()
await env.close_all()
asyncio.run(run())
@ -166,8 +144,8 @@ def test_embedding_node_embeddings():
"""get_node_embeddings fills embedding field on EmbNode objects."""
async def run():
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
store = app.context.components[ComponentEnum.EMBEDDING_STORE]["default"]
nodes = [
@ -182,7 +160,7 @@ def test_embedding_node_embeddings():
print(f"\n [node{i}] len={len(node.embedding)}, first5={node.embedding[:5].tolist()}")
print("✓ test_embedding_node_embeddings passed")
finally:
await app.close()
await env.close_all()
asyncio.run(run())

View file

@ -5,42 +5,19 @@ environment or a .env file at the repo root. Hits the real Anthropic API.
"""
import asyncio
import os
import tempfile
import sys
from pathlib import Path
from typing import Literal
from pydantic import BaseModel, Field
from reme4 import Application
from reme4.config import resolve_app_config
from reme4.steps.common.llm_demo import LLMDemoStep
from reme4.utils import load_env
INTEGRATION_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(INTEGRATION_DIR))
load_env()
# pylint: disable=wrong-import-position
from _vault_fixture import vault_env # noqa: E402
class _temp_chdir:
"""chdir to path for the duration of the block; restore on exit."""
def __init__(self, path):
self.path = path
self._old = None
def __enter__(self):
self._old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self._old)
async def _make_app() -> Application:
"""Build and start an Application from the default config (LLM wired via env vars)."""
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
app = Application(**cfg)
await app.start()
return app
from reme4.steps.common.llm_demo import LLMDemoStep # noqa: E402
class MathResult(BaseModel):
@ -65,7 +42,7 @@ class SentimentAnalysis(BaseModel):
)
async def _run_basic_chat(app: Application) -> None:
async def _run_basic_chat(app) -> None:
step = LLMDemoStep(app_context=app.context)
response = await step(
query="What is 1 + 1? Reply with just the number.",
@ -77,7 +54,7 @@ async def _run_basic_chat(app: Application) -> None:
print("✓ test_llm_demo_step_basic_chat passed")
async def _run_with_tool(app: Application) -> None:
async def _run_with_tool(app) -> None:
step = LLMDemoStep(app_context=app.context)
response = await step(
query="Use the add tool to compute 21 + 21 and report the result.",
@ -90,7 +67,7 @@ async def _run_with_tool(app: Application) -> None:
print("✓ test_llm_demo_step_with_tool passed")
async def _run_structured_output(app: Application) -> None:
async def _run_structured_output(app) -> None:
step = LLMDemoStep(app_context=app.context)
response = await step(
query="What is 15 multiplied by 7? Show your work.",
@ -107,7 +84,7 @@ async def _run_structured_output(app: Application) -> None:
print("✓ test_llm_demo_step_structured_output passed")
async def _run_structured_output_enum(app: Application) -> None:
async def _run_structured_output_enum(app) -> None:
step = LLMDemoStep(app_context=app.context)
response = await step(
query="Analyze the sentiment: 'I absolutely love this product! It exceeded all my expectations.'",
@ -125,15 +102,15 @@ async def _run_structured_output_enum(app: Application) -> None:
async def _run_all() -> None:
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
await _run_basic_chat(app)
await _run_with_tool(app)
await _run_structured_output(app)
await _run_structured_output_enum(app)
finally:
await app.close()
await env.close_all()
if __name__ == "__main__":

View file

@ -5,49 +5,25 @@ environment or a .env file at the repo root. Hits the real LLM API.
"""
import asyncio
import os
import sys
import tempfile
from pathlib import Path
from reme4 import Application
from reme4.config import resolve_app_config
from reme4.enumeration import ChunkEnum
from reme4.schema import StreamChunk
from reme4.steps.common.stream_llm_demo import StreamLLMDemoStep
from reme4.utils import load_env
from reme4.utils.common_utils import execute_stream_task
INTEGRATION_DIR = Path(__file__).resolve().parent
sys.path.insert(0, str(INTEGRATION_DIR))
load_env()
# pylint: disable=wrong-import-position
from _vault_fixture import vault_env # noqa: E402
class _temp_chdir:
"""chdir to path for the duration of the block; restore on exit."""
def __init__(self, path):
self.path = path
self._old = None
def __enter__(self):
self._old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self._old)
async def _make_app() -> Application:
"""Build and start an Application from the default config."""
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
app = Application(**cfg)
await app.start()
return app
from reme4.enumeration import ChunkEnum # noqa: E402
from reme4.schema import StreamChunk # noqa: E402
from reme4.steps.common.stream_llm_demo import StreamLLMDemoStep # noqa: E402
from reme4.utils.common_utils import execute_stream_task # noqa: E402
async def _test_stream_llm_basic_chat():
"""StreamLLMDemoStep streams text chunks via add_stream_string."""
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
step = StreamLLMDemoStep(app_context=app.context)
queue: asyncio.Queue = asyncio.Queue()
@ -85,13 +61,13 @@ async def _test_stream_llm_basic_chat():
assert streamed_text.strip() == text, f"Stream text mismatch: {streamed_text!r} vs {text!r}"
print("✓ test_stream_llm_basic_chat passed")
finally:
await app.close()
await env.close_all()
async def _test_stream_llm_with_tool():
"""StreamLLMDemoStep streams tool call events when tools are used."""
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
step = StreamLLMDemoStep(app_context=app.context)
queue: asyncio.Queue = asyncio.Queue()
@ -138,13 +114,13 @@ async def _test_stream_llm_with_tool():
assert "42" in text, f"Expected '42' in response, got: {text!r}"
print("✓ test_stream_llm_with_tool passed")
finally:
await app.close()
await env.close_all()
async def _test_stream_llm_fallback_no_stream():
"""Without stream_queue, still uses streaming under the hood for real-time output."""
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
step = StreamLLMDemoStep(app_context=app.context)
queue: asyncio.Queue = asyncio.Queue()
@ -176,7 +152,7 @@ async def _test_stream_llm_fallback_no_stream():
assert "2" in text, f"Expected '2' in response, got: {text!r}"
print("✓ test_stream_llm_fallback_no_stream passed")
finally:
await app.close()
await env.close_all()
def test_stream_llm_basic_chat():
@ -196,8 +172,8 @@ def test_stream_llm_fallback_no_stream():
async def _demo_stream_print():
"""Real-time streaming print demo — ask a longer question to see chunked output."""
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(tmp):
app = await _make_app()
with vault_env() as env:
app = await env.make_app()
try:
step = StreamLLMDemoStep(app_context=app.context)
queue: asyncio.Queue = asyncio.Queue()
@ -231,7 +207,7 @@ async def _demo_stream_print():
sys.stdout.flush()
print()
finally:
await app.close()
await env.close_all()
async def _run_all():

View file

@ -13,7 +13,8 @@ Successful dreams (and Phase 1 vacuous skips) upsert the current
``st_mtime`` so the next tick re-dreams only what actually changed.
Failures leave the catalog untouched.
We mock ``dream_one`` (needs an LLM) and inject a fake ``file_catalog``
We mock ``run_job`` (the dispatch hop that calls the configured
``dream`` job needs an LLM) and inject a fake ``file_catalog``
recording every get / upsert / delete / dump.
"""
@ -28,7 +29,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
from reme4.components.file_catalog import BaseFileCatalog
from reme4.components.runtime_context import RuntimeContext
from reme4.schema import FileNode
from reme4.schema import FileNode, Response
from reme4.steps import AutoDreamStep
from reme4.steps.evolve.dream import DreamResult
@ -36,6 +37,17 @@ warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba")
warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")
def _dream_response(path: str, **dream_fields) -> Response:
"""Wrap a DreamResult as the dispatched job would return it.
Mirrors what DreamStep.execute does:
self.context.response.metadata.update(result.model_dump())
"""
dr = DreamResult(path=path, **dream_fields)
success = not dr.error
return Response(success=success, answer=dr.summary or "ok", metadata=dr.model_dump())
class temp_chdir:
"""Context manager that temporarily ``chdir``s into a directory and restores cwd on exit."""
@ -104,11 +116,12 @@ def test_scans_date_md_and_date_folder():
seen: list[str] = []
async def _fake_dream(rel, _hint):
seen.append(rel)
return DreamResult(used_llm=True, path=rel, summary="ok")
async def _fake_run_job(name, **kwargs):
assert name == "dream", f"expected dispatch to 'dream' job, got {name!r}"
seen.append(kwargs["path"])
return _dream_response(kwargs["path"], used_llm=True, summary="ok")
with patch.object(step, "dream_one", side_effect=_fake_dream):
with patch.object(step, "run_job", side_effect=_fake_run_job):
resp = await step(ctx)
assert resp.success
@ -134,13 +147,13 @@ def test_resource_dir_is_not_scanned():
step.app_context.app_config.resource_dir = "resource"
ctx = RuntimeContext(date=today)
async def _fake_dream(rel, _hint):
return DreamResult(used_llm=True, path=rel, summary="ok")
async def _fake_run_job(_name, **kwargs):
return _dream_response(kwargs["path"], used_llm=True, summary="ok")
with patch.object(step, "dream_one", side_effect=_fake_dream) as dream_mock:
with patch.object(step, "run_job", side_effect=_fake_run_job) as run_job_mock:
await step(ctx)
paths = [c.args[0] for c in dream_mock.call_args_list]
paths = [c.kwargs["path"] for c in run_job_mock.call_args_list]
assert paths == [f"daily/{today}.md"]
print("✓ test_resource_dir_is_not_scanned passed")
@ -160,9 +173,9 @@ def test_unchanged_files_skipped_via_catalog_mtime():
step = _make_step(vault, today, existing_nodes=existing)
ctx = RuntimeContext(date=today)
with patch.object(step, "dream_one") as dream_mock:
with patch.object(step, "run_job") as run_job_mock:
resp = await step(ctx)
dream_mock.assert_not_called()
run_job_mock.assert_not_called()
assert resp.success
assert resp.metadata["files_unchanged"] == 1
@ -188,10 +201,10 @@ def test_changed_file_dreamed_and_catalog_updated():
step = _make_step(vault, today, existing_nodes=existing)
ctx = RuntimeContext(date=today)
async def _fake_dream(rel, _hint):
return DreamResult(used_llm=True, path=rel, summary="ok")
async def _fake_run_job(_name, **kwargs):
return _dream_response(kwargs["path"], used_llm=True, summary="ok")
with patch.object(step, "dream_one", side_effect=_fake_dream):
with patch.object(step, "run_job", side_effect=_fake_run_job):
resp = await step(ctx)
assert resp.success
@ -220,9 +233,9 @@ def test_deleted_file_dropped_from_catalog():
step = _make_step(vault, today, existing_nodes=existing)
ctx = RuntimeContext(date=today)
with patch.object(step, "dream_one") as dream_mock:
with patch.object(step, "run_job") as run_job_mock:
resp = await step(ctx)
dream_mock.assert_not_called()
run_job_mock.assert_not_called()
assert resp.success
assert resp.metadata["files_deleted"] == 1
@ -271,10 +284,10 @@ def test_failure_does_not_upsert():
step = _make_step(vault, today)
ctx = RuntimeContext(date=today)
async def _fake_dream(rel, _hint):
return DreamResult(used_llm=False, path=rel, error="boom")
async def _fake_run_job(_name, **kwargs):
return _dream_response(kwargs["path"], used_llm=False, error="boom")
with patch.object(step, "dream_one", side_effect=_fake_dream):
with patch.object(step, "run_job", side_effect=_fake_run_job):
resp = await step(ctx)
assert not resp.success
@ -299,10 +312,10 @@ def test_phase1_empty_still_upserts():
step = _make_step(vault, today)
ctx = RuntimeContext(date=today)
async def _fake_dream(rel, _hint):
return DreamResult(used_llm=True, path=rel, skipped=True, summary="empty")
async def _fake_run_job(_name, **kwargs):
return _dream_response(kwargs["path"], used_llm=True, skipped=True, summary="empty")
with patch.object(step, "dream_one", side_effect=_fake_dream):
with patch.object(step, "run_job", side_effect=_fake_run_job):
resp = await step(ctx)
assert resp.success
@ -327,12 +340,12 @@ def test_partial_failure_does_not_block_other_files():
step = _make_step(vault, today)
ctx = RuntimeContext(date=today)
async def _fake_dream(rel, _hint):
if rel.endswith("a.md"):
return DreamResult(used_llm=False, path=rel, error="boom")
return DreamResult(used_llm=True, path=rel, summary="ok")
async def _fake_run_job(_name, **kwargs):
if kwargs["path"].endswith("a.md"):
return _dream_response(kwargs["path"], used_llm=False, error="boom")
return _dream_response(kwargs["path"], used_llm=True, summary="ok")
with patch.object(step, "dream_one", side_effect=_fake_dream):
with patch.object(step, "run_job", side_effect=_fake_run_job):
resp = await step(ctx)
assert not resp.success

View file

@ -0,0 +1,315 @@
"""Unit tests for the ``cron`` job — schedule math + dispatch wiring.
Strategy:
* ``_parse_hh_mm`` / ``_next_fire_delay`` are exercised directly without
the BackgroundJob supervisor. Avoids wall-clock waits.
* ``__call__`` is tested by driving the job's own ``_stop_event``: the
loop runs at most one iteration with a tiny ``interval_seconds`` and
``run_on_start=True`` verifies the fire path actually dispatches the
downstream step exactly once.
* The dispatch target is a tiny in-test counter step registered into
the same step registry the production code uses, so we exercise the
real ``R.get(ComponentEnum.STEP, name) instantiate __call__``
path rather than mocking it.
* Tests run under a tempdir so the implicit Application context is
isolated from the real vault.
"""
# pylint: disable=protected-access
import asyncio
import datetime
import os
import tempfile
import zoneinfo
from contextlib import contextmanager
from pathlib import Path
from reme4 import Application
from reme4.components import R
from reme4.components.job.cron_job import CronJob
from reme4.config import resolve_app_config
from reme4.steps.base_step import BaseStep
@R.register("test_cron_counter_step")
class _CounterStep(BaseStep):
"""In-test counter — increments a class-level fire count on each invocation."""
fires: int = 0
async def execute(self):
type(self).fires += 1
if self.context is not None:
self.context.response.success = True
return self.context.response
@contextmanager
def _temp_chdir(path: Path):
old = os.getcwd()
os.chdir(path)
try:
yield
finally:
os.chdir(old)
def _make_job(**kwargs) -> CronJob:
# Default to a known-registered step so most tests can ignore dispatch wiring.
kwargs.setdefault("dispatch_step", "version_step")
return CronJob(**kwargs)
def _test_parse_hh_mm_valid() -> None:
job = _make_job(daily_at="03:00")
assert job._fire_hour == 3 and job._fire_minute == 0
job = _make_job(daily_at="23:59")
assert job._fire_hour == 23 and job._fire_minute == 59
print("OK parse_hh_mm_valid")
def _test_parse_hh_mm_invalid() -> None:
for bad in ["24:00", "03:60", "abc", "3", "03:", ":00"]:
try:
_make_job(daily_at=bad)
except ValueError:
continue
raise AssertionError(f"expected ValueError for daily_at={bad!r}")
print("OK parse_hh_mm_invalid")
def _test_requires_dispatch_step() -> None:
# neither dispatch_step nor dispatch_steps → ValueError
try:
CronJob(daily_at="03:00")
except ValueError:
print("OK requires_dispatch_step")
return
raise AssertionError("expected ValueError when no dispatch step is configured")
def _test_dispatch_steps_list() -> None:
# Mirrors watch_changes_step: dispatch_steps takes priority over dispatch_step,
# both forms accepted, defaults coalesce.
job1 = CronJob(dispatch_step="version_step", interval_seconds=60)
assert job1.dispatch_steps == ["version_step"]
job2 = CronJob(dispatch_steps=["a", "b"], interval_seconds=60)
assert job2.dispatch_steps == ["a", "b"]
job3 = CronJob(dispatch_step="x", dispatch_steps=["y", "z"], interval_seconds=60)
assert job3.dispatch_steps == ["y", "z"]
print("OK dispatch_steps_list")
def _test_dispatch_jobs_list() -> None:
# dispatch_job / dispatch_jobs coalesce the same way and satisfy the
# "at least one dispatch target" requirement on their own.
job1 = CronJob(dispatch_job="auto_dream", interval_seconds=60)
assert job1.dispatch_jobs == ["auto_dream"] and job1.dispatch_steps == []
job2 = CronJob(dispatch_jobs=["a", "b"], interval_seconds=60)
assert job2.dispatch_jobs == ["a", "b"]
print("OK dispatch_jobs_list")
def _test_requires_exactly_one_schedule() -> None:
# none of the three set
try:
_make_job()
except ValueError:
pass
else:
raise AssertionError("expected ValueError when no schedule is set")
# any two together
pairs = [
{"daily_at": "03:00", "interval_seconds": 60},
{"daily_at": "03:00", "cron": "0 3 * * *"},
{"interval_seconds": 60, "cron": "0 3 * * *"},
]
for kw in pairs:
try:
_make_job(**kw)
except ValueError:
continue
raise AssertionError(f"expected ValueError when two schedules set: {kw}")
# all three
try:
_make_job(daily_at="03:00", interval_seconds=60, cron="0 3 * * *")
except ValueError:
pass
else:
raise AssertionError("expected ValueError when all three schedules set")
print("OK requires_exactly_one_schedule")
def _test_cron_expression_valid() -> None:
for expr in ["0 3 * * *", "*/15 * * * *", "0 */6 * * *", "0 3 * * 1-5", "30 2 1 * *"]:
job = _make_job(cron=expr)
assert job.cron == expr
print("OK cron_expression_valid")
def _test_cron_expression_invalid() -> None:
# Eager validation — bad expressions must raise at construction time,
# so a typo fails at app start rather than at 3am.
for bad in ["not a cron", "0 25 * * *", "60 * * * *", "* * * 13 *", ""]:
try:
_make_job(cron=bad)
except ValueError:
continue
raise AssertionError(f"expected ValueError for cron={bad!r}")
print("OK cron_expression_invalid")
class _FrozenJob(CronJob):
"""CronJob subclass with deterministic 'now' for daily_at / cron delay math."""
def __init__(self, frozen_now: datetime.datetime, **kwargs):
kwargs.setdefault("dispatch_step", "version_step")
super().__init__(**kwargs)
self._frozen_now = frozen_now
def _next_fire_delay(self) -> float:
if self.interval_seconds:
return float(self.interval_seconds)
if self.cron:
from croniter import croniter
nxt = croniter(self.cron, self._frozen_now).get_next(datetime.datetime)
return (nxt - self._frozen_now).total_seconds()
target = self._frozen_now.replace(
hour=self._fire_hour,
minute=self._fire_minute,
second=0,
microsecond=0,
)
if target <= self._frozen_now:
target = target + datetime.timedelta(days=1)
return (target - self._frozen_now).total_seconds()
def _test_next_fire_delay_before_target() -> None:
tz = zoneinfo.ZoneInfo("Asia/Shanghai")
job = _FrozenJob(
frozen_now=datetime.datetime(2026, 6, 7, 2, 0, 0, tzinfo=tz),
daily_at="03:00",
)
assert job._next_fire_delay() == 3600
print("OK next_fire_delay_before_target")
def _test_next_fire_delay_after_target() -> None:
tz = zoneinfo.ZoneInfo("Asia/Shanghai")
job = _FrozenJob(
frozen_now=datetime.datetime(2026, 6, 7, 4, 0, 0, tzinfo=tz),
daily_at="03:00",
)
assert job._next_fire_delay() == 23 * 3600
print("OK next_fire_delay_after_target")
def _test_next_fire_delay_interval() -> None:
job = _make_job(interval_seconds=30)
assert job._next_fire_delay() == 30.0
print("OK next_fire_delay_interval")
def _test_next_fire_delay_cron_daily() -> None:
# cron "0 3 * * *" is exactly equivalent to daily_at "03:00" — same math,
# but exercised via the croniter path.
tz = zoneinfo.ZoneInfo("Asia/Shanghai")
job = _FrozenJob(
frozen_now=datetime.datetime(2026, 6, 7, 2, 0, 0, tzinfo=tz),
cron="0 3 * * *",
)
assert job._next_fire_delay() == 3600
print("OK next_fire_delay_cron_daily")
def _test_next_fire_delay_cron_every_6h() -> None:
tz = zoneinfo.ZoneInfo("Asia/Shanghai")
# "0 */6 * * *" fires at 00:00 / 06:00 / 12:00 / 18:00. At 02:00,
# next fire is 06:00 → 4 hours out.
job = _FrozenJob(
frozen_now=datetime.datetime(2026, 6, 7, 2, 0, 0, tzinfo=tz),
cron="0 */6 * * *",
)
assert job._next_fire_delay() == 4 * 3600
print("OK next_fire_delay_cron_every_6h")
def _test_next_fire_delay_cron_weekday_only() -> None:
tz = zoneinfo.ZoneInfo("Asia/Shanghai")
# 2026-06-07 is a Sunday. "0 3 * * 1-5" → next fire is Mon 2026-06-08 03:00.
# From Sun 02:00, that's 25 hours.
job = _FrozenJob(
frozen_now=datetime.datetime(2026, 6, 7, 2, 0, 0, tzinfo=tz),
cron="0 3 * * 1-5",
)
assert job._next_fire_delay() == 25 * 3600
print("OK next_fire_delay_cron_weekday_only")
async def _drive_one_fire(_tmp: Path) -> int:
"""Stand up an Application + dispatch the counter step on a cron tick."""
cfg = resolve_app_config(log_to_console=False, log_to_file=False, enable_logo=False)
cfg["enable_logo"] = False
app = Application(**cfg)
await app.start()
_CounterStep.fires = 0
try:
job = CronJob(
dispatch_step="test_cron_counter_step",
interval_seconds=1,
run_on_start=True,
)
job.app_context = app.context
# The BackgroundJob supervisor normally creates this in _start(); here we
# drive __call__ directly, so wire up the stop_event by hand.
job._stop_event = asyncio.Event()
# Fire once on start, then signal stop so the loop exits before the
# next interval elapses.
task = asyncio.create_task(job())
await asyncio.sleep(0.3) # let run_on_start fire propagate
job._stop_event.set()
await asyncio.wait_for(task, timeout=5.0)
return _CounterStep.fires
finally:
await app.close()
def _test_run_on_start_dispatches_once() -> None:
with tempfile.TemporaryDirectory() as tmp, _temp_chdir(Path(tmp)):
count = asyncio.run(_drive_one_fire(Path(tmp)))
assert count >= 1, f"expected at least one dispatch, got {count}"
print(f"OK run_on_start_dispatches_once (count={count})")
def main() -> None:
"""Entry point for the cron job unit tests."""
print("=== cron job unit tests ===")
_test_parse_hh_mm_valid()
_test_parse_hh_mm_invalid()
_test_requires_dispatch_step()
_test_dispatch_steps_list()
_test_dispatch_jobs_list()
_test_requires_exactly_one_schedule()
_test_cron_expression_valid()
_test_cron_expression_invalid()
_test_next_fire_delay_before_target()
_test_next_fire_delay_after_target()
_test_next_fire_delay_interval()
_test_next_fire_delay_cron_daily()
_test_next_fire_delay_cron_every_6h()
_test_next_fire_delay_cron_weekday_only()
_test_run_on_start_dispatches_once()
print("=== passed ===")
if __name__ == "__main__":
main()

View file

@ -42,7 +42,6 @@ from reme4.steps.file_io import (
stat as crud_stat,
write as crud_write,
)
from reme4.steps.transfer import download as crud_download
from reme4.utils.wikilink_handler import WikilinkHandler
warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba")
@ -198,42 +197,9 @@ def test_list_respects_limit_and_non_recursive():
# -- download ------------------------------------------------------------
def test_download_to_explicit_path():
"""download copies the vault file to dst_path."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store({"topics/a.md": "alpha"})
target = Path(tmp) / "out" / "a.md"
step = crud_download.DownloadStep(file_store=store)
await step(src_path="topics/a.md", dst_path=str(target))
payload = _metadata(step)
assert "error" not in payload
assert payload["dst_path"] == str(target)
assert target.read_text(encoding="utf-8") == "alpha"
await store.close()
print("✓ test_download_to_explicit_path passed")
asyncio.run(run())
def test_download_to_temp_when_dst_path_empty():
"""Without dst_path, download lands the file in a temp file and returns the path."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store({"topics/a.md": "alpha"})
step = crud_download.DownloadStep(file_store=store)
await step(src_path="topics/a.md")
payload = _metadata(step)
assert "error" not in payload
assert Path(payload["dst_path"]).read_text(encoding="utf-8") == "alpha"
await store.close()
print("✓ test_download_to_temp_when_dst_path_empty passed")
asyncio.run(run())
#
# DownloadStep lives in reme_cc (the local plugin overlay), not reme4
# main-line. See reme_cc/tests/ for its coverage.
# -- move ----------------------------------------------------------------
@ -1122,13 +1088,11 @@ def test_all_read_cases_one_store():
if __name__ == "__main__":
print("\n=== crud step tests (opaque-byte surface) ===")
# stat / list / download / move / delete
# stat / list / move / delete
test_stat_indexed_file()
test_stat_directory_fallback()
test_list_lists_files()
test_list_respects_limit_and_non_recursive()
test_download_to_explicit_path()
test_download_to_temp_when_dst_path_empty()
test_move_relocates_within_vault()
test_move_refuses_overwrite_without_flag()
test_move_default_retargets_inbound_links()

View file

@ -1,605 +0,0 @@
"""Tests for the resource ingest path: ``IngestStep`` + helpers.
``ingest`` is the **passive** ingest entry point external channels
push assets into ``resource/<YYYY-MM-DD>/``, where each call appends a
:class:`FileNode` row to ``meta.json`` (provenance on
``front_matter``) and regenerates the day's ``<date>.md`` view from
the updated meta. These tests exercise that contract end-to-end on a
temp vault, plus the pure ``_assemble_day_md`` helper in isolation.
The bucket file name is always derived: ``<channel>__<HHMMSS>__<basename>``.
Duplicates surface as errors (no silent suffixing).
"""
# pylint: disable=protected-access
import asyncio
import datetime
import json
import os
import re
import tempfile
import warnings
from pathlib import Path
from reme4.components.file_store import LocalFileStore
from reme4.schema import FileFrontMatter, FileNode
from reme4.steps.transfer import ingest as crud_ingest
from reme4.steps.transfer.ingest import _assemble_day_md
warnings.filterwarnings("ignore", category=DeprecationWarning, module="jieba")
warnings.filterwarnings("ignore", category=DeprecationWarning, module="pkg_resources")
class temp_chdir:
"""Context manager to temporarily chdir into a path and restore on exit."""
def __init__(self, path):
self.path = path
self.old = None
def __enter__(self):
self.old = os.getcwd()
os.chdir(self.path)
return self
def __exit__(self, *exc):
os.chdir(self.old)
async def _make_store() -> LocalFileStore:
"""Minimal LocalFileStore (embedding disabled). vault_path resolves to CWD."""
store = LocalFileStore(name="t", embedding_store="")
await store.start()
return store
def _metadata(step) -> dict:
return step.context.response.metadata
def _meta(tmp: str, date: str) -> list[dict]:
return json.loads((Path(tmp) / "resource" / date / "meta.json").read_text(encoding="utf-8"))
def _today() -> str:
return datetime.datetime.now().strftime("%Y-%m-%d")
# Matches the canonical derived filename shape.
_NAME_RE = re.compile(r"^([a-z0-9][a-z0-9-]*)__(\d{6})__(.+)$")
# -- _assemble_day_md ----------------------------------------------------
def _entry(name: str, **fm_fields) -> FileNode:
"""Build a FileNode resource entry: path = resource/<date>/<name>, all
provenance fields go onto front_matter (extras allowed)."""
return FileNode(
path=f"resource/2026-05-22/{name}",
st_mtime=0.0,
front_matter=FileFrontMatter(**fm_fields),
)
def test_assemble_day_md_renders_entries():
"""The derived view lists entries with channel / source / time / description."""
entries = [
_entry(
"wechat__143000__report.pdf",
description="Q1 report",
channel="wechat",
source="design-group",
received_at="2026-05-22T14:30:00",
),
_entry("browser__095501__bare.png", channel="browser"),
]
md = _assemble_day_md(entries, "2026-05-22")
assert "name: 2026-05-22" in md
assert "assets: [wechat__143000__report.pdf, browser__095501__bare.png]" in md
assert (
"- [[resource/2026-05-22/wechat__143000__report.pdf]] — wechat from `design-group` at 14:30 — Q1 report" in md
)
assert "- [[resource/2026-05-22/browser__095501__bare.png]] — browser" in md
print("✓ test_assemble_day_md_renders_entries passed")
def test_assemble_day_md_empty_bucket():
"""An empty bucket still produces a well-formed frontmatter + header."""
md = _assemble_day_md([], "2026-05-22")
assert "assets: []" in md
assert "# 2026-05-22 resources" in md
print("✓ test_assemble_day_md_empty_bucket passed")
# -- _validate_basename (direct, pure) -----------------------------------
def test_validate_basename_rejects_path_separators():
"""Path-separator basenames are rejected even if the public API can no
longer reach this code path (Path(...).name strips them) defense in depth."""
for bad in ("evil/payload.pdf", "..\\winpath.pdf", "../escape.pdf"):
err = crud_ingest._validate_basename(bad)
assert "path separators" in err or "reserved" in err, (bad, err)
print("✓ test_validate_basename_rejects_path_separators passed")
def test_validate_basename_rejects_dot_segments():
"""`.` and `..` are explicitly reserved."""
for bad in (".", ".."):
err = crud_ingest._validate_basename(bad)
assert "reserved" in err or "start with '.'" in err, (bad, err)
print("✓ test_validate_basename_rejects_dot_segments passed")
# -- _validate_channel (direct, pure) ------------------------------------
def test_validate_channel_accepts_safe_identifiers():
"""Lowercase letters / digits / dashes, starting alnum — all accepted."""
for ok in ("wechat", "email", "api", "browser", "slack-1", "ch1"):
assert crud_ingest._validate_channel(ok) == "", ok
print("✓ test_validate_channel_accepts_safe_identifiers passed")
def test_validate_channel_rejects_unsafe_identifiers():
"""Uppercase, underscores, leading dash, empty, special chars — rejected."""
for bad in ("", "WeChat", "we_chat", "-leading", "we chat", "we/chat", ""):
err = crud_ingest._validate_channel(bad)
assert err, bad
print("✓ test_validate_channel_rejects_unsafe_identifiers passed")
# -- IngestStep end-to-end --------------------------------------
def test_upload_first_call_creates_bucket():
"""First upload creates resource/<date>/, copies the asset under the derived name."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "incoming.pdf"
src.write_bytes(b"%PDF-fake")
step = crud_ingest.IngestStep(file_store=store)
await step(
path=str(src),
channel="wechat",
description="Q1 report",
metadata={"source": "design-group"},
)
payload = _metadata(step)
assert "error" not in payload, payload
date = payload["date"]
assert date == _today()
m = _NAME_RE.match(payload["name"])
assert m is not None, payload["name"]
assert m.group(1) == "wechat"
assert m.group(3) == "incoming.pdf"
assert payload["path"] == f"resource/{date}/{payload['name']}"
bucket = Path(tmp) / "resource" / date
assert (bucket / payload["name"]).read_bytes() == b"%PDF-fake"
meta = _meta(tmp, date)
assert len(meta) == 1
assert Path(meta[0]["path"]).name == payload["name"]
fm = meta[0]["front_matter"]
assert fm["channel"] == "wechat"
assert fm["source"] == "design-group"
assert fm["description"] == "Q1 report"
day_md = (bucket / f"{date}.md").read_text(encoding="utf-8")
assert f"name: {date}" in day_md
assert payload["name"] in day_md
await store.close()
print("✓ test_upload_first_call_creates_bucket passed")
asyncio.run(run())
def test_upload_metadata_optional():
"""metadata is optional — minimal call is just path + channel + description."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "small.txt"
src.write_text("x")
step = crud_ingest.IngestStep(file_store=store)
await step(path=str(src), channel="api", description="minimal")
payload = _metadata(step)
assert "error" not in payload, payload
row = _meta(tmp, payload["date"])[0]
fm = row["front_matter"]
assert fm["channel"] == "api"
assert fm.get("source", "") == ""
await store.close()
print("✓ test_upload_metadata_optional passed")
asyncio.run(run())
def test_upload_appends_to_existing_meta():
"""Subsequent uploads append to meta.json and regenerate <date>.md."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
names = []
for i, suffix in enumerate(("first", "second"), start=1):
src = Path(tmp) / f"{suffix}.txt"
src.write_text(f"payload-{i}")
step = crud_ingest.IngestStep(file_store=store)
await step(path=str(src), channel="email", description=f"item {i}")
payload = _metadata(step)
assert "error" not in payload, payload
names.append(payload["name"])
date = _today()
meta = _meta(tmp, date)
assert [Path(row["path"]).name for row in meta] == names
day_md = (Path(tmp) / "resource" / date / f"{date}.md").read_text(encoding="utf-8")
assert f"assets: [{', '.join(names)}]" in day_md
for name in names:
assert f"[[resource/{date}/{name}]]" in day_md
await store.close()
print("✓ test_upload_appends_to_existing_meta passed")
asyncio.run(run())
def test_upload_errors_on_duplicate_same_second(monkeypatch):
"""Two uploads of the same (channel, second, basename) → second one errors out."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
# Pin time so both calls land in the same HHMMSS slot deterministically.
fixed = datetime.datetime(2026, 5, 22, 15, 30, 22)
class _FrozenDT(datetime.datetime):
@classmethod
def now(cls, tz=None): # pylint: disable=unused-argument
return fixed
monkeypatch.setattr(crud_ingest.datetime, "datetime", _FrozenDT)
for i, body in enumerate((b"alpha", b"beta")):
src = Path(tmp) / "incoming.pdf"
src.write_bytes(body)
step = crud_ingest.IngestStep(file_store=store)
await step(path=str(src), channel="wechat", description="dup test")
payload = _metadata(step)
if i == 0:
assert "error" not in payload, payload
else:
assert "duplicate" in payload.get("error", "").lower(), payload
bucket = Path(tmp) / "resource" / "2026-05-22"
# Only the first upload's file should be on disk.
payloads = [p for p in bucket.iterdir() if p.is_file() and p.name.endswith(".pdf")]
assert len(payloads) == 1
assert payloads[0].read_bytes() == b"alpha"
meta = _meta(tmp, "2026-05-22")
assert len(meta) == 1
await store.close()
print("✓ test_upload_errors_on_duplicate_same_second passed")
asyncio.run(run())
def test_upload_errors_on_duplicate_against_on_disk_stray(monkeypatch):
"""A stray file on disk (no meta row) still counts as a collision → error."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
fixed = datetime.datetime(2026, 5, 22, 15, 30, 22)
class _FrozenDT(datetime.datetime):
@classmethod
def now(cls, tz=None): # pylint: disable=unused-argument
return fixed
monkeypatch.setattr(crud_ingest.datetime, "datetime", _FrozenDT)
bucket = Path(tmp) / "resource" / "2026-05-22"
bucket.mkdir(parents=True)
stray = bucket / "api__153022__report.pdf"
stray.write_bytes(b"orphan")
src = Path(tmp) / "report.pdf"
src.write_bytes(b"fresh")
step = crud_ingest.IngestStep(file_store=store)
await step(path=str(src), channel="api", description="fresh copy")
payload = _metadata(step)
assert "duplicate" in payload.get("error", "").lower(), payload
# Stray untouched.
assert stray.read_bytes() == b"orphan"
await store.close()
print("✓ test_upload_errors_on_duplicate_against_on_disk_stray passed")
asyncio.run(run())
def test_upload_rejects_missing_source():
"""Missing local file → error, no bucket created."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
step = crud_ingest.IngestStep(file_store=store)
await step(
path=str(Path(tmp) / "ghost.txt"),
channel="email",
description="x",
)
payload = _metadata(step)
assert "not found" in payload.get("error", "")
assert not (Path(tmp) / "resource").exists()
await store.close()
print("✓ test_upload_rejects_missing_source passed")
asyncio.run(run())
def test_upload_requires_channel():
"""Missing / blank / malformed channel → error."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "x.txt"
src.write_text("x")
for bad in (" ", "WeChat", "we_chat"):
step = crud_ingest.IngestStep(file_store=store)
await step(path=str(src), channel=bad, description="x")
payload = _metadata(step)
assert "channel" in payload.get("error", ""), (bad, payload)
await store.close()
print("✓ test_upload_requires_channel passed")
asyncio.run(run())
def test_upload_requires_description():
"""Empty / blank description → error."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "x.txt"
src.write_text("x")
step = crud_ingest.IngestStep(file_store=store)
await step(path=str(src), channel="api", description=" ")
payload = _metadata(step)
assert "description" in payload.get("error", "")
await store.close()
print("✓ test_upload_requires_description passed")
asyncio.run(run())
def test_upload_rejects_non_dict_metadata():
"""Passing a non-dict in `metadata=` → error."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "x.txt"
src.write_text("x")
step = crud_ingest.IngestStep(file_store=store)
await step(
path=str(src),
channel="api",
description="x",
metadata="source=foo",
)
payload = _metadata(step)
assert "metadata" in payload.get("error", "")
await store.close()
print("✓ test_upload_rejects_non_dict_metadata passed")
asyncio.run(run())
def test_upload_rejects_reserved_metadata_keys():
"""Step-managed keys (`name`, `channel`, `received_at`, `description`)
can't be smuggled in via metadata."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "x.txt"
src.write_text("x")
for bad in ("name", "channel", "received_at", "description"):
step = crud_ingest.IngestStep(file_store=store)
await step(
path=str(src),
channel="api",
description="x",
metadata={bad: "evil"},
)
payload = _metadata(step)
assert "reserved" in payload.get("error", ""), (bad, payload)
await store.close()
print("✓ test_upload_rejects_reserved_metadata_keys passed")
asyncio.run(run())
def test_upload_preserves_extra_metadata_keys():
"""Arbitrary keys in `metadata` land on the meta.json row verbatim."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "x.txt"
src.write_text("x")
step = crud_ingest.IngestStep(file_store=store)
await step(
path=str(src),
channel="api",
description="tagged",
metadata={
"source": "https://example.com",
"tag": "design",
"priority": 3,
},
)
payload = _metadata(step)
assert "error" not in payload, payload
row = _meta(tmp, payload["date"])[0]
fm = row["front_matter"]
assert fm["tag"] == "design"
assert fm["priority"] == 3
assert fm["source"] == "https://example.com"
await store.close()
print("✓ test_upload_preserves_extra_metadata_keys passed")
asyncio.run(run())
# -- basename-derivation safety ------------------------------------------
def test_upload_rejects_dotfile_source():
"""A source file whose basename starts with '.' → error."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
for bad in (".hidden", ".lock", ".env"):
src = Path(tmp) / bad
src.write_text("x")
step = crud_ingest.IngestStep(file_store=store)
await step(
path=str(src),
channel="api",
description="x",
)
payload = _metadata(step)
assert "start with '.'" in payload.get("error", ""), payload
src.unlink()
await store.close()
print("✓ test_upload_rejects_dotfile_source passed")
asyncio.run(run())
# -- payload-shape sanity ------------------------------------------------
def test_upload_records_received_at_internally():
"""`received_at` is not a caller param but the step stamps it from the
system clock so the day's <date>.md HH:MM column renders."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "doc.pdf"
src.write_bytes(b"%PDF")
step = crud_ingest.IngestStep(file_store=store)
await step(path=str(src), channel="api", description="x")
payload = _metadata(step)
assert "error" not in payload
meta = _meta(tmp, payload["date"])
assert len(meta) == 1
stamped = meta[0]["front_matter"]["received_at"]
parsed = datetime.datetime.fromisoformat(stamped)
assert parsed.strftime("%Y-%m-%d") == payload["date"]
# The HHMMSS slot in the name matches the stamped time.
m = _NAME_RE.match(payload["name"])
assert m is not None
assert m.group(2) == parsed.strftime("%H%M%S")
await store.close()
print("✓ test_upload_records_received_at_internally passed")
asyncio.run(run())
def test_upload_preserves_description_verbatim_in_meta():
"""meta.json carries the verbatim multi-line description (downstream agents
rely on it for analysis hints); only the day.md bullet flattens for display."""
async def run():
with tempfile.TemporaryDirectory() as tmp, temp_chdir(tmp):
store = await _make_store()
src = Path(tmp) / "doc.pdf"
src.write_bytes(b"%PDF")
multi = "wechat group screenshot\nfrom design-group at 14:30\nlikely a Q1 KPI table — extract numbers"
step = crud_ingest.IngestStep(file_store=store)
await step(
path=str(src),
channel="api",
description=multi,
)
payload = _metadata(step)
assert "error" not in payload, payload
# meta.json preserves the original — downstream dreamer sees full hint.
meta = _meta(tmp, payload["date"])
assert meta[0]["front_matter"]["description"] == multi
# day.md bullet is flattened (single line, no embedded newlines).
day_md = (Path(tmp) / "resource" / payload["date"] / f"{payload['date']}.md").read_text(encoding="utf-8")
flat = " ".join(multi.split())
assert flat in day_md
# The bullet line itself must contain the flattened text — and no embedded newline.
bullet_prefix = f"- [[resource/{payload['date']}/{payload['name']}]]"
bullets = [line for line in day_md.splitlines() if line.startswith(bullet_prefix)]
assert len(bullets) == 1, day_md
assert flat in bullets[0]
await store.close()
print("✓ test_upload_preserves_description_verbatim_in_meta passed")
asyncio.run(run())
def test_assemble_day_md_flattens_multiline_description():
"""Pure helper: a multi-line description on an entry renders as a single
bullet line with newlines collapsed."""
entries = [
_entry(
"api__120000__doc.pdf",
description="line one\nline two\n line three",
channel="api",
received_at="2026-05-22T12:00:00",
),
]
md = _assemble_day_md(entries, "2026-05-22")
assert "line one line two line three" in md
# The bullet line must contain the flattened text on a single line.
bullets = [line for line in md.splitlines() if line.startswith("- [[resource/2026-05-22/api__120000__doc.pdf]]")]
assert len(bullets) == 1, md
assert "line one line two line three" in bullets[0]
print("✓ test_assemble_day_md_flattens_multiline_description passed")
if __name__ == "__main__":
print("\n=== resource step tests ===")
test_assemble_day_md_renders_entries()
test_assemble_day_md_empty_bucket()
test_validate_basename_rejects_path_separators()
test_validate_basename_rejects_dot_segments()
test_validate_channel_accepts_safe_identifiers()
test_validate_channel_rejects_unsafe_identifiers()
test_upload_first_call_creates_bucket()
test_upload_metadata_optional()
test_upload_appends_to_existing_meta()
test_upload_rejects_missing_source()
test_upload_requires_channel()
test_upload_requires_description()
test_upload_rejects_non_dict_metadata()
test_upload_rejects_reserved_metadata_keys()
test_upload_preserves_extra_metadata_keys()
test_upload_rejects_dotfile_source()
test_upload_records_received_at_internally()
test_upload_preserves_description_verbatim_in_meta()
test_assemble_day_md_flattens_multiline_description()
print("\n所有测试通过!")