diff --git a/reme4/components/agent_wrapper/cc_agent_wrapper.py b/reme4/components/agent_wrapper/cc_agent_wrapper.py index ba81d117..c2731847 100644 --- a/reme4/components/agent_wrapper/cc_agent_wrapper.py +++ b/reme4/components/agent_wrapper/cc_agent_wrapper.py @@ -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 diff --git a/reme4/components/job/__init__.py b/reme4/components/job/__init__.py index af8c0255..5d923d66 100644 --- a/reme4/components/job/__init__.py +++ b/reme4/components/job/__init__.py @@ -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", ] diff --git a/reme4/components/job/cron_job.py b/reme4/components/job/cron_job.py new file mode 100644 index 00000000..d0306a4c --- /dev/null +++ b/reme4/components/job/cron_job.py @@ -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 diff --git a/reme4/config/default.yaml b/reme4/config/default.yaml index 71c28356..5bf8c019 100644 --- a/reme4/config/default.yaml +++ b/reme4/config/default.yaml @@ -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 /.md and session notes under //*.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: diff --git a/reme4/pyproject.toml b/reme4/pyproject.toml index 13243b46..ce0e8d09 100644 --- a/reme4/pyproject.toml +++ b/reme4/pyproject.toml @@ -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", diff --git a/reme4/steps/__init__.py b/reme4/steps/__init__.py index 30e67b5d..ec586895 100644 --- a/reme4/steps/__init__.py +++ b/reme4/steps/__init__.py @@ -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", ] diff --git a/reme4/steps/evolve/auto_dream.py b/reme4/steps/evolve/auto_dream.py index 544d783b..e9505667 100644 --- a/reme4/steps/evolve/auto_dream.py +++ b/reme4/steps/evolve/auto_dream.py @@ -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 ``/``: @@ -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/.md`` + ``daily//`` and dream each file - whose ``st_mtime`` doesn't already match its ``file_catalog`` entry.""" +class AutoDreamStep(BaseStep): + """Scan ``daily/.md`` + ``daily//`` 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. diff --git a/reme4/steps/evolve/auto_resource.py b/reme4/steps/evolve/auto_resource.py index 5f02d86f..9f5936f1 100644 --- a/reme4/steps/evolve/auto_resource.py +++ b/reme4/steps/evolve/auto_resource.py @@ -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}") diff --git a/reme4/steps/index/watch_changes.py b/reme4/steps/index/watch_changes.py index 1b50808a..af0ce3f9 100644 --- a/reme4/steps/index/watch_changes.py +++ b/reme4/steps/index/watch_changes.py @@ -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) diff --git a/tests4/integration/_dreamer_fixture.py b/tests4/integration/_dreamer_fixture.py deleted file mode 100644 index bc10f8da..00000000 --- a/tests4/integration/_dreamer_fixture.py +++ /dev/null @@ -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 credentials、refresh 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]} ", 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() diff --git a/tests4/integration/_vault_fixture.py b/tests4/integration/_vault_fixture.py new file mode 100644 index 00000000..987115a5 --- /dev/null +++ b/tests4/integration/_vault_fixture.py @@ -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 +``/.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 credentials、refresh 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 ``/__.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 ``/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 ``/`` 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//``. 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//.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//`` (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 ``/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 ``/`` 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]} ", 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() diff --git a/tests4/integration/test_agent_session.py b/tests4/integration/test_agent_session.py index 446e4926..127eee9d 100644 --- a/tests4/integration/test_agent_session.py +++ b/tests4/integration/test_agent_session.py @@ -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: diff --git a/tests4/integration/test_auto_memory.py b/tests4/integration/test_auto_memory.py index 653642ff..a1f0d4c5 100644 --- a/tests4/integration/test_auto_memory.py +++ b/tests4/integration/test_auto_memory.py @@ -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_.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 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()) diff --git a/tests4/integration/test_auto_resource.py b/tests4/integration/test_auto_resource.py index 44402080..67a81373 100644 --- a/tests4/integration/test_auto_resource.py +++ b/tests4/integration/test_auto_resource.py @@ -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()) diff --git a/tests4/integration/test_dreamer_inproc.py b/tests4/integration/test_dreamer_inproc.py index a5102efa..d0514f78 100644 --- a/tests4/integration/test_dreamer_inproc.py +++ b/tests4/integration/test_dreamer_inproc.py @@ -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 ``/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 /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__": diff --git a/tests4/integration/test_embedding.py b/tests4/integration/test_embedding.py index b04a7200..a788622b 100644 --- a/tests4/integration/test_embedding.py +++ b/tests4/integration/test_embedding.py @@ -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()) diff --git a/tests4/integration/test_llm.py b/tests4/integration/test_llm.py index f5e30351..ff105a16 100644 --- a/tests4/integration/test_llm.py +++ b/tests4/integration/test_llm.py @@ -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__": diff --git a/tests4/integration/test_stream_llm.py b/tests4/integration/test_stream_llm.py index 25a0b4bc..3605ca41 100644 --- a/tests4/integration/test_stream_llm.py +++ b/tests4/integration/test_stream_llm.py @@ -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(): diff --git a/tests4/unit/test_auto_dream.py b/tests4/unit/test_auto_dream.py index 4752fb85..c783f0d7 100644 --- a/tests4/unit/test_auto_dream.py +++ b/tests4/unit/test_auto_dream.py @@ -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 diff --git a/tests4/unit/test_cron_job.py b/tests4/unit/test_cron_job.py new file mode 100644 index 00000000..d100fb83 --- /dev/null +++ b/tests4/unit/test_cron_job.py @@ -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() diff --git a/tests4/unit/test_crud_steps.py b/tests4/unit/test_crud_steps.py index f9fb26ef..3066e369 100644 --- a/tests4/unit/test_crud_steps.py +++ b/tests4/unit/test_crud_steps.py @@ -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() diff --git a/tests4/unit/test_resource_steps.py b/tests4/unit/test_resource_steps.py deleted file mode 100644 index 0815637f..00000000 --- a/tests4/unit/test_resource_steps.py +++ /dev/null @@ -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//``, where each call appends a -:class:`FileNode` row to ``meta.json`` (provenance on -``front_matter``) and regenerates the day's ``.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: ``____``. -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//, 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//, 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 .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 .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所有测试通过!")