diff --git a/.planning/COMPILED-USER-PROFILE-V1.md b/.planning/COMPILED-USER-PROFILE-V1.md new file mode 100644 index 0000000..3ecba1f --- /dev/null +++ b/.planning/COMPILED-USER-PROFILE-V1.md @@ -0,0 +1,47 @@ + +# Compiled User Profile V1 +# Covers: evidence input, weekly GLM map/reduce, deterministic projection, and injection. +# Key terms: verbatim_memories, exact supports, user.md, SessionStart, feature flag. +# Read when: changing profile extraction, lifecycle, scheduling, rendering, or rollout. +# Status: implemented on an isolated feature branch; integration and activation pending. + + +## Contract + +MemoryMaster's SQLite database remains authoritative. The generated +`~/.memorymaster/projections/user.md` is a bounded, disposable view of stable +user facts and preferences; it is never parsed back into claims and contains +facts, not agent instructions. + +- Input is incremental sanitized `verbatim_memories`: user turns are evidence; + the preceding assistant turn is bounded context only. +- GLM map output proposes allowlisted facts with exact verbatim row IDs. GLM + reduce output must partition every candidate into add, reinforce, replace, or + ignore. Unknown IDs, sensitive content, malformed JSON, and instruction-shaped + text fail closed. +- New or replacement facts require support from at least two independent + sessions. SQLite records exact row IDs, session IDs, message hashes, and dates. +- Stable facts survive silence. Preferences expire after 90 unsupported days. +- A deterministic renderer writes at most 40 facts within an 800-token budget. +- The existing Dreaming task runs at most three map calls per invocation and + resumes from durable watermarks. `MEMORYMASTER_COMPILED_PROFILE=1` enables it; + the default is off. +- SessionStart injects only a bounded file carrying MemoryMaster's generated + marker. Hand-written or oversized files are ignored. + +## Operator commands + +```powershell +python -m memorymaster.profile status --db .\memorymaster.db +python -m memorymaster.profile run --db .\memorymaster.db --workspace . --force +``` + +## Acceptance evidence + +- Focused engine tests cover incremental extraction, strict support validation, + resumability, exact support lineage, independent-session gating, preference + expiry, stable-fact retention, and deterministic budget bounds. +- Surface tests cover feature-off scheduling, fail-closed enabled scheduling, + generated-only SessionStart loading, and CLI status/help. +- Activation, a public package release, and historical transcript bootstrap are + separate operator actions; this implementation does not perform them. diff --git a/DOCS-MAP.md b/DOCS-MAP.md index 74b190a..6f29280 100644 --- a/DOCS-MAP.md +++ b/DOCS-MAP.md @@ -1,10 +1,10 @@ - + # DOCS-MAP - memorymaster # Covers: trust verdicts and replacements for every canonical documentation surface. -# Key terms: CURRENT, SUPERSEDED, ABANDONED, GENERATED, roadmap, paper radar, ADR. +# Key terms: CURRENT, SUPERSEDED, compiled user profile, roadmap, ADR. # Read when: locating authoritative project documentation before reading doc bodies. -# Updated: 2026-08-12 after PR #189 merge, governed runtime deploy, and PPR-7 authorization. -# Rule: PPR-7 is isolated and opt-in; public release and automatic recall remain separately gated. +# Updated: 2026-08-12 for the evidence-bound compiled user-profile implementation. +# Rule: generated profiles are disposable projections; MemoryMaster remains authority. | File | Verdict | Last change | Reason | @@ -12,6 +12,7 @@ | CHANGELOG.md | CURRENT | 2026-08-04 | Public release history; v4.6.0 records governed universal capture, measured quality changes, security evidence, and known follow-ups. | | ROADMAP.md | CURRENT | 2026-08-12 | Sole authoritative roadmap; P5 is merged and locally deployed, while isolated PPR-7 implementation is active. | | .planning/GRAPH-OBSERVATIONS-V1.md | CURRENT | 2026-08-12 | Bounded PPR-7 implementation ledger for deterministic supported components, governed synthesis, lifecycle, opt-in recall, and verification. | +| .planning/COMPILED-USER-PROFILE-V1.md | CURRENT | 2026-08-12 | Implementation contract for the zero-curation, evidence-bound user profile projection and weekly GLM map/reduce job. | | .planning/PAPER-RADAR-REVIEW-2026-08-08.md | CURRENT | 2026-08-08 | Primary-paper ledger covers 57-paper triage, 18 deep reviews, exact MemoryMaster gaps, and ordered PPR-1 through PPR-6 decisions subordinate to ROADMAP.md. | | .planning/PAPER-RESEARCH-IMPLEMENTATION-2026-08-08.md | CURRENT | 2026-08-08 | Executable status ledger for PPR-1 through PPR-6; records acceptance criteria and evidence without competing with ROADMAP.md. | | .planning/HERMES-SCOPE-SKILLS-INTEGRATION-2026-08-07.md | CURRENT | 2026-08-12 | Executable Tencent-derived ledger; repair5 passed and PR #189 is open without merge authority. | diff --git a/ROADMAP.md b/ROADMAP.md index 3a6a112..8d8d9c0 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -1,10 +1,10 @@ - + # MemoryMaster roadmap -# Covers: post-v4.6 sequence, Tencent-derived work, paper research, and deferrals. -# Key terms: Hermes, governed skills, paper radar, temporal projection, sustainability. +# Covers: post-v4.6 sequence, governed observations, compiled profile, and deferrals. +# Key terms: Hermes, governed skills, graph observations, user profile, sustainability. # Read when: choosing release scope, accepting a feature, or checking deferrals. # Authority: sole roadmap; planning ledgers implement it and never replace it. -# Safety: SQLite authority and steward promotion remain fixed; PPR-7 stays additive, opt-in, and unreleased pending its own evidence. +# Safety: SQLite remains authority; generated user.md is disposable and feature-off by default. ## Shipped in v4.6.0 @@ -68,6 +68,10 @@ its governed-claims authority: opt-in; its separate PR/CI, feature-off deployment, and rollout evidence are still required before activation. No public release or automatic ordinary-recall inclusion is implied. +- The compiled user profile is being implemented as a disposable projection of + sanitized transcript evidence. Weekly bounded GLM map/reduce work stores + exact supports in SQLite; deterministic rendering produces `user.md` for + SessionStart injection. The feature remains off until explicitly enabled. - The invalid earlier window remains incident evidence only because it included a VM OOM/gateway interruption and did not contain P5. - Keep v4.6.0 operational while the post-release Obsidian opt-in and OpenCode @@ -91,6 +95,9 @@ its governed-claims authority: exact support signatures, deterministic discovery, candidate-only synthesis, observation-specific steward validation, and explicit recall inclusion are implemented with local evidence, while PR/CI and feature-off rollout remain. +- Integrate the bounded compiled-profile implementation described in + `.planning/COMPILED-USER-PROFILE-V1.md`; then perform a one-time local run on + existing sanitized transcript rows before enabling its weekly task. - Improve personal/local backup guidance beyond the already verified disposable backup/restore and migration procedure. - Keep semantic recall optional and disabled unless a local user deliberately diff --git a/docs/generated/release-truth.json b/docs/generated/release-truth.json index e8ccf09..4c79af5 100644 --- a/docs/generated/release-truth.json +++ b/docs/generated/release-truth.json @@ -134,7 +134,7 @@ "console_entrypoints": 8, "mcp_tools": 50, "ops_cli_commands": 5, - "pytest_test_functions": 3781 + "pytest_test_functions": 3793 }, "feature_profile_matrix": { "capture_hook": [ diff --git a/docs/generated/release-truth.md b/docs/generated/release-truth.md index 3a418e6..46e53e7 100644 --- a/docs/generated/release-truth.md +++ b/docs/generated/release-truth.md @@ -7,7 +7,7 @@ Do not edit this file by hand. Run `python scripts/generate_release_truth.py`. - Main CLI commands: **118** - Operations CLI commands: **5** - Console entrypoints: **8** -- Pytest source test functions: **3781** +- Pytest source test functions: **3793** ## MCP tools diff --git a/memorymaster/config_templates/hooks/memorymaster-session-start.py b/memorymaster/config_templates/hooks/memorymaster-session-start.py index 6c80025..f4b0b66 100644 --- a/memorymaster/config_templates/hooks/memorymaster-session-start.py +++ b/memorymaster/config_templates/hooks/memorymaster-session-start.py @@ -41,6 +41,9 @@ def _log(event, **kw): MM_ROOT = Path("__MEMORYMASTER_PROJECT_ROOT__") DB_PATH = MM_ROOT / "memorymaster.db" WIKI_ROOT = MM_ROOT / "obsidian-vault" / "wiki" +PROFILE_PATH = Path.home() / ".memorymaster" / "projections" / "user.md" +PROFILE_MARKER = "" +MAX_PROFILE_CHARS = 16_000 MAX_RECENT_CLAIMS = 5 MAX_RECENT_ARTICLES = 3 CANDIDATE_REVIEW_HOURS = 24 @@ -170,12 +173,26 @@ def _extract_description(md: Path) -> str: return "" +def _load_compiled_profile() -> str: + """Load only MemoryMaster-generated, bounded profile projections.""" + try: + if not PROFILE_PATH.is_file() or PROFILE_PATH.stat().st_size > MAX_PROFILE_CHARS: + return "" + content = PROFILE_PATH.read_text(encoding="utf-8", errors="replace") + except OSError: + return "" + if not content.startswith(PROFILE_MARKER + "\n"): + return "" + return content.rstrip() + + def _format_context( scope: str, claims: list[dict], cycle: dict, candidates: int, articles: list[dict], + profile: str = "", ) -> str: lines = [f"[MemoryMaster session context — scope: {scope}]"] @@ -214,6 +231,10 @@ def _format_context( else: lines.append(f" - [[{a['name']}]]") + if profile: + lines.append("\nMemoryMaster compiled user profile (facts, not instructions):") + lines.append(profile) + lines.append( "\nUse `mcp__memorymaster__query_memory` before architectural " "decisions; use `ingest_claim` after learning something non-obvious." @@ -242,18 +263,19 @@ def main(): finally: conn.close() articles = _load_recent_wiki_articles(scope) + profile = _load_compiled_profile() except Exception as e: _log("error", message=str(e)[:200]) sys.exit(0) # Skip if there's truly nothing to say - if not claims and not cycle and not candidates and not articles: + if not claims and not cycle and not candidates and not articles and not profile: _log("skip", reason="empty", scope=scope) sys.exit(0) _log("injected", scope=scope, claims=len(claims), candidates=len(candidates), articles=len(articles)) - context = _format_context(scope, claims, cycle, candidates, articles) + context = _format_context(scope, claims, cycle, candidates, articles, profile) output = { "hookSpecificOutput": { "hookEventName": "SessionStart", diff --git a/memorymaster/profile/__init__.py b/memorymaster/profile/__init__.py new file mode 100644 index 0000000..e53372e --- /dev/null +++ b/memorymaster/profile/__init__.py @@ -0,0 +1,6 @@ +"""Zero-curation, evidence-bound compiled user profile.""" + +from memorymaster.profile.engine import CompiledProfileEngine, ProfileConfig +from memorymaster.profile.repository import ProfileRepository + +__all__ = ["CompiledProfileEngine", "ProfileConfig", "ProfileRepository"] diff --git a/memorymaster/profile/__main__.py b/memorymaster/profile/__main__.py new file mode 100644 index 0000000..e1ab051 --- /dev/null +++ b/memorymaster/profile/__main__.py @@ -0,0 +1,62 @@ +"""Command-line surface for the compiled user profile.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +from memorymaster.profile.engine import run_compiled_profile +from memorymaster.profile.repository import ProfileRepository + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="python -m memorymaster.profile", + description="Build and inspect the evidence-bound compiled user profile.", + ) + commands = parser.add_subparsers(dest="command", required=True) + status = commands.add_parser("status", help="Show current compiled user profile state") + status.add_argument("--db", required=True) + run = commands.add_parser("run", help="Run or resume profile compilation") + run.add_argument("--db", required=True) + run.add_argument("--workspace", default=".") + run.add_argument("--output-dir", default="") + run.add_argument("--force", action="store_true") + run.add_argument("--max-map-calls", type=int) + return parser + + +def _status(db_path: Path) -> dict[str, object]: + if not db_path.is_file(): + return {"active_run": None, "facts": 0, "latest_completed_run": None} + repository = ProfileRepository(db_path) + active = repository.active_run() + latest = repository.latest_completed_run() + return { + "active_run": int(active["id"]) if active else None, + "facts": len(repository.active_facts()), + "latest_completed_run": int(latest["id"]) if latest else None, + } + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + if args.command == "status": + result = _status(Path(args.db)) + else: + from memorymaster.core.service import MemoryService + + MemoryService(args.db, workspace_root=Path(args.workspace)).init_db() + result = run_compiled_profile( + args.db, + output_dir=args.output_dir or None, + force=bool(args.force), + max_map_calls=args.max_map_calls, + ) + print(json.dumps(result, ensure_ascii=False, sort_keys=True)) + return 0 if result.get("ok", True) else 1 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/memorymaster/profile/engine.py b/memorymaster/profile/engine.py new file mode 100644 index 0000000..f968db4 --- /dev/null +++ b/memorymaster/profile/engine.py @@ -0,0 +1,217 @@ +"""Resumable weekly map/reduce engine for the compiled user profile.""" + +from __future__ import annotations + +import hashlib +import json +import os +from dataclasses import asdict, dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any, Callable, Protocol + +from memorymaster.profile.models import ProfileCandidate, ProfileDecision, ProfileFact, ProfileMessage +from memorymaster.profile.renderer import render_profile +from memorymaster.profile.repository import ProfileRepository + + +class ProfileMapper(Protocol): + model: str + + def map(self, messages: tuple[ProfileMessage, ...]) -> tuple[ProfileCandidate, ...]: ... + + +class ProfileReducer(Protocol): + model: str + + def reduce( + self, candidates: tuple[ProfileCandidate, ...], facts: tuple[ProfileFact, ...] + ) -> tuple[ProfileDecision, ...]: ... + + +@dataclass(frozen=True, slots=True) +class ProfileConfig: + cadence_days: int = 7 + max_map_calls: int = 3 + max_messages: int = 1000 + max_input_chars: int = 500_000 + min_independent_sessions: int = 2 + preference_ttl_days: int = 90 + token_budget: int = 800 + max_facts: int = 40 + + @classmethod + def from_env(cls) -> "ProfileConfig": + return cls( + cadence_days=_env_int("MEMORYMASTER_PROFILE_CADENCE_DAYS", 7), + max_map_calls=_env_int("MEMORYMASTER_PROFILE_MAX_MAP_CALLS", 3), + max_messages=_env_int("MEMORYMASTER_PROFILE_MAX_MESSAGES", 1000), + max_input_chars=_env_int("MEMORYMASTER_PROFILE_MAX_INPUT_CHARS", 500_000), + min_independent_sessions=_env_int("MEMORYMASTER_PROFILE_MIN_SESSIONS", 2), + preference_ttl_days=_env_int("MEMORYMASTER_PROFILE_PREFERENCE_TTL_DAYS", 90), + token_budget=_env_int("MEMORYMASTER_PROFILE_TOKEN_BUDGET", 800), + max_facts=_env_int("MEMORYMASTER_PROFILE_MAX_FACTS", 40), + ) + + +def _env_int(name: str, default: int) -> int: + try: + return max(1, int(os.environ.get(name, str(default)))) + except ValueError: + return default + + +class CompiledProfileEngine: + def __init__( + self, + repository: ProfileRepository, + mapper: ProfileMapper, + reducer: ProfileReducer, + *, + output_dir: str | Path, + config: ProfileConfig | None = None, + now: Callable[[], datetime] | None = None, + ) -> None: + self.repo = repository + self.mapper = mapper + self.reducer = reducer + self.output_dir = Path(output_dir) + self.config = config or ProfileConfig.from_env() + self.now = now or (lambda: datetime.now(timezone.utc)) + + def run(self, *, force: bool = False, max_map_calls: int | None = None) -> dict[str, Any]: + current = self.now() + active = self.repo.active_run() + if active is None and not force and not self.repo.due( + now=current, cadence_days=self.config.cadence_days + ): + return {"ok": True, "status": "not_due"} + run = active or self._start_run(current) + if run is None: + rendered = self._write_projection(current) + return {"ok": True, "status": "no_changes", "facts": len(rendered.fact_ids)} + try: + result = self._advance_mapping(run, current, max_map_calls) + if result is not None: + return result + return self._reduce_and_complete(int(run["id"]), current) + except Exception as exc: + self.repo.record_error(int(run["id"]), type(exc).__name__, now=current) + return {"ok": False, "status": str(run["status"]), "error": type(exc).__name__} + + def _start_run(self, now: datetime) -> dict[str, Any] | None: + target = self.repo.max_user_id() + latest = self.repo.latest_completed_run() + start = int(latest["target_watermark"]) if latest else 0 + if target <= start: + return None + return self.repo.start_run( + target=target, + map_model=self.mapper.model, + reduce_model=self.reducer.model, + now=now, + ) + + def _advance_mapping( + self, run: dict[str, Any], now: datetime, max_map_calls: int | None + ) -> dict[str, Any] | None: + limit = max_map_calls if max_map_calls is not None else self.config.max_map_calls + calls = 0 + while int(run["current_watermark"]) < int(run["target_watermark"]): + batch = self.repo.message_batch( + after_id=int(run["current_watermark"]), + through_id=int(run["target_watermark"]), + max_messages=self.config.max_messages, + max_chars=self.config.max_input_chars, + ) + provider_called = bool(batch.messages) + if provider_called and calls >= max(1, limit): + break + candidates = self.mapper.map(batch.messages) if provider_called else () + calls += int(provider_called) + self.repo.save_mapping( + int(run["id"]), candidates, batch.scanned_through_id, + now=now, provider_called=provider_called, + ) + run = self.repo.run(int(run["id"])) + if batch.scanned_through_id <= int(run["start_watermark"]): + break + if int(run["current_watermark"]) < int(run["target_watermark"]): + return {"ok": True, "status": "mapping", "run_id": int(run["id"]), "map_calls": calls} + self.repo.mark_reducing(int(run["id"]), now=now) + return None + + def _reduce_and_complete(self, run_id: int, now: datetime) -> dict[str, Any]: + candidates = self.repo.candidates(run_id) + facts = self.repo.active_facts() + decisions = self.reducer.reduce(candidates, facts) if candidates else () + stats = self.repo.apply_decisions( + run_id, + decisions, + now=now, + min_sessions=self.config.min_independent_sessions, + ) + expired = self.repo.expire_preferences( + now=now, ttl_days=self.config.preference_ttl_days + ) + rendered = self._write_projection(now) + output_hash = hashlib.sha256(rendered.markdown.encode("utf-8")).hexdigest() + self.repo.complete_run(run_id, output_hash, now=now) + return { + "ok": True, + "status": "completed", + "run_id": run_id, + "facts": len(rendered.fact_ids), + "applied": stats["applied"], + "rejected": stats["rejected"], + "expired": expired, + } + + def _write_projection(self, now: datetime): + facts = self.repo.active_facts() + rendered = render_profile( + facts, + token_budget=self.config.token_budget, + max_facts=self.config.max_facts, + ) + selected = {fact.fact_id: fact for fact in facts if fact.fact_id in rendered.fact_ids} + manifest = { + "schema": "memorymaster.compiled-profile.v1", + "generated_at": now.isoformat(), + "facts": [asdict(selected[fact_id]) for fact_id in rendered.fact_ids], + } + self.output_dir.mkdir(parents=True, exist_ok=True) + self._atomic_write(self.output_dir / "user.md", rendered.markdown) + self._atomic_write( + self.output_dir / "user-profile.json", + json.dumps(manifest, ensure_ascii=False, sort_keys=True, indent=2) + "\n", + ) + return rendered + + @staticmethod + def _atomic_write(path: Path, content: str) -> None: + temporary = path.with_suffix(path.suffix + ".tmp") + temporary.write_text(content, encoding="utf-8") + os.replace(temporary, path) + + +def run_compiled_profile( + db_path: str | Path, + *, + output_dir: str | Path | None = None, + force: bool = False, + max_map_calls: int | None = None, +) -> dict[str, Any]: + from memorymaster.profile.providers import GLMProfileMapper, GLMProfileReducer + + directory = output_dir or Path.home() / ".memorymaster" / "projections" + engine = CompiledProfileEngine( + ProfileRepository(db_path), + GLMProfileMapper(), + GLMProfileReducer(), + output_dir=directory, + ) + return engine.run(force=force, max_map_calls=max_map_calls) + + +__all__ = ["CompiledProfileEngine", "ProfileConfig", "run_compiled_profile"] diff --git a/memorymaster/profile/models.py b/memorymaster/profile/models.py new file mode 100644 index 0000000..698aa2f --- /dev/null +++ b/memorymaster/profile/models.py @@ -0,0 +1,141 @@ +"""Strict data contracts for compiled profile extraction and rendering.""" + +from __future__ import annotations + +import re +from dataclasses import dataclass + +from memorymaster.core.security import scan_persisted_value + + +PROFILE_SCHEMA_VERSION = "memorymaster.compiled-profile.v1" +PROFILE_ALGORITHM_VERSION = "compiled-profile-map-reduce-v1" +PROFILE_CATEGORIES = ( + "identity_locale", + "products_systems", + "working_style", + "standing_constraints", +) +PREDICATE_CATEGORY = { + "location": "identity_locale", + "timezone": "identity_locale", + "currency": "identity_locale", + "language": "identity_locale", + "role": "identity_locale", + "operates_product": "products_systems", + "maintains_system": "products_systems", + "works_in_domain": "products_systems", + "primary_stack": "products_systems", + "communication_style": "working_style", + "workflow_preference": "working_style", + "verification_preference": "working_style", + "autonomy_preference": "working_style", + "manual_work_constraint": "standing_constraints", + "security_constraint": "standing_constraints", + "deployment_constraint": "standing_constraints", + "tooling_constraint": "standing_constraints", +} +PROFILE_VOLATILITIES = ("stable", "preference") +PROFILE_ACTIONS = ("add", "reinforce", "replace", "ignore") +_INSTRUCTION = re.compile( + r"(?i)\b(?:you|assistant|agents?|claude|codex)\b.{0,50}\b(?:must|should|always|never)\b" +) + + +class ProfileValidationError(ValueError): + """Profile provider output failed a deterministic boundary.""" + + +def validate_fact_fields(category: str, predicate: str, value: str, volatility: str) -> None: + if category not in PROFILE_CATEGORIES: + raise ProfileValidationError("unknown profile category") + if PREDICATE_CATEGORY.get(predicate) != category: + raise ProfileValidationError("unknown or mismatched profile predicate") + if volatility not in PROFILE_VOLATILITIES: + raise ProfileValidationError("unknown profile volatility") + if not value.strip() or len(value.strip()) > 240 or "\n" in value: + raise ProfileValidationError("profile value is malformed or oversized") + if _INSTRUCTION.search(value): + raise ProfileValidationError("profile value is instruction-shaped") + if scan_persisted_value(value): + raise ProfileValidationError("profile value contains sensitive material") + + +@dataclass(frozen=True, slots=True) +class ProfileMessage: + message_id: int + session_id: str + scope: str + text: str + assistant_context: str + + +@dataclass(frozen=True, slots=True) +class ProfileCandidate: + candidate_id: str + category: str + predicate: str + value: str + volatility: str + support_ids: tuple[int, ...] + + def __post_init__(self) -> None: + validate_fact_fields(self.category, self.predicate, self.value, self.volatility) + if not self.candidate_id or len(self.candidate_id) > 120: + raise ProfileValidationError("candidate id is malformed") + if not self.support_ids or len(self.support_ids) > 20: + raise ProfileValidationError("candidate supports are malformed") + if len(set(self.support_ids)) != len(self.support_ids): + raise ProfileValidationError("candidate supports contain duplicates") + + +@dataclass(frozen=True, slots=True) +class ProfileDecision: + candidate_ids: tuple[str, ...] + action: str + category: str = "" + predicate: str = "" + value: str = "" + volatility: str = "stable" + target_fact_id: int | None = None + confidence: float = 0.0 + rationale: str = "" + + def __post_init__(self) -> None: + if self.action not in PROFILE_ACTIONS or not self.candidate_ids: + raise ProfileValidationError("profile decision action or candidates are invalid") + if self.action in {"add", "replace"}: + validate_fact_fields(self.category, self.predicate, self.value, self.volatility) + if self.action in {"reinforce", "replace"} and not self.target_fact_id: + raise ProfileValidationError("profile decision target is required") + if not 0.0 <= float(self.confidence) <= 1.0 or len(self.rationale) > 500: + raise ProfileValidationError("profile decision metadata is invalid") + + +@dataclass(frozen=True, slots=True) +class ProfileFact: + fact_id: int + category: str + predicate: str + value: str + volatility: str + status: str + support_hash: str + support_count: int + independent_sessions: int + first_seen_at: str + last_supported_at: str + support_ids: tuple[int, ...] = () + + +@dataclass(frozen=True, slots=True) +class MessageBatch: + messages: tuple[ProfileMessage, ...] + scanned_through_id: int + + +@dataclass(frozen=True, slots=True) +class RenderedProfile: + markdown: str + fact_ids: tuple[int, ...] + tokens_used: int diff --git a/memorymaster/profile/providers.py b/memorymaster/profile/providers.py new file mode 100644 index 0000000..52d762d --- /dev/null +++ b/memorymaster/profile/providers.py @@ -0,0 +1,211 @@ +"""Structured GLM map/reduce providers for compiled profile facts.""" + +from __future__ import annotations + +import json +import os +from dataclasses import asdict +from pathlib import Path +from typing import Any + +from memorymaster.core.opencode_client import OpenCodeClient +from memorymaster.profile.models import ( + PREDICATE_CATEGORY, + PROFILE_CATEGORIES, + PROFILE_VOLATILITIES, + ProfileCandidate, + ProfileDecision, + ProfileFact, + ProfileMessage, + ProfileValidationError, +) + + +def _json_object(raw: str) -> dict[str, Any]: + text = raw.strip() + if text.startswith("```"): + lines = text.splitlines() + text = "\n".join(lines[1:-1]).strip() + try: + parsed = json.loads(text) + except json.JSONDecodeError as exc: + raise ProfileValidationError("profile provider returned malformed JSON") from exc + if not isinstance(parsed, dict): + raise ProfileValidationError("profile provider output must be an object") + return parsed + + +def _string(row: dict[str, Any], key: str, *, max_length: int = 500) -> str: + value = row.get(key) + if not isinstance(value, str) or len(value) > max_length: + raise ProfileValidationError(f"profile field {key} is invalid") + return value.strip() + + +def parse_map_output( + raw: str, messages: tuple[ProfileMessage, ...] +) -> tuple[ProfileCandidate, ...]: + rows = _json_object(raw).get("candidates") + if not isinstance(rows, list) or len(rows) > 80: + raise ProfileValidationError("profile candidates must be a bounded array") + known_supports = {message.message_id for message in messages} + candidates: list[ProfileCandidate] = [] + seen_ids: set[str] = set() + for row in rows: + if not isinstance(row, dict): + raise ProfileValidationError("profile candidate must be an object") + candidate = _candidate_from_row(row) + if not set(candidate.support_ids).issubset(known_supports): + raise ProfileValidationError("profile candidate has unknown support") + if candidate.candidate_id in seen_ids: + raise ProfileValidationError("profile candidate id is duplicated") + seen_ids.add(candidate.candidate_id) + candidates.append(candidate) + return tuple(candidates) + + +def _candidate_from_row(row: dict[str, Any]) -> ProfileCandidate: + support_ids = row.get("support_ids") + if not isinstance(support_ids, list) or not all(isinstance(item, int) for item in support_ids): + raise ProfileValidationError("profile candidate supports are invalid") + return ProfileCandidate( + candidate_id=_string(row, "candidate_id", max_length=120), + category=_string(row, "category", max_length=40), + predicate=_string(row, "predicate", max_length=80), + value=_string(row, "value", max_length=240), + volatility=_string(row, "volatility", max_length=20), + support_ids=tuple(support_ids), + ) + + +def parse_reduce_output( + raw: str, + candidates: tuple[ProfileCandidate, ...], + facts: tuple[ProfileFact, ...], +) -> tuple[ProfileDecision, ...]: + rows = _json_object(raw).get("decisions") + if not isinstance(rows, list) or len(rows) > max(1, len(candidates)): + raise ProfileValidationError("profile decisions must be a bounded array") + known_candidates = {candidate.candidate_id for candidate in candidates} + known_facts = {fact.fact_id for fact in facts} + decisions = tuple(_decision_from_row(row, known_facts) for row in rows) + consumed = [item for decision in decisions for item in decision.candidate_ids] + if len(consumed) != len(set(consumed)) or set(consumed) != known_candidates: + raise ProfileValidationError("profile candidates must appear exactly once") + return decisions + + +def _decision_from_row(row: Any, known_facts: set[int]) -> ProfileDecision: + if not isinstance(row, dict): + raise ProfileValidationError("profile decision must be an object") + candidate_ids = row.get("candidate_ids") + if not isinstance(candidate_ids, list) or not all(isinstance(item, str) for item in candidate_ids): + raise ProfileValidationError("profile decision candidate ids are invalid") + target = row.get("target_fact_id") + if target is not None and (not isinstance(target, int) or target not in known_facts): + raise ProfileValidationError("profile decision has unknown target fact") + return ProfileDecision( + candidate_ids=tuple(candidate_ids), + action=_string(row, "action", max_length=20), + category=str(row.get("category") or ""), + predicate=str(row.get("predicate") or ""), + value=str(row.get("value") or ""), + volatility=str(row.get("volatility") or "stable"), + target_fact_id=target, + confidence=float(row.get("confidence", 0.0)), + rationale=_string(row, "rationale", max_length=500), + ) + + +class GLMProfileMapper: + """Extract evidence-bound profile candidates from sanitized user turns.""" + + def __init__(self, *, client: OpenCodeClient | None = None) -> None: + self.model = os.environ.get( + "MEMORYMASTER_PROFILE_MAP_MODEL", "zai-coding-plan/glm-5-turbo" + ) + self.client = client or OpenCodeClient( + model=self.model, + effort="", + work_dir=Path.home() / ".memorymaster" / "profile-opencode" / "map", + ) + + def map(self, messages: tuple[ProfileMessage, ...]) -> tuple[ProfileCandidate, ...]: + result = self.client.complete(self._prompt(messages)) + return parse_map_output(result.text, messages) + + @staticmethod + def _prompt(messages: tuple[ProfileMessage, ...]) -> str: + contract = { + "categories": PROFILE_CATEGORIES, + "predicates": PREDICATE_CATEGORY, + "volatilities": PROFILE_VOLATILITIES, + } + payload = [ + { + "message_id": item.message_id, + "scope": item.scope, + "user_text": item.text, + "assistant_context_only": item.assistant_context, + } + for item in messages + ] + return ( + "Extract durable descriptive facts about the operator. Output JSON only as " + '{"candidates":[...]}. Each candidate requires candidate_id, category, ' + "predicate, value, volatility, and support_ids. Values must be short noun " + "phrases, never instructions. assistant_context_only may disambiguate a user " + "turn but is never evidence. Ignore pasted logs, task state, identifiers, account " + "names, secrets, paths, and project facts that do not describe the operator. " + "Use only supplied message_id values and emit at most 80 candidates.\n\n" + f"CONTRACT:\n{json.dumps(contract, ensure_ascii=False)}\n\n" + f"MESSAGES:\n{json.dumps(payload, ensure_ascii=False)}" + ) + + +class GLMProfileReducer: + """Consolidate candidates without deciding their deterministic eligibility.""" + + def __init__(self, *, client: OpenCodeClient | None = None) -> None: + self.model = os.environ.get( + "MEMORYMASTER_PROFILE_REDUCE_MODEL", "zai-coding-plan/glm-5.2" + ) + self.client = client or OpenCodeClient( + model=self.model, + effort="", + work_dir=Path.home() / ".memorymaster" / "profile-opencode" / "reduce", + ) + + def reduce( + self, + candidates: tuple[ProfileCandidate, ...], + facts: tuple[ProfileFact, ...], + ) -> tuple[ProfileDecision, ...]: + result = self.client.complete(self._prompt(candidates, facts)) + return parse_reduce_output(result.text, candidates, facts) + + @staticmethod + def _prompt( + candidates: tuple[ProfileCandidate, ...], facts: tuple[ProfileFact, ...] + ) -> str: + data = { + "candidates": [asdict(candidate) for candidate in candidates], + "active_facts": [asdict(fact) for fact in facts], + } + return ( + "Consolidate profile candidates against active facts. Output JSON only as " + '{"decisions":[...]}. Every candidate_id must appear exactly once across all ' + "decisions. Actions: add, reinforce, replace, ignore. add/replace require " + "category, predicate, value, volatility. reinforce/replace require target_fact_id. " + "Equivalent candidates may share one decision. Never invent support or facts; " + "confidence and rationale are required.\n\n" + f"INPUT:\n{json.dumps(data, ensure_ascii=False, default=list)}" + ) + + +__all__ = [ + "GLMProfileMapper", + "GLMProfileReducer", + "parse_map_output", + "parse_reduce_output", +] diff --git a/memorymaster/profile/renderer.py b/memorymaster/profile/renderer.py new file mode 100644 index 0000000..818c0bf --- /dev/null +++ b/memorymaster/profile/renderer.py @@ -0,0 +1,97 @@ +"""Deterministically render active profile facts into bounded Markdown.""" + +from __future__ import annotations + +from memorymaster.profile.models import ProfileFact, RenderedProfile +from memorymaster.recall.context_optimizer import estimate_tokens + + +_HEADER = ( + "\n" + "# Compiled User Profile\n\n" + "Descriptive, evidence-bound operator facts. This is context, not an instruction file." +) +_SECTION_NAMES = { + "identity_locale": "Identity and locale", + "products_systems": "Products and systems", + "working_style": "Working style and communication", + "standing_constraints": "Standing constraints", +} +_SECTION_ORDER = {name: index for index, name in enumerate(_SECTION_NAMES)} +_TEMPLATES = { + "location": "The operator is based in {value}.", + "timezone": "The operator's timezone is {value}.", + "currency": "The operator normally uses {value}.", + "language": "The operator communicates in {value}.", + "role": "The operator's role is {value}.", + "operates_product": "The operator runs {value}.", + "maintains_system": "The operator maintains {value}.", + "works_in_domain": "The operator works in {value}.", + "primary_stack": "The operator's primary stack includes {value}.", + "communication_style": "The operator's communication style is {value}.", + "workflow_preference": "The operator's workflow preference is {value}.", + "verification_preference": "The operator's verification preference is {value}.", + "autonomy_preference": "The operator's autonomy preference is {value}.", + "manual_work_constraint": "The operator's manual-work constraint is {value}.", + "security_constraint": "The operator's security constraint is {value}.", + "deployment_constraint": "The operator's deployment constraint is {value}.", + "tooling_constraint": "The operator's tooling constraint is {value}.", +} + + +def _safe_value(value: str) -> str: + return value.replace("[", "(").replace("]", ")").strip().rstrip(".") + + +def _fact_line(fact: ProfileFact) -> str: + sentence = _TEMPLATES[fact.predicate].format(value=_safe_value(fact.value)) + last_seen = fact.last_supported_at[:10] + return f"- {sentence} [support={fact.support_count}; last={last_seen}]" + + +def _sorted_facts(facts: tuple[ProfileFact, ...]) -> list[ProfileFact]: + return sorted( + (fact for fact in facts if fact.status == "active"), + key=lambda fact: ( + _SECTION_ORDER[fact.category], + -fact.support_count, + fact.predicate, + fact.value.casefold(), + fact.fact_id, + ), + ) + + +def _markdown(selected: list[ProfileFact]) -> str: + lines = [_HEADER] + for category in _SECTION_NAMES: + section = [fact for fact in selected if fact.category == category] + if not section: + continue + lines.extend(["", f"## {_SECTION_NAMES[category]}"]) + lines.extend(_fact_line(fact) for fact in section) + lines.extend(["", "Support counts summarize exact source IDs retained in the profile manifest."]) + return "\n".join(lines).strip() + "\n" + + +def render_profile( + facts: tuple[ProfileFact, ...], *, token_budget: int = 800, max_facts: int = 40 +) -> RenderedProfile: + if token_budget < estimate_tokens(_HEADER): + raise ValueError("compiled profile token budget is too small") + selected: list[ProfileFact] = [] + for fact in _sorted_facts(facts): + if len(selected) >= max(1, max_facts): + break + candidate = [*selected, fact] + if estimate_tokens(_markdown(candidate)) <= token_budget: + selected = candidate + markdown = _markdown(selected) + return RenderedProfile( + markdown=markdown, + fact_ids=tuple(fact.fact_id for fact in selected), + tokens_used=estimate_tokens(markdown), + ) + + +__all__ = ["render_profile"] diff --git a/memorymaster/profile/repository.py b/memorymaster/profile/repository.py new file mode 100644 index 0000000..ae7dde2 --- /dev/null +++ b/memorymaster/profile/repository.py @@ -0,0 +1,567 @@ +"""SQLite repository for resumable compiled-profile projection work.""" + +from __future__ import annotations + +import hashlib +import json +import re +import sqlite3 +from contextlib import closing +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any, Iterable + +from memorymaster.core.security import scan_persisted_value +from memorymaster.profile.models import ( + MessageBatch, + ProfileCandidate, + ProfileDecision, + ProfileFact, + ProfileMessage, +) +from memorymaster.stores._storage_shared import open_conn + + +_WRAPPER = re.compile( + r"(?is)^\s*(?: None: + self.db_path = str(db_path) + + def connect(self) -> sqlite3.Connection: + return open_conn(self.db_path) + + def max_user_id(self) -> int: + with closing(self.connect()) as conn: + if not self._has_table(conn, "verbatim_memories"): + return 0 + row = conn.execute( + "SELECT COALESCE(MAX(id), 0) FROM verbatim_memories WHERE role='user'" + ).fetchone() + return int(row[0]) if row else 0 + + def active_run(self) -> dict[str, Any] | None: + with closing(self.connect()) as conn: + row = conn.execute( + "SELECT * FROM compiled_profile_runs WHERE active_slot=1 LIMIT 1" + ).fetchone() + return dict(row) if row else None + + def latest_completed_run(self) -> dict[str, Any] | None: + with closing(self.connect()) as conn: + row = conn.execute( + """SELECT * FROM compiled_profile_runs WHERE status='completed' + ORDER BY id DESC LIMIT 1""" + ).fetchone() + return dict(row) if row else None + + def due(self, *, now: datetime, cadence_days: int) -> bool: + latest = self.latest_completed_run() + if latest is None or not latest.get("completed_at"): + return True + completed = datetime.fromisoformat(str(latest["completed_at"])) + return completed <= now - timedelta(days=max(1, cadence_days)) + + def start_run( + self, *, target: int, map_model: str, reduce_model: str, now: datetime + ) -> dict[str, Any]: + active = self.active_run() + if active is not None: + return active + latest = self.latest_completed_run() + start = int(latest["target_watermark"]) if latest else 0 + timestamp = now.isoformat() + with closing(self.connect()) as conn: + cur = conn.execute( + """INSERT INTO compiled_profile_runs + (status, active_slot, start_watermark, current_watermark, + target_watermark, map_model, reduce_model, started_at, updated_at) + VALUES ('mapping',1,?,?,?,?,?,?,?)""", + (start, start, target, map_model, reduce_model, timestamp, timestamp), + ) + conn.commit() + run_id = int(cur.lastrowid) + return self.run(run_id) + + def run(self, run_id: int) -> dict[str, Any]: + with closing(self.connect()) as conn: + row = conn.execute( + "SELECT * FROM compiled_profile_runs WHERE id=?", (run_id,) + ).fetchone() + if row is None: + raise KeyError(f"compiled profile run {run_id} does not exist") + return dict(row) + + def message_batch( + self, + *, + after_id: int, + through_id: int, + max_messages: int, + max_chars: int, + ) -> MessageBatch: + rows = self._message_rows(after_id, through_id, max_messages) + messages: list[ProfileMessage] = [] + scanned = after_id + used = 0 + for row in rows: + message = self._profile_message(row) + if message is None: + scanned = int(row["id"]) + continue + size = len(message.text) + len(message.assistant_context) + if messages and used + size > max_chars: + break + messages.append(message) + used += size + scanned = int(row["id"]) + if len(messages) >= max_messages: + break + if not rows: + scanned = through_id + return MessageBatch(tuple(messages), scanned) + + def _message_rows( + self, after_id: int, through_id: int, max_messages: int + ) -> list[sqlite3.Row]: + fetch_limit = max(100, max_messages * 10) + with closing(self.connect()) as conn: + if not self._has_table(conn, "verbatim_memories"): + return [] + return conn.execute( + """SELECT v.id, v.session_id, v.scope, v.content, + COALESCE((SELECT a.content FROM verbatim_memories a + WHERE a.session_id=v.session_id AND a.id? AND v.id<=? + ORDER BY v.id LIMIT ?""", + (after_id, through_id, fetch_limit), + ).fetchall() + + @staticmethod + def _profile_message(row: sqlite3.Row) -> ProfileMessage | None: + text = str(row["content"] or "").strip() + context = str(row["context"] or "")[-400:] + if not text or _WRAPPER.search(text) or scan_persisted_value({"text": text, "context": context}): + return None + return ProfileMessage( + int(row["id"]), + str(row["session_id"]), + str(row["scope"]), + text[:16_000], + context, + ) + + def save_mapping( + self, + run_id: int, + candidates: tuple[ProfileCandidate, ...], + scanned_through_id: int, + *, + now: datetime, + provider_called: bool, + ) -> None: + timestamp = now.isoformat() + with closing(self.connect()) as conn: + for candidate in candidates: + self._insert_candidate(conn, run_id, candidate, timestamp) + conn.execute( + """UPDATE compiled_profile_runs + SET current_watermark=?, map_calls=map_calls+?, updated_at=? + WHERE id=? AND status='mapping'""", + (scanned_through_id, int(provider_called), timestamp, run_id), + ) + conn.commit() + + @staticmethod + def _insert_candidate( + conn: sqlite3.Connection, + run_id: int, + candidate: ProfileCandidate, + timestamp: str, + ) -> None: + material = json.dumps( + [candidate.category, candidate.predicate, candidate.value, candidate.support_ids], + ensure_ascii=False, + ) + candidate_id = "pc-" + hashlib.sha256(material.encode("utf-8")).hexdigest()[:24] + conn.execute( + """INSERT OR IGNORE INTO compiled_profile_candidates + (run_id, candidate_id, category, predicate, value, volatility, + support_ids_json, created_at) VALUES (?,?,?,?,?,?,?,?)""", + ( + run_id, + candidate_id, + candidate.category, + candidate.predicate, + candidate.value.strip(), + candidate.volatility, + json.dumps(candidate.support_ids), + timestamp, + ), + ) + + def mark_reducing(self, run_id: int, *, now: datetime) -> None: + with closing(self.connect()) as conn: + conn.execute( + "UPDATE compiled_profile_runs SET status='reducing', updated_at=? WHERE id=?", + (now.isoformat(), run_id), + ) + conn.commit() + + def candidates(self, run_id: int) -> tuple[ProfileCandidate, ...]: + with closing(self.connect()) as conn: + rows = conn.execute( + """SELECT * FROM compiled_profile_candidates + WHERE run_id=? ORDER BY candidate_id""", + (run_id,), + ).fetchall() + return tuple( + ProfileCandidate( + str(row["candidate_id"]), + str(row["category"]), + str(row["predicate"]), + str(row["value"]), + str(row["volatility"]), + tuple(json.loads(row["support_ids_json"])), + ) + for row in rows + ) + + def active_facts(self) -> tuple[ProfileFact, ...]: + return self._facts("status='active'", ()) + + def fact(self, fact_id: int) -> ProfileFact: + rows = self._facts("id=?", (fact_id,)) + if not rows: + raise KeyError(f"compiled profile fact {fact_id} does not exist") + return rows[0] + + def _facts(self, where: str, params: tuple[Any, ...]) -> tuple[ProfileFact, ...]: + with closing(self.connect()) as conn: + rows = conn.execute( + f"SELECT * FROM compiled_profile_facts WHERE {where} ORDER BY id", params + ).fetchall() + supports = self._supports_by_fact(conn, [int(row["id"]) for row in rows]) + return tuple(self._fact_from_row(row, supports.get(int(row["id"]), ())) for row in rows) + + @staticmethod + def _supports_by_fact( + conn: sqlite3.Connection, fact_ids: list[int] + ) -> dict[int, tuple[int, ...]]: + if not fact_ids: + return {} + placeholders = ",".join("?" for _ in fact_ids) + rows = conn.execute( + f"""SELECT fact_id, verbatim_id FROM compiled_profile_supports + WHERE fact_id IN ({placeholders}) ORDER BY fact_id, verbatim_id""", + fact_ids, + ).fetchall() + grouped: dict[int, list[int]] = {} + for row in rows: + grouped.setdefault(int(row["fact_id"]), []).append(int(row["verbatim_id"])) + return {key: tuple(value) for key, value in grouped.items()} + + @staticmethod + def _fact_from_row(row: sqlite3.Row, supports: tuple[int, ...]) -> ProfileFact: + return ProfileFact( + fact_id=int(row["id"]), + category=str(row["category"]), + predicate=str(row["predicate"]), + value=str(row["value"]), + volatility=str(row["volatility"]), + status=str(row["status"]), + support_hash=str(row["support_hash"]), + support_count=int(row["support_count"]), + independent_sessions=int(row["independent_sessions"]), + first_seen_at=str(row["first_seen_at"]), + last_supported_at=str(row["last_supported_at"]), + support_ids=supports, + ) + + def apply_decisions( + self, + run_id: int, + decisions: tuple[ProfileDecision, ...], + *, + now: datetime, + min_sessions: int, + ) -> dict[str, int]: + candidates = {item.candidate_id: item for item in self.candidates(run_id)} + stats = {"applied": 0, "rejected": 0} + with closing(self.connect()) as conn: + for decision in decisions: + support_ids = self._decision_supports(decision, candidates) + applied = self._apply_decision( + conn, decision, support_ids, now=now, min_sessions=min_sessions + ) + stats["applied" if applied else "rejected"] += 1 + conn.commit() + return stats + + @staticmethod + def _decision_supports( + decision: ProfileDecision, candidates: dict[str, ProfileCandidate] + ) -> tuple[int, ...]: + support_ids = { + support_id + for candidate_id in decision.candidate_ids + for support_id in candidates[candidate_id].support_ids + } + return tuple(sorted(support_ids)) + + def _apply_decision( + self, + conn: sqlite3.Connection, + decision: ProfileDecision, + support_ids: tuple[int, ...], + *, + now: datetime, + min_sessions: int, + ) -> bool: + if decision.action == "ignore": + return True + support_rows = self._support_rows(conn, support_ids) + sessions = {str(row["session_id"]) for row in support_rows} + if len(support_rows) != len(support_ids): + return False + if decision.action in {"add", "replace"} and len(sessions) < min_sessions: + return False + if decision.action == "reinforce": + return self._reinforce(conn, int(decision.target_fact_id or 0), support_rows, now) + fact_id = self._upsert_fact(conn, decision, support_rows, now) + if decision.action == "replace": + return self._supersede(conn, int(decision.target_fact_id or 0), fact_id, now) + return bool(fact_id) + + @staticmethod + def _support_rows( + conn: sqlite3.Connection, support_ids: tuple[int, ...] + ) -> list[sqlite3.Row]: + if not support_ids: + return [] + placeholders = ",".join("?" for _ in support_ids) + return conn.execute( + f"""SELECT id, session_id, content, timestamp FROM verbatim_memories + WHERE role='user' AND id IN ({placeholders}) ORDER BY id""", + support_ids, + ).fetchall() + + def _upsert_fact( + self, + conn: sqlite3.Connection, + decision: ProfileDecision, + support_rows: list[sqlite3.Row], + now: datetime, + ) -> int: + key = self._fact_key(decision.category, decision.predicate, decision.value) + existing = conn.execute( + "SELECT id FROM compiled_profile_facts WHERE fact_key=?", (key,) + ).fetchone() + if existing: + fact_id = int(existing["id"]) + conn.execute( + "UPDATE compiled_profile_facts SET status='active', updated_at=? WHERE id=?", + (now.isoformat(), fact_id), + ) + else: + fact_id = self._insert_fact(conn, decision, key, support_rows, now) + self._attach_supports(conn, fact_id, support_rows, now) + self._refresh_support_stats(conn, fact_id, now) + return fact_id + + @staticmethod + def _insert_fact( + conn: sqlite3.Connection, + decision: ProfileDecision, + key: str, + support_rows: list[sqlite3.Row], + now: datetime, + ) -> int: + timestamps = [str(row["timestamp"]) for row in support_rows] + first_seen = min(timestamps) if timestamps else now.isoformat() + last_seen = max(timestamps) if timestamps else now.isoformat() + cur = conn.execute( + """INSERT INTO compiled_profile_facts + (fact_key, category, predicate, value, volatility, status, + support_hash, support_count, independent_sessions, first_seen_at, + last_supported_at, created_at, updated_at) + VALUES (?,?,?,?,?,'active',?,0,0,?,?,?,?)""", + ( + key, + decision.category, + decision.predicate, + decision.value.strip(), + decision.volatility, + hashlib.sha256(b"").hexdigest(), + first_seen, + last_seen, + now.isoformat(), + now.isoformat(), + ), + ) + return int(cur.lastrowid) + + @staticmethod + def _attach_supports( + conn: sqlite3.Connection, + fact_id: int, + rows: Iterable[sqlite3.Row], + now: datetime, + ) -> None: + for row in rows: + content_hash = hashlib.sha256(str(row["content"]).encode("utf-8")).hexdigest() + conn.execute( + """INSERT OR IGNORE INTO compiled_profile_supports + (fact_id, verbatim_id, session_id, message_hash, supported_at, created_at) + VALUES (?,?,?,?,?,?)""", + ( + fact_id, + int(row["id"]), + str(row["session_id"]), + content_hash, + str(row["timestamp"]), + now.isoformat(), + ), + ) + + @staticmethod + def _refresh_support_stats( + conn: sqlite3.Connection, fact_id: int, now: datetime + ) -> None: + rows = conn.execute( + """SELECT verbatim_id, session_id, message_hash, supported_at + FROM compiled_profile_supports WHERE fact_id=? ORDER BY verbatim_id""", + (fact_id,), + ).fetchall() + material = "|".join(f"{row['verbatim_id']}:{row['message_hash']}" for row in rows) + support_hash = hashlib.sha256(material.encode("utf-8")).hexdigest() + sessions = len({str(row["session_id"]) for row in rows}) + last_seen = max(str(row["supported_at"]) for row in rows) + conn.execute( + """UPDATE compiled_profile_facts + SET support_hash=?, support_count=?, independent_sessions=?, + last_supported_at=?, updated_at=? WHERE id=?""", + (support_hash, len(rows), sessions, last_seen, now.isoformat(), fact_id), + ) + + def _reinforce( + self, + conn: sqlite3.Connection, + fact_id: int, + rows: list[sqlite3.Row], + now: datetime, + ) -> bool: + target = conn.execute( + "SELECT id FROM compiled_profile_facts WHERE id=? AND status='active'", (fact_id,) + ).fetchone() + if target is None: + return False + self._attach_supports(conn, fact_id, rows, now) + self._refresh_support_stats(conn, fact_id, now) + return True + + @staticmethod + def _supersede( + conn: sqlite3.Connection, target_id: int, replacement_id: int, now: datetime + ) -> bool: + if target_id == replacement_id: + return False + target = conn.execute( + "SELECT id FROM compiled_profile_facts WHERE id=? AND status='active'", (target_id,) + ).fetchone() + if target is None: + return False + conn.execute( + """UPDATE compiled_profile_facts + SET status='superseded', replaced_by_fact_id=?, updated_at=? WHERE id=?""", + (replacement_id, now.isoformat(), target_id), + ) + return True + + def expire_preferences(self, *, now: datetime, ttl_days: int) -> int: + threshold = (now - timedelta(days=max(1, ttl_days))).isoformat() + with closing(self.connect()) as conn: + cur = conn.execute( + """UPDATE compiled_profile_facts SET status='expired', updated_at=? + WHERE status='active' AND volatility='preference' + AND last_supported_at None: + with closing(self.connect()) as conn: + conn.execute( + """UPDATE compiled_profile_runs + SET status='completed', active_slot=NULL, output_hash=?, + updated_at=?, completed_at=? WHERE id=?""", + (output_hash, now.isoformat(), now.isoformat(), run_id), + ) + conn.commit() + + def record_error(self, run_id: int, code: str, *, now: datetime) -> None: + with closing(self.connect()) as conn: + conn.execute( + "UPDATE compiled_profile_runs SET error_code=?, updated_at=? WHERE id=?", + (code[:120], now.isoformat(), run_id), + ) + conn.commit() + + def insert_fact_for_test( + self, + *, + category: str, + predicate: str, + value: str, + volatility: str, + last_supported_at: datetime, + support_count: int = 1, + ) -> int: + key = self._fact_key(category, predicate, value) + timestamp = last_supported_at.isoformat() + with closing(self.connect()) as conn: + cur = conn.execute( + """INSERT INTO compiled_profile_facts + (fact_key, category, predicate, value, volatility, status, + support_hash, support_count, independent_sessions, first_seen_at, + last_supported_at, created_at, updated_at) + VALUES (?,?,?,?,?,'active',?,?,?,?,?,?,?)""", + ( + key, + category, + predicate, + value, + volatility, + hashlib.sha256(key.encode()).hexdigest(), + support_count, + support_count, + timestamp, + timestamp, + timestamp, + timestamp, + ), + ) + conn.commit() + return int(cur.lastrowid) + + @staticmethod + def _fact_key(category: str, predicate: str, value: str) -> str: + normalized = " ".join(value.lower().split()) + return hashlib.sha256(f"{category}|{predicate}|{normalized}".encode("utf-8")).hexdigest() + + @staticmethod + def _has_table(conn: sqlite3.Connection, table: str) -> bool: + return conn.execute( + "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (table,) + ).fetchone() is not None + + +__all__ = ["ProfileRepository"] diff --git a/memorymaster/stores/migrations/0021_compiled_user_profile.py b/memorymaster/stores/migrations/0021_compiled_user_profile.py new file mode 100644 index 0000000..2bf070a --- /dev/null +++ b/memorymaster/stores/migrations/0021_compiled_user_profile.py @@ -0,0 +1,89 @@ +"""Add SQLite persistence for the resumable compiled user profile.""" + +from __future__ import annotations + +from typing import Any + + +VERSION = 21 +DESCRIPTION = "Add compiled user profile runs, candidates, facts, and supports" + + +_SQLITE_SCHEMA = """ +CREATE TABLE IF NOT EXISTS compiled_profile_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + status TEXT NOT NULL CHECK (status IN ( + 'mapping','reducing','completed','failed','cancelled' + )), + active_slot INTEGER CHECK (active_slot IS NULL OR active_slot = 1), + start_watermark INTEGER NOT NULL DEFAULT 0, + current_watermark INTEGER NOT NULL DEFAULT 0, + target_watermark INTEGER NOT NULL DEFAULT 0, + map_model TEXT NOT NULL, + reduce_model TEXT NOT NULL, + map_calls INTEGER NOT NULL DEFAULT 0, + error_code TEXT, + output_hash TEXT CHECK (output_hash IS NULL OR length(output_hash) = 64), + started_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT +); +CREATE UNIQUE INDEX IF NOT EXISTS idx_compiled_profile_active_run + ON compiled_profile_runs(active_slot) WHERE active_slot IS NOT NULL; +CREATE INDEX IF NOT EXISTS idx_compiled_profile_runs_status + ON compiled_profile_runs(status, updated_at); + +CREATE TABLE IF NOT EXISTS compiled_profile_candidates ( + run_id INTEGER NOT NULL REFERENCES compiled_profile_runs(id) ON DELETE CASCADE, + candidate_id TEXT NOT NULL, + category TEXT NOT NULL, + predicate TEXT NOT NULL, + value TEXT NOT NULL CHECK (length(value) BETWEEN 1 AND 240), + volatility TEXT NOT NULL CHECK (volatility IN ('stable','preference')), + support_ids_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (run_id, candidate_id) +); + +CREATE TABLE IF NOT EXISTS compiled_profile_facts ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + fact_key TEXT NOT NULL UNIQUE CHECK (length(fact_key) = 64), + category TEXT NOT NULL, + predicate TEXT NOT NULL, + value TEXT NOT NULL CHECK (length(value) BETWEEN 1 AND 240), + volatility TEXT NOT NULL CHECK (volatility IN ('stable','preference')), + status TEXT NOT NULL CHECK (status IN ('active','superseded','expired')), + support_hash TEXT NOT NULL CHECK (length(support_hash) = 64), + support_count INTEGER NOT NULL DEFAULT 0, + independent_sessions INTEGER NOT NULL DEFAULT 0, + first_seen_at TEXT NOT NULL, + last_supported_at TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + replaced_by_fact_id INTEGER REFERENCES compiled_profile_facts(id) +); +CREATE INDEX IF NOT EXISTS idx_compiled_profile_facts_active + ON compiled_profile_facts(status, category, predicate); + +CREATE TABLE IF NOT EXISTS compiled_profile_supports ( + fact_id INTEGER NOT NULL REFERENCES compiled_profile_facts(id) ON DELETE CASCADE, + verbatim_id INTEGER NOT NULL, + session_id TEXT NOT NULL, + message_hash TEXT NOT NULL CHECK (length(message_hash) = 64), + supported_at TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (fact_id, verbatim_id) +); +CREATE INDEX IF NOT EXISTS idx_compiled_profile_support_verbatim + ON compiled_profile_supports(verbatim_id, fact_id); +""" + + +def apply_sqlite(conn: Any) -> None: + conn.executescript(_SQLITE_SCHEMA) + conn.commit() + + +def apply_postgres(conn: Any) -> None: + del conn + raise RuntimeError("migration 21 is SQLite-only; PostgreSQL rollout is deferred") diff --git a/memorymaster/stores/snapshot.py b/memorymaster/stores/snapshot.py index 7d7c5b4..f779493 100644 --- a/memorymaster/stores/snapshot.py +++ b/memorymaster/stores/snapshot.py @@ -1,6 +1,7 @@ """Git-backed DB versioning: snapshot, list, rollback, diff.""" from __future__ import annotations +import hashlib import os import sqlite3 import subprocess @@ -163,13 +164,15 @@ def default_vacuum_dir() -> Path: def vacuum_dir_for(db_path: str | Path) -> Path: - """Per-DB snapshot dir: ``//``. + """Per-DB snapshot dir: ``/-/``. - Namespaced by DB filename (mirroring the spool's ``spool//`` - layout, spec §2.2) so two DBs sharing the base dir cannot evict each - other's ``mm-YYYYMMDD.db`` rotations. + The resolved source path participates in the namespace so disposable and + authoritative databases with the same filename cannot rotate each other's + recovery artifacts. """ - return default_vacuum_dir() / Path(db_path).stem + resolved = str(Path(db_path).resolve()).casefold() + suffix = hashlib.sha256(resolved.encode("utf-8")).hexdigest()[:12] + return default_vacuum_dir() / f"{Path(db_path).stem}-{suffix}" def vacuum_into( diff --git a/memorymaster/surfaces/scheduled_task.py b/memorymaster/surfaces/scheduled_task.py index 5345de5..5c833d0 100644 --- a/memorymaster/surfaces/scheduled_task.py +++ b/memorymaster/surfaces/scheduled_task.py @@ -50,6 +50,15 @@ def _capture_error_count(capture: object) -> int: return int(getattr(capture, "errors", 0) or 0) +def _compiled_profile_enabled() -> bool: + return os.environ.get("MEMORYMASTER_COMPILED_PROFILE", "").strip().lower() in { + "1", + "true", + "yes", + "on", + } + + def _run_dream(args: argparse.Namespace) -> int: _apply_dream_provider_contract(args) from memorymaster.capture.worker import run_capture_worker @@ -72,8 +81,25 @@ def _run_dream(args: argparse.Namespace) -> int: args.workspace, apply_candidates=bool(args.apply_candidates), ) - print({"queued": queued.to_dict(), "capture": capture, "dream": dream}) - passed = dream.get("ok") and not dream.get("errors") and not _capture_error_count(capture) + profile = {"ok": True, "status": "disabled"} + if _compiled_profile_enabled(): + from memorymaster.profile.engine import run_compiled_profile + + profile = run_compiled_profile(args.db) + print( + { + "queued": queued.to_dict(), + "capture": capture, + "dream": dream, + "compiled_profile": profile, + } + ) + passed = ( + dream.get("ok") + and not dream.get("errors") + and not _capture_error_count(capture) + and profile.get("ok") + ) return 0 if passed else 1 diff --git a/tests/test_compiled_profile.py b/tests/test_compiled_profile.py new file mode 100644 index 0000000..8f88fe7 --- /dev/null +++ b/tests/test_compiled_profile.py @@ -0,0 +1,254 @@ +from __future__ import annotations + +import json +from datetime import datetime, timedelta, timezone +from pathlib import Path + +import pytest + +from memorymaster.core.service import MemoryService +from memorymaster.profile.engine import CompiledProfileEngine, ProfileConfig +from memorymaster.profile.models import ( + ProfileCandidate, + ProfileDecision, + ProfileMessage, + ProfileValidationError, +) +from memorymaster.profile.providers import parse_map_output, parse_reduce_output +from memorymaster.profile.renderer import render_profile +from memorymaster.profile.repository import ProfileRepository +from memorymaster.recall.verbatim_store import ensure_verbatim_schema, store_verbatim + + +UTC = timezone.utc + + +def _database(tmp_path: Path) -> tuple[Path, ProfileRepository]: + db = tmp_path / "memory.db" + service = MemoryService(db, workspace_root=tmp_path) + service.init_db() + ensure_verbatim_schema(str(db)) + return db, ProfileRepository(db) + + +def _store( + db: Path, + session: str, + role: str, + content: str, + *, + scope: str = "project:test", + when: str = "2026-08-01T12:00:00+00:00", +) -> int: + row_id = store_verbatim( + str(db), session, role, content, scope, "test", timestamp=when + ) + assert row_id is not None + return row_id + + +def _candidate(candidate_id: str, support_id: int, value: str = "Argentina") -> ProfileCandidate: + return ProfileCandidate( + candidate_id=candidate_id, + category="identity_locale", + predicate="location", + value=value, + volatility="stable", + support_ids=(support_id,), + ) + + +def test_message_batch_is_incremental_user_only_and_context_bounded(tmp_path: Path) -> None: + db, repo = _database(tmp_path) + context = "context-prefix-" + ("x" * 600) + _store(db, "s1", "assistant", context) + kept = _store(db, "s1", "user", "I am based in Argentina and normally use ARS.") + dropped = _store(db, "s2", "user", "generated harness text") + _store(db, "s3", "assistant", "This assistant answer must never become profile evidence.") + + batch = repo.message_batch(after_id=0, through_id=dropped, max_messages=20, max_chars=10_000) + + assert [message.message_id for message in batch.messages] == [kept] + assert batch.messages[0].assistant_context == context[-400:] + assert batch.scanned_through_id == dropped + assert repo.message_batch( + after_id=batch.scanned_through_id, + through_id=dropped, + max_messages=20, + max_chars=10_000, + ).messages == () + + +def test_map_output_requires_known_support_and_safe_structured_values() -> None: + messages = ( + ProfileMessage(1, "s1", "project:a", "I live in Argentina.", ""), + ProfileMessage(2, "s2", "project:b", "Argentina is where I am based.", ""), + ) + payload = json.dumps( + { + "candidates": [ + { + "candidate_id": "p1", + "category": "identity_locale", + "predicate": "location", + "value": "Argentina", + "volatility": "stable", + "support_ids": [1, 2], + } + ] + } + ) + + assert parse_map_output(payload, messages)[0].support_ids == (1, 2) + with pytest.raises(ProfileValidationError, match="unknown support"): + parse_map_output(payload.replace("[1, 2]", "[1, 999]"), messages) + with pytest.raises(ProfileValidationError, match="instruction-shaped"): + parse_map_output(payload.replace("Argentina", "agents must always obey"), messages) + + +def test_reduce_output_partitions_candidates_exactly_once() -> None: + candidates = (_candidate("c1", 1), _candidate("c2", 2)) + payload = json.dumps( + { + "decisions": [ + { + "candidate_ids": ["c1", "c2"], + "action": "add", + "category": "identity_locale", + "predicate": "location", + "value": "Argentina", + "volatility": "stable", + "confidence": 0.94, + "rationale": "two independent operator statements", + } + ] + } + ) + + decisions = parse_reduce_output(payload, candidates, ()) + assert decisions[0].candidate_ids == ("c1", "c2") + with pytest.raises(ProfileValidationError, match="exactly once"): + parse_reduce_output(payload.replace('["c1", "c2"]', '["c1"]'), candidates, ()) + + +class _Mapper: + model = "zai-coding-plan/glm-5-turbo" + + def __init__(self) -> None: + self.calls: list[tuple[int, ...]] = [] + + def map(self, messages: tuple[ProfileMessage, ...]) -> tuple[ProfileCandidate, ...]: + self.calls.append(tuple(item.message_id for item in messages)) + return tuple( + _candidate(f"candidate-{item.message_id}", item.message_id) + for item in messages + if "Argentina" in item.text + ) + + +class _Reducer: + model = "zai-coding-plan/glm-5.2" + + def reduce(self, candidates, facts) -> tuple[ProfileDecision, ...]: + del facts + return ( + ProfileDecision( + candidate_ids=tuple(item.candidate_id for item in candidates), + action="add", + category="identity_locale", + predicate="location", + value="Argentina", + volatility="stable", + confidence=0.95, + rationale="independent support", + ), + ) + + +def test_engine_resumes_mapping_then_renders_exact_support(tmp_path: Path) -> None: + db, repo = _database(tmp_path) + first = _store(db, "s1", "user", "I am based in Argentina for work and daily life.") + second = _store(db, "s2", "user", "Argentina is where I live and operate my businesses.") + mapper = _Mapper() + output = tmp_path / "projection" + config = ProfileConfig(cadence_days=7, max_map_calls=1, max_messages=1) + engine = CompiledProfileEngine(repo, mapper, _Reducer(), output_dir=output, config=config) + + partial = engine.run(force=True) + completed = engine.run(force=True) + + assert partial["status"] == "mapping" + assert completed["status"] == "completed" + facts = repo.active_facts() + assert len(facts) == 1 + assert facts[0].support_ids == (first, second) + assert facts[0].independent_sessions == 2 + rendered = (output / "user.md").read_text(encoding="utf-8") + assert "The operator is based in Argentina." in rendered + assert "support=2" in rendered + manifest = json.loads((output / "user-profile.json").read_text(encoding="utf-8")) + assert manifest["facts"][0]["support_ids"] == [first, second] + + +def test_engine_rejects_single_session_add(tmp_path: Path) -> None: + db, repo = _database(tmp_path) + _store(db, "s1", "user", "I am based in Argentina for work and daily life.") + engine = CompiledProfileEngine( + repo, + _Mapper(), + _Reducer(), + output_dir=tmp_path / "projection", + config=ProfileConfig(max_map_calls=1), + ) + + result = engine.run(force=True) + + assert result["status"] == "completed" + assert result["rejected"] == 1 + assert repo.active_facts() == () + + +def test_stable_facts_survive_silence_and_preferences_expire(tmp_path: Path) -> None: + _db, repo = _database(tmp_path) + now = datetime(2026, 8, 12, tzinfo=UTC) + stable = repo.insert_fact_for_test( + category="identity_locale", + predicate="location", + value="Argentina", + volatility="stable", + last_supported_at=now - timedelta(days=500), + ) + preference = repo.insert_fact_for_test( + category="working_style", + predicate="communication_style", + value="concise and direct", + volatility="preference", + last_supported_at=now - timedelta(days=91), + ) + + expired = repo.expire_preferences(now=now, ttl_days=90) + + assert expired == 1 + assert repo.fact(stable).status == "active" + assert repo.fact(preference).status == "expired" + + +def test_renderer_is_deterministic_and_token_bounded(tmp_path: Path) -> None: + _db, repo = _database(tmp_path) + now = datetime(2026, 8, 12, tzinfo=UTC) + for index in range(60): + repo.insert_fact_for_test( + category="products_systems", + predicate="operates_product", + value=f"product-{index}-" + ("x" * 40), + volatility="stable", + last_supported_at=now, + support_count=60 - index, + ) + + first = render_profile(repo.active_facts(), token_budget=800, max_facts=40) + second = render_profile(tuple(reversed(repo.active_facts())), token_budget=800, max_facts=40) + + assert first.markdown == second.markdown + assert first.tokens_used <= 800 + assert len(first.fact_ids) <= 40 diff --git a/tests/test_compiled_profile_surfaces.py b/tests/test_compiled_profile_surfaces.py new file mode 100644 index 0000000..c1bf2c7 --- /dev/null +++ b/tests/test_compiled_profile_surfaces.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import importlib.util +import json +import subprocess +import sys +from pathlib import Path + + +ROOT = Path(__file__).resolve().parents[1] +HOOK = ROOT / "memorymaster" / "config_templates" / "hooks" / "memorymaster-session-start.py" + + +def _hook_module(): + spec = importlib.util.spec_from_file_location("compiled_profile_session_hook", HOOK) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + return module + + +def test_session_start_loads_only_generated_bounded_profile(tmp_path: Path) -> None: + module = _hook_module() + profile = tmp_path / "user.md" + module.PROFILE_PATH = profile + profile.write_text( + "\n" + "# Compiled User Profile\n\n- The operator is based in Argentina.\n", + encoding="utf-8", + ) + + loaded = module._load_compiled_profile() + context = module._format_context("project:test", [], {}, 0, [], loaded) + + assert "Compiled User Profile" in loaded + assert "MemoryMaster compiled user profile" in context + profile.write_text("# hand-written file", encoding="utf-8") + assert module._load_compiled_profile() == "" + + +def test_profile_module_cli_status_and_help(tmp_path: Path) -> None: + db = tmp_path / "profile.db" + help_run = subprocess.run( + [sys.executable, "-m", "memorymaster.profile", "--help"], + capture_output=True, + text=True, + encoding="utf-8", + timeout=20, + ) + status_run = subprocess.run( + [sys.executable, "-m", "memorymaster.profile", "status", "--db", str(db)], + capture_output=True, + text=True, + encoding="utf-8", + timeout=20, + ) + + assert help_run.returncode == 0 + assert "compiled user profile" in help_run.stdout.lower() + assert status_run.returncode == 0 + payload = json.loads(status_run.stdout) + assert payload == {"active_run": None, "facts": 0, "latest_completed_run": None} diff --git a/tests/test_integrity_job.py b/tests/test_integrity_job.py index 9184dcf..a562687 100644 --- a/tests/test_integrity_job.py +++ b/tests/test_integrity_job.py @@ -344,3 +344,20 @@ def test_snapshot_dir_default_outside_db_tree(monkeypatch: pytest.MonkeyPatch) - monkeypatch.setenv("MEMORYMASTER_SNAPSHOT_DIR", str(Path("X:/custom/snaps"))) assert snapshot.default_vacuum_dir() == Path("X:/custom/snaps") assert os.environ["MEMORYMASTER_SNAPSHOT_DIR"] # guard against typo'd env name + + +def test_vacuum_dir_isolated_for_same_named_databases( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + """Same DB filename in different roots must never share rotation state.""" + monkeypatch.setenv("MEMORYMASTER_SNAPSHOT_DIR", str(tmp_path / "snapshots")) + authoritative = tmp_path / "authoritative" / "memorymaster.db" + disposable = tmp_path / "disposable" / "memorymaster.db" + + authoritative_dir = snapshot.vacuum_dir_for(authoritative) + disposable_dir = snapshot.vacuum_dir_for(disposable) + + assert authoritative_dir != disposable_dir + assert authoritative_dir.name.startswith("memorymaster-") + assert disposable_dir.name.startswith("memorymaster-") + assert authoritative_dir.parent == disposable_dir.parent == tmp_path / "snapshots" diff --git a/tests/test_scheduled_task_runtime.py b/tests/test_scheduled_task_runtime.py index 6db7af8..1d9a271 100644 --- a/tests/test_scheduled_task_runtime.py +++ b/tests/test_scheduled_task_runtime.py @@ -123,6 +123,72 @@ def test_scheduled_dream_fails_when_capture_provider_errors( assert _run_dream(args) == 1 +def test_scheduled_dream_runs_enabled_compiled_profile( + tmp_path: Path, monkeypatch, +) -> None: + calls: list[tuple[str, object]] = [] + monkeypatch.setenv("MEMORYMASTER_COMPILED_PROFILE", "1") + monkeypatch.setattr( + "memorymaster.public.v1.improve", + lambda **_kwargs: SimpleNamespace(to_dict=dict), + ) + monkeypatch.setattr( + "memorymaster.capture.worker.run_capture_worker", + lambda *_args, **_kwargs: {"errors": 0}, + ) + monkeypatch.setattr( + "memorymaster.dreaming.worker.run_dream", + lambda *_args, **_kwargs: {"ok": True, "errors": 0}, + ) + + def fake_profile(db, **kwargs): + calls.append((db, kwargs)) + return {"ok": True, "status": "mapping"} + + monkeypatch.setattr("memorymaster.profile.engine.run_compiled_profile", fake_profile) + workspace = tmp_path / "memorymaster" + workspace.mkdir() + args = Namespace( + db=str(tmp_path / "scheduled.db"), + workspace=str(workspace), + apply_candidates=True, + ) + + assert _run_dream(args) == 0 + assert calls == [(args.db, {})] + + +def test_scheduled_dream_fails_closed_on_enabled_profile_error( + tmp_path: Path, monkeypatch, +) -> None: + monkeypatch.setenv("MEMORYMASTER_COMPILED_PROFILE", "1") + monkeypatch.setattr( + "memorymaster.public.v1.improve", + lambda **_kwargs: SimpleNamespace(to_dict=dict), + ) + monkeypatch.setattr( + "memorymaster.capture.worker.run_capture_worker", + lambda *_args, **_kwargs: {"errors": 0}, + ) + monkeypatch.setattr( + "memorymaster.dreaming.worker.run_dream", + lambda *_args, **_kwargs: {"ok": True, "errors": 0}, + ) + monkeypatch.setattr( + "memorymaster.profile.engine.run_compiled_profile", + lambda *_args, **_kwargs: {"ok": False, "status": "mapping"}, + ) + workspace = tmp_path / "memorymaster" + workspace.mkdir() + args = Namespace( + db=str(tmp_path / "scheduled.db"), + workspace=str(workspace), + apply_candidates=True, + ) + + assert _run_dream(args) == 1 + + def test_scheduled_main_logs_bound_dream_execution(tmp_path: Path, monkeypatch) -> None: log_path = tmp_path / "dream.log" seen: list[str] = []