diff --git a/README.md b/README.md index 34cb7bd6..5c387a1a 100644 --- a/README.md +++ b/README.md @@ -162,8 +162,8 @@ upgrade is picked up by the next session ([Settings Reference](docs/reference/settings-json.md)). Pinning a version is the debugging/repro path, not the default: `uv tool -install 'agents-remember-mcp==3.0.0rc5'`, or one-shot without installing, -`uvx --from 'agents-remember-mcp==3.0.0rc5' agents-remember dashboard`. +install 'agents-remember-mcp==3.0.0rc6'`, or one-shot without installing, +`uvx --from 'agents-remember-mcp==3.0.0rc6' agents-remember dashboard`. > **Pre-release note (until 3.0.0 final):** the dashboard currently ships in > `3.0.0rcN` pre-releases, which default version resolution skips. Install with @@ -239,7 +239,7 @@ ar-coordination/ ## Status -Agents Remember is at `3.0.0rc5` and actively developed. The core path — by-path onboarding, drift checks, and approval-gated updates — is in real use and stable enough to rely on. The public contracts listed under [Stability](#stability) are held stable across minor releases and change only on a major bump; the internals beneath them and the optional semantic/relationship providers may still evolve, so pin a version and read the notes for your target version in [GitHub Releases](https://github.com/Foxfire1st/agents-remember/releases) — the repository's canonical changelog — before upgrading. The Claude Code path is the most exercised; other harnesses are supported but less battle-tested. +Agents Remember is at `3.0.0rc6` and actively developed. The core path — by-path onboarding, drift checks, and approval-gated updates — is in real use and stable enough to rely on. The public contracts listed under [Stability](#stability) are held stable across minor releases and change only on a major bump; the internals beneath them and the optional semantic/relationship providers may still evolve, so pin a version and read the notes for your target version in [GitHub Releases](https://github.com/Foxfire1st/agents-remember/releases) — the repository's canonical changelog — before upgrading. The Claude Code path is the most exercised; other harnesses are supported but less battle-tested. The 3.0 arc: the working session itself is now observable and steerable — a system-managed agent lifecycle with durable approval gates and an event/projection layer, served as the mission-control browser cockpit directly from the MCP package (`agents-remember dashboard`; [#2](https://github.com/Foxfire1st/agents-remember/issues/2), [#43](https://github.com/Foxfire1st/agents-remember/issues/43)). The `rc` tag means the cockpit surface is still settling toward the final 3.0.0 contract; the architecture beneath it is the one described above. diff --git a/mcp/pyproject.toml b/mcp/pyproject.toml index e314eaa5..4c2f01de 100644 --- a/mcp/pyproject.toml +++ b/mcp/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "agents-remember-mcp" -version = "3.0.0rc5" +version = "3.0.0rc6" description = "Model Context Protocol server for Agents Remember." readme = "README.md" requires-python = ">=3.11" @@ -15,6 +15,7 @@ dependencies = [ "fastapi>=0.135.1,<1", "python-multipart>=0.0.9,<1", "uvicorn>=0.30,<1", + "watchfiles>=1.1,<2", "websockets>=12,<16", ] diff --git a/mcp/src/agents_remember/cli/dashboard.py b/mcp/src/agents_remember/cli/dashboard.py index 4034f35d..5997ddd9 100644 --- a/mcp/src/agents_remember/cli/dashboard.py +++ b/mcp/src/agents_remember/cli/dashboard.py @@ -19,6 +19,7 @@ from agents_remember.mcp.config import ConfigError, load_config from agents_remember.serving import daemon as serving_daemon from agents_remember.serving.app import create_app +from agents_remember.serving.change_watcher import DEFAULT_HEARTBEAT_SECONDS from agents_remember.serving.sim import SimError, build_sim, parse_sim_speed # Dev hot-reload (``--reload``): uvicorn's reloader re-imports the app per worker restart, so it @@ -26,12 +27,18 @@ # The factory reads the resolved config from the environment the parent ``run`` sets. _DEV_CONFIG_ENV = "AR_DASHBOARD_DEV_CONFIG" _DEV_INTERVAL_ENV = "AR_DASHBOARD_DEV_INTERVAL" +_DEV_HEARTBEAT_ENV = "AR_DASHBOARD_DEV_HEARTBEAT" def _dev_app(): """Zero-arg app factory for ``uvicorn --reload`` (live state only; never sim).""" config = load_config(os.environ[_DEV_CONFIG_ENV]) - return create_app(config, interval=float(os.environ.get(_DEV_INTERVAL_ENV, "1.0"))) + heartbeat_env = os.environ.get(_DEV_HEARTBEAT_ENV) + return create_app( + config, + interval=float(os.environ.get(_DEV_INTERVAL_ENV, "1.0")), + heartbeat=float(heartbeat_env) if heartbeat_env else None, + ) def add_arguments(parser: argparse.ArgumentParser) -> None: @@ -52,7 +59,22 @@ def add_arguments(parser: argparse.ArgumentParser) -> None: help="Bind port (default: the dashboard.port settings key, else 8765).", ) parser.add_argument( - "--interval", type=float, default=1.0, help="Projection refresh interval, seconds." + "--interval", + type=float, + default=1.0, + help="Fast-path projection cadence floor, seconds: change-driven re-projections are " + "never spaced closer than this, a continuously-busy world still projects once per " + "interval, and it stays the fixed tick cadence under --sim or when the change " + "watcher is unavailable. Also the /api/events raw-tail poll cadence.", + ) + parser.add_argument( + "--heartbeat", + type=float, + default=None, + help="Idle re-projection heartbeat, seconds (default " + f"{DEFAULT_HEARTBEAT_SECONDS:g}). With no detected input change the projection " + "still refreshes at this cadence -- the staleness bound for /api/state and for " + "time-derived fields (ageSeconds/staleSeconds and stale/overdue flips).", ) parser.add_argument( "--reload", @@ -118,6 +140,8 @@ def run(args: argparse.Namespace) -> int: # re-import on change; watch only the package source so node_modules/.git don't churn it. os.environ[_DEV_CONFIG_ENV] = str(Path(config_path).resolve()) os.environ[_DEV_INTERVAL_ENV] = str(args.interval) + if args.heartbeat is not None: + os.environ[_DEV_HEARTBEAT_ENV] = str(args.heartbeat) uvicorn.run( "agents_remember.cli.dashboard:_dev_app", factory=True, @@ -140,7 +164,7 @@ def run(args: argparse.Namespace) -> int: sim.config, interval=args.interval, now=sim.clock.now, before_tick=sim.feeder.feed ) else: - app = create_app(config, interval=args.interval) + app = create_app(config, interval=args.interval, heartbeat=args.heartbeat) uvicorn.run(app, host=args.host, port=port, access_log=not args.no_access_log) return 0 @@ -166,6 +190,8 @@ def _run_daemon_command( if args.stop: print(f"dashboard daemon: {serving_daemon.stop(directory)}") return 0 - result = serving_daemon.ensure(config, host=args.host, port=port, interval=args.interval) + result = serving_daemon.ensure( + config, host=args.host, port=port, interval=args.interval, heartbeat=args.heartbeat + ) print(f"dashboard daemon {result.action}: {result.detail}") return 0 if result.action in ("adopted", "started", "restarted") else 1 diff --git a/mcp/src/agents_remember/mcp/__init__.py b/mcp/src/agents_remember/mcp/__init__.py index 42e350ea..4812a268 100644 --- a/mcp/src/agents_remember/mcp/__init__.py +++ b/mcp/src/agents_remember/mcp/__init__.py @@ -8,4 +8,4 @@ # Single source of truth: the installed package metadata (mcp/pyproject.toml). SERVER_VERSION = version("agents-remember-mcp") except PackageNotFoundError: # running from a source checkout without an install - SERVER_VERSION = "3.0.0rc5" + SERVER_VERSION = "3.0.0rc6" diff --git a/mcp/src/agents_remember/observer/contract_snapshot.py b/mcp/src/agents_remember/observer/contract_snapshot.py new file mode 100644 index 00000000..5505f09c --- /dev/null +++ b/mcp/src/agents_remember/observer/contract_snapshot.py @@ -0,0 +1,145 @@ +"""One immutable leaf-enclosure-contract snapshot per projection tick (260712-PTS-L2). + +Each projection tick previously enumerated ``iter_leaf_enclosure_contracts`` and +re-parsed EVERY leaf enclosure contract three times: ``read_enclosures``, +``read_engine_process_facts``, and drift-snapshot pruning each ran their own +walk + ``load_contract`` pass. :class:`ContractSnapshotCache.build` performs that +pass ONCE at tick start and hands the result to all three readers as an immutable +:class:`ContractSnapshot`, backed by a stat-identity parse cache (path + +``mtime_ns`` + size + ``ctime_ns``) so an unchanged contract file is not re-read +or re-parsed on later ticks at all. + +Concurrency discipline (mirrors ``projection_store._lifecycle_log_cache``): the +cache is mutated only inside ``build``, which runs on the projection worker +thread -- the projector serializes ticks by awaiting each ``asyncio.to_thread`` +call, so builds never overlap. The published :class:`ContractSnapshot` is an +immutable value (read-only mapping of frozen contracts) that may be handed to any +consumer without locks. The landing refresher and supervisor sweep keep their own +independent passes; they never touch this cache. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType + +from agents_remember.worktrees.task_resolver import iter_leaf_enclosure_contracts +from agents_remember.worktrees.worktree_contract import ( + ContractError, + WorktreeContract, + load_contract, +) + + +@dataclass(frozen=True) +class ContractSnapshot: + """Immutable per-tick view of every live leaf enclosure contract. + + ``contracts`` preserves the sorted enumeration order of + ``iter_leaf_enclosure_contracts`` (insertion-ordered mapping), so consumers + iterate exactly the paths, contracts, and order they previously walked + themselves. ``skipped`` carries the paths whose parse failed this tick -- the + same skip-never-fatal containment each reader applied inline before. + """ + + contracts: Mapping[Path, WorktreeContract] + skipped: frozenset[Path] + + +@dataclass(frozen=True) +class _ParseCacheEntry: + mtime_ns: int + size: int + ctime_ns: int + contract: WorktreeContract + + +class ContractSnapshotCache: + """Cross-tick contract parse cache keyed by stat identity (R2) with live-set pruning (R3). + + A cache entry is reused only while the contract file's ``(mtime_ns, size, + ctime_ns)`` is unchanged; any stat change re-parses. ``ctime_ns`` is part of + the identity (adversarial-review hardening): a ``chmod 000`` changes neither + mtime nor size, so without it the cache would serve the old good parse + forever where the pre-cache readers degraded to skip-every-tick, and a + rewrite whose ``(mtime_ns, size)`` was pinned via ``os.utime`` would never be + seen; ctime changes on both, while staying untouched for genuinely unchanged + files -- so the hardening costs zero extra parses. Parse FAILURES are never + cached: an unreadable or malformed contract is skipped this build and + re-attempted on the next one, exactly the retry-every-tick containment the + readers had when they parsed inline (a transient ``OSError`` therefore + self-heals without waiting for a stat change). Entries whose paths left the + enumeration are dropped on the next build, so retention is bounded by the + live contract set. + """ + + def __init__(self) -> None: + self._entries: dict[Path, _ParseCacheEntry] = {} + + def build(self, tasks_root: Path) -> ContractSnapshot: + """Enumerate + parse the live leaf contracts once; publish an immutable snapshot.""" + contracts: dict[Path, WorktreeContract] = {} + skipped: set[Path] = set() + seen: set[Path] = set() + for path in iter_leaf_enclosure_contracts(tasks_root): + seen.add(path) + stat = _safe_stat(path) + cached = self._cached_contract(path, stat) + if cached is not None: + contracts[path] = cached + continue + try: + contract = load_contract(path) + except (ContractError, OSError): + self._entries.pop(path, None) + skipped.add(path) + continue + if stat is not None: + self._entries[path] = _ParseCacheEntry( + mtime_ns=stat.st_mtime_ns, + size=stat.st_size, + ctime_ns=stat.st_ctime_ns, + contract=contract, + ) + else: + self._entries.pop(path, None) + contracts[path] = contract + for stale in [path for path in self._entries if path not in seen]: + del self._entries[stale] + return ContractSnapshot(contracts=MappingProxyType(contracts), skipped=frozenset(skipped)) + + def _cached_contract(self, path: Path, stat: os.stat_result | None) -> WorktreeContract | None: + """The cached parse, only while the stat identity (mtime_ns + size + ctime_ns) holds.""" + if stat is None: + return None + entry = self._entries.get(path) + if ( + entry is None + or entry.mtime_ns != stat.st_mtime_ns + or entry.size != stat.st_size + or entry.ctime_ns != stat.st_ctime_ns + ): + return None + return entry.contract + + +def _safe_stat(path: Path) -> os.stat_result | None: + """Stat is a cache concern only: the readers never stat'ed before this leaf, so a + failed stat must not introduce a new skip path -- the caller falls through to the + (uncached) parse attempt, which applies the original containment.""" + try: + return path.stat() + except OSError: + return None + + +def build_contract_snapshot(tasks_root: Path) -> ContractSnapshot: + """One-shot snapshot for standalone reader calls (no cross-tick cache). + + Readers keep their public signatures: a call without an injected snapshot + builds its own local pass -- identical cost and behavior to the pre-L2 walk. + """ + return ContractSnapshotCache().build(tasks_root) diff --git a/mcp/src/agents_remember/observer/drift_snapshots.py b/mcp/src/agents_remember/observer/drift_snapshots.py index e43b46ef..890f984e 100644 --- a/mcp/src/agents_remember/observer/drift_snapshots.py +++ b/mcp/src/agents_remember/observer/drift_snapshots.py @@ -9,9 +9,11 @@ from agents_remember.memory_quality.integrity.onboarding_drift_check.report import ( sanitize_report_token, ) +from agents_remember.observer.contract_snapshot import ( + ContractSnapshot, + build_contract_snapshot, +) from agents_remember.observer.paths import DRIFT_SNAPSHOT_SCHEMA, drift_snapshot_dir -from agents_remember.worktrees.task_resolver import iter_leaf_enclosure_contracts -from agents_remember.worktrees.worktree_contract import ContractError, load_contract def drift_snapshot_path(coordination_root: Path, *, repository: str, branch: str) -> Path: @@ -31,8 +33,15 @@ def remove_drift_snapshot( ) -def prune_orphaned_drift_snapshots(config: Any) -> dict[str, object]: - """Remove valid drift snapshots that no longer map to a configured repo or live worktree.""" +def prune_orphaned_drift_snapshots( + config: Any, *, contracts: ContractSnapshot | None = None +) -> dict[str, object]: + """Remove valid drift snapshots that no longer map to a configured repo or live worktree. + + 260712-PTS-L2: the projection tick passes its shared per-tick + :class:`ContractSnapshot` so pruning adds ZERO contract parses; a standalone + call (``contracts=None``) builds a local snapshot with identical behavior. + """ directory = drift_snapshot_dir(config.coordination_root) if not directory.is_dir(): return {"removed": [], "kept": 0, "skipped": 0} @@ -40,7 +49,9 @@ def prune_orphaned_drift_snapshots(config: Any) -> dict[str, object]: configured_repositories = { scope.path.name for scope in config.repositories.values() if scope.path.name } - active_worktrees = _active_worktree_snapshot_keys(config.coordination_root) + active_worktrees = _active_worktree_snapshot_keys( + config.coordination_root, contracts=contracts + ) removed: list[dict[str, object]] = [] kept = 0 skipped = 0 @@ -60,13 +71,14 @@ def prune_orphaned_drift_snapshots(config: Any) -> dict[str, object]: return {"removed": removed, "kept": kept, "skipped": skipped} -def _active_worktree_snapshot_keys(coordination_root: Path) -> set[tuple[str, str]]: +def _active_worktree_snapshot_keys( + coordination_root: Path, *, contracts: ContractSnapshot | None = None +) -> set[tuple[str, str]]: + snapshot = ( + contracts if contracts is not None else build_contract_snapshot(coordination_root / "tasks") + ) keys: set[tuple[str, str]] = set() - for path in iter_leaf_enclosure_contracts(coordination_root / "tasks"): - try: - contract = load_contract(path) - except (ContractError, OSError): - continue + for contract in snapshot.contracts.values(): if contract.code_worktree.exists(): keys.add((contract.code_worktree.name, contract.code_work_branch)) return keys diff --git a/mcp/src/agents_remember/observer/projection_store.py b/mcp/src/agents_remember/observer/projection_store.py index ddac8e03..dce5cd7d 100644 --- a/mcp/src/agents_remember/observer/projection_store.py +++ b/mcp/src/agents_remember/observer/projection_store.py @@ -22,6 +22,7 @@ from typing import TYPE_CHECKING, Any, Protocol from agents_remember.controlplane.attention_dismissals import AttentionDismissalStore +from agents_remember.observer.contract_snapshot import ContractSnapshotCache from agents_remember.observer.drift_snapshots import prune_orphaned_drift_snapshots from agents_remember.observer.event_retention import prune_expired_lifecycle_event_logs from agents_remember.observer.events import Event @@ -98,6 +99,14 @@ class _LifecycleLogCacheEntry: log_events: list[Event] +# 260712-PTS-L2 R1/R2: ONE leaf-enclosure-contract enumeration + parse pass per tick, shared by +# read_enclosures, read_engine_process_facts, and drift-snapshot pruning (each previously ran its +# own walk + load_contract pass = 3x per tick). The cache reuses a parsed contract while its +# (mtime_ns, size, ctime_ns) is unchanged and prunes to the live enumeration each build. Mutated only on the +# projection worker thread (ticks are serialized by the projector's awaited asyncio.to_thread), +# matching the module-level cache discipline of _lifecycle_log_cache below. +_contract_snapshot_cache = ContractSnapshotCache() + # 260707-HFX2-L12 F9/F7: the projection re-read + re-validated EVERY lifecycle's full events.jsonl # each 1s tick. Heartbeats append only every 15s (and stop entirely once a leaf is parked past its # inactivity cutoff), so most logs are byte-identical between ticks. This cache reuses the parsed @@ -201,10 +210,14 @@ def project_and_write( moment = now or datetime.now(UTC) root = observer_root(config) coordination_root = config.coordination_root + # 260712-PTS-L2: ONE contract pass per tick -- the snapshot is built here on the projection + # worker thread and handed (immutable) to read_enclosures, drift-snapshot pruning, and + # read_engine_process_facts below, replacing their three independent walk+parse passes. + contract_snapshot = _contract_snapshot_cache.build(coordination_root / "tasks") # Read the durable enclosures FIRST so retention is enclosure-aware: a not-yet-retired master # series protects every one of its leaves' event logs from the inactivity TTL (a running durable # task must never lose its history just because it has been a while since its last lifecycle event). - enclosures = read_enclosures(coordination_root) + enclosures = read_enclosures(coordination_root, contracts=contract_snapshot) prune_expired_lifecycle_event_logs( root, now=moment, @@ -214,7 +227,7 @@ def project_and_write( sidecar_staleness, route_coverage, ledgers = _gather_repo_surfaces_cached(config, moment) provider_groups = admitted_worktree_groups(enclosures, lifecycle_logs, now=moment) engine_groups = active_enclosure_worktree_groups(enclosures, lifecycle_logs, now=moment) - prune_orphaned_drift_snapshots(config) + prune_orphaned_drift_snapshots(config, contracts=contract_snapshot) if provider_refresher is not None: provider_refresher.maybe_refresh(config, now=moment) attention_store = AttentionDismissalStore(root) @@ -241,6 +254,7 @@ def project_and_write( active_worktree_groups=engine_groups, now=moment, landing_state=landing_state, + contracts=contract_snapshot, ), engine_start_progress=read_start_progress_entries(coordination_root, now=moment), gates=read_gates(coordination_root, now=moment), diff --git a/mcp/src/agents_remember/observer/snapshots.py b/mcp/src/agents_remember/observer/snapshots.py index 56dd75d1..a74a7b94 100644 --- a/mcp/src/agents_remember/observer/snapshots.py +++ b/mcp/src/agents_remember/observer/snapshots.py @@ -43,6 +43,10 @@ discover_onboarding_files, parse_table_metadata, ) +from agents_remember.observer.contract_snapshot import ( + ContractSnapshot, + build_contract_snapshot, +) from agents_remember.observer.paths import ( DRIFT_SNAPSHOT_SCHEMA, drift_snapshot_dir, @@ -106,9 +110,8 @@ from agents_remember.worktrees.task_resolver import ( ARCHIVE_DIR, ENCLOSURES_DIR, - iter_leaf_enclosure_contracts, ) -from agents_remember.worktrees.worktree_contract import ContractError, load_contract +from agents_remember.worktrees.worktree_contract import WorktreeContract if TYPE_CHECKING: from agents_remember.observer.landing_state import LandingStateReader @@ -470,27 +473,27 @@ def _worktree_runtime_summary( } -def read_enclosures(coordination_root: Path) -> list[EnclosureNode]: +def read_enclosures( + coordination_root: Path, *, contracts: ContractSnapshot | None = None +) -> list[EnclosureNode]: """Surfaces 5/6: every active leaf enclosure contract. Leaf contracts live below ``enclosures//series-contract.md``. Root series contracts describe integration branches and are not live worktree processes. A malformed contract is skipped, never fatal to the projection. + + 260712-PTS-L2: the projection tick passes its shared per-tick + :class:`ContractSnapshot` so this reader adds ZERO contract parses; a + standalone call (``contracts=None``) builds a local snapshot, preserving the + public signature and the walk-and-skip behavior it had before. """ - tasks_root = coordination_root / "tasks" - nodes: list[EnclosureNode] = [] - for path in iter_leaf_enclosure_contracts(tasks_root): - node = _enclosure_from_contract(path) - if node is not None: - nodes.append(node) - return nodes + snapshot = ( + contracts if contracts is not None else build_contract_snapshot(coordination_root / "tasks") + ) + return [_enclosure_from_contract(contract) for contract in snapshot.contracts.values()] -def _enclosure_from_contract(path: Path) -> EnclosureNode | None: - try: - contract = load_contract(path) - except (ContractError, OSError): - return None +def _enclosure_from_contract(contract: WorktreeContract) -> EnclosureNode: return EnclosureNode( enclosure=contract.contract_path.as_posix(), enclosureId=contract.leaf_id or contract.contract_path.parent.name, @@ -639,25 +642,26 @@ def read_engine_process_facts( active_worktree_groups: set[str] | None = None, now: datetime | None = None, landing_state: LandingStateReader | None = None, + contracts: ContractSnapshot | None = None, ) -> list[EngineProcessFacts]: """Slice 5e: gather one fact bundle per leaf enclosure for the Engine Room map. - Globs the same leaf enclosure files as :func:`read_enclosures`, but - enriches each with the status-guidance facts the structural ``EnclosureNode`` omits (the - code/memory branches, base commits, worktree paths, existence/dirty flags, base freshness, - and provider-boot status). ``contract_payload`` and ``lifecycle_guidance`` are pure; only - ``status_payload`` touches git, and it is best-effort so a contract pointing at absent or - fake worktrees degrades to ``status=None`` (rendered as missing/derived) instead of - crashing the projection tick. A malformed contract is skipped, never fatal. + Reads the same leaf enclosure contracts as :func:`read_enclosures` (via the shared + per-tick :class:`ContractSnapshot` when the projection passes one; a standalone call + builds a local snapshot), but enriches each with the status-guidance facts the + structural ``EnclosureNode`` omits (the code/memory branches, base commits, worktree + paths, existence/dirty flags, base freshness, and provider-boot status). + ``contract_payload`` and ``lifecycle_guidance`` are pure; only ``status_payload`` + touches git, and it is best-effort so a contract pointing at absent or fake worktrees + degrades to ``status=None`` (rendered as missing/derived) instead of crashing the + projection tick. A malformed contract is skipped, never fatal. """ - tasks_root = coordination_root / "tasks" + snapshot = ( + contracts if contracts is not None else build_contract_snapshot(coordination_root / "tasks") + ) facts: list[EngineProcessFacts] = [] seen_status_keys: set[str] = set() - for path in iter_leaf_enclosure_contracts(tasks_root): - try: - contract = load_contract(path) - except (ContractError, OSError): - continue + for path, contract in snapshot.contracts.items(): if ( active_worktree_groups is not None and contract.worktree_group.name not in active_worktree_groups diff --git a/mcp/src/agents_remember/serving/app.py b/mcp/src/agents_remember/serving/app.py index 83c3e402..dcc9f02b 100644 --- a/mcp/src/agents_remember/serving/app.py +++ b/mcp/src/agents_remember/serving/app.py @@ -91,6 +91,7 @@ ) from agents_remember.serving.actions import ActionRequest, evaluate_action from agents_remember.serving.build_info import ServingBuild, resolve_serving_build +from agents_remember.serving.change_watcher import ProjectionInputWatcher from agents_remember.serving.changeset import register_changeset_routes from agents_remember.serving.events import stream_raw_events from agents_remember.serving.files import register_files_routes @@ -466,10 +467,12 @@ def create_app( config: McpRuntimeConfig, *, interval: float = 1.0, + heartbeat: float | None = None, now: Callable[[], datetime] | None = None, before_tick: Callable[[datetime], object] | None = None, refresh_provider_state: bool | None = None, refresh_landing_state: bool | None = None, + watch_changes: bool | None = None, terminal_host: TerminalHost | None = None, terminal_catalog: TerminalCatalog | None = None, terminal_paster: TerminalPaster | None = None, @@ -477,21 +480,29 @@ def create_app( """Build the dashboard app bound to one shared projector for ``config``. ``now`` / ``before_tick`` default to live behaviour; sim wires a replay clock + feeder. Live - serving enables the landing-state refresher by default; sim disables it unless explicitly set. - ``terminal_host`` defaults to a fresh :class:`TerminalHost` (the Mode B2 terminal backend); - tests inject a fake to drive the WebSocket bridge without a real PTY. + serving enables the landing-state refresher and the change-driven projection watcher by + default; sim disables both unless explicitly set (replay must stay time-driven -- the sim + feeder only writes *inside* a tick, so a change-gated loop would never wake). ``interval`` + is the fast-path projection cadence floor; ``heartbeat`` bounds quiet-world ``/api/state`` + staleness (260712-PTS-L3, default ``DEFAULT_HEARTBEAT_SECONDS``). ``terminal_host`` + defaults to a fresh :class:`TerminalHost` (the Mode B2 terminal backend); tests inject a + fake to drive the WebSocket bridge without a real PTY. """ if refresh_provider_state is None: refresh_provider_state = before_tick is None if refresh_landing_state is None: refresh_landing_state = before_tick is None + if watch_changes is None: + watch_changes = before_tick is None projector = Projector( config, interval=interval, + heartbeat=heartbeat, now=now, before_tick=before_tick, provider_refresher=ProviderStateRefresher() if refresh_provider_state else None, landing_refresher=LandingStateRefresher(config) if refresh_landing_state else None, + change_watcher=ProjectionInputWatcher(config) if watch_changes else None, ) host = terminal_host if terminal_host is not None else TerminalHost() catalog = terminal_catalog or TerminalCatalog(terminal_catalog_path(config.coordination_root)) diff --git a/mcp/src/agents_remember/serving/change_watcher.py b/mcp/src/agents_remember/serving/change_watcher.py new file mode 100644 index 00000000..285cc174 --- /dev/null +++ b/mcp/src/agents_remember/serving/change_watcher.py @@ -0,0 +1,390 @@ +"""Change-driven projection pacing: the debounced input watcher + the wake scheduler. + +260712-PTS-L3: the projector used to re-project the whole world unconditionally every +``--interval`` seconds (production 1.0s) even when nothing changed -- ~160% steady CPU on +a quiet-but-large tree. This module makes waking *change-driven*: a filesystem watcher +(``watchfiles``, inotify-backed) over the projection's actual input surfaces feeds a +small debounce/coalesce state machine (:class:`ChangePacer`), and a slow heartbeat +covers everything a watcher cannot see. The projector's tick body is untouched -- only +*when* it wakes changes, never *what* a tick does. + +The authoritative input list (R1), derived reader-by-reader from +``observer.projection_store.project_and_write``: + +======================================== ===================================================== +watched root readers it feeds +======================================== ===================================================== +``/tasks`` read_enclosures, read_task_documents, + read_series_documents, read_engine_process_facts + (leaf enclosure contracts) +``/lifecycles`` read_lifecycle_logs (``events.jsonl`` + + ``heartbeat.json`` sidecars), per-lifecycle + ``gates.jsonl`` (read_gates) +``/workspace`` workspace ``gates.jsonl``, attention dismissals, + operator inbox (read_agent_pickups), expectation rows +``/drift`` read_drift_snapshots +``/logs/providers/status`` workspace provider ``current.json`` (read_providers) +``/logs/providers/setup`` read_setup_summaries +``/temp/worktree-start`` read_start_progress_entries +``/temp/tool-reports`` read_tool_reports +======================================== ===================================================== + +(```` = ``/logs/observer``.) Deliberately NOT watched -- heartbeat-covered +blind spots: + +* the per-repo memory roots (sidecar staleness / route coverage / ledgers): they sit + behind ``REPO_SURFACE_REFRESH_TTL_SECONDS`` (15s) already, and their onboarding trees + are large, so their freshness bound is TTL + heartbeat; +* landing state: an in-memory 30s git-derived refresher with no file signal; +* the entire ``worktrees/`` tree: the checkouts are ~50k dirs (recursive watching would + re-create the scan cost this change removes) and each task's ``provider-runtime`` holds + live container data (Postgres/grepai) that is unreadable to the daemon user and churns on + every container write -- watching it crashed the watcher on permission-denied and would + have re-projected on WAL writes. Worktree ``provider-state.json`` / ``setup-progress.json`` + changes are infrequent and heartbeat-covered; central provider status is watched via + ``logs/providers``. + +Self-trigger safety: the projection's own per-tick outputs (``latest-state.json`` / +``latest-metrics.json`` and their ``*.tmp`` siblings) live at the observer root, *outside* +every watched subdirectory, so a tick cannot re-wake itself. Inside ``/workspace`` +the non-input churn (the raw event river + cursor/lock, the supervisor heartbeat) is +name-filtered. TTL-gated writers that run inside a tick (gate-log compaction, event +retention, provider current-state refresh) cost at most one debounced echo tick per TTL +window, whose diff emits nothing. + +Freshness bounds (R5, unchanged SSE semantics): a write into a quiet world becomes an SSE +delta within ``debounce + projection time``; under sustained writes projections are floored +to one per ``--interval`` and bounded by ``max_delay = interval`` (R2 -- the busy world keeps +the former 1s cadence); with no writes at all, ``/api/state`` staleness -- and every +time-*derived* field or state flip (``ageSeconds``/``staleSeconds``, stale/overdue decays) -- +is bounded by the heartbeat (R3/R4; default :data:`DEFAULT_HEARTBEAT_SECONDS`). The +volatile age fields were already stripped from the delta stream and advanced client-side +(``dashboard/src/data/servedAges.ts``), so heartbeat-cadence refresh does not change what +an SSE client displays between emissions. + +Failure posture (R7): ``watchfiles`` missing, zero watchable roots, or a crashed watch +degrades LOUDLY (error log) to fixed ``--interval`` ticking -- exactly today's behaviour -- +and keeps retrying; fail-open, never fail-silent. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import time +from pathlib import Path +from typing import TYPE_CHECKING, Protocol + +from agents_remember.observer.paths import drift_snapshot_dir, observer_logs_root +from agents_remember.observer.projection_store import LATEST_METRICS, LATEST_STATE +from agents_remember.observer.store import WORKSPACE_CURSOR_FILE, WORKSPACE_LOCK_FILE +from agents_remember.serving.supervisor_heartbeat import supervisor_heartbeat_path + +if TYPE_CHECKING: + from collections.abc import Callable + + from watchfiles import Change + + from agents_remember.mcp.config import McpRuntimeConfig + +try: # R7: a missing wheel must degrade the daemon loudly, never crash it at import time. + import watchfiles +except ImportError: # pragma: no cover - exercised via the injected-None tests + watchfiles = None # type: ignore[assignment] + +logger = logging.getLogger(__name__) + +#: Idle re-projection cadence (R3): the self-heal bound for watcher blind spots and the +#: resolution of time-derived projection fields (R4). Configurable via ``--heartbeat``. +DEFAULT_HEARTBEAT_SECONDS = 15.0 + +#: Settle window after the last observed change before projecting (R2): long enough to +#: coalesce a multi-file write burst, short enough to feel immediate on the dashboard. +DEBOUNCE_SECONDS = 0.1 + +#: How the raw watchfiles batches are grouped before they reach the pacer. The library +#: default (1600ms) could delay first detection beyond the 1s max-delay bound under +#: sustained writes, so batching is kept well under ``--interval``. +_AWATCH_DEBOUNCE_MS = 200 +_AWATCH_STEP_MS = 50 + +#: Cadence for re-deriving the watch-root set (new worktree groups / newly created input +#: dirs) and for retrying after a watch failure. Gaps in between are heartbeat-covered. +WATCH_REFRESH_SECONDS = 30.0 + +#: Basenames excluded everywhere (defensive: the projection's own atomic outputs). +_EXCLUDED_NAMES = frozenset({LATEST_STATE, LATEST_METRICS}) + +#: ``/workspace`` files that are NOT projection inputs: the raw event river (served +#: by ``/api/events``, never read by ``project_and_write``) with its cursor + lock, the +#: operator-inbox flock file (opened ``a+b`` by every inbox access, including each tick's +#: ``read_agent_pickups`` -- its boot-time creation would emit one spurious change-tick), +#: and the supervisor's own heartbeat (written every sweep on its own cadence). +_EXCLUDED_WORKSPACE_NAMES = frozenset( + { + "events.jsonl", + WORKSPACE_CURSOR_FILE, + WORKSPACE_LOCK_FILE, + "operator-inbox.lock", + supervisor_heartbeat_path(Path(".")).name, + } +) + + +def projection_input_roots(config: McpRuntimeConfig) -> list[Path]: + """The currently-existing watch roots, derived from what ``project_and_write`` reads. + + Only existing directories are returned (``watchfiles`` refuses missing paths); dirs + that appear later are picked up by the periodic watch-set refresh. See the module + docstring for the reader-by-reader derivation and the deliberate omissions. + """ + coordination_root = config.coordination_root + observer = observer_logs_root(coordination_root) + candidates: list[Path] = [ + coordination_root / "tasks", + observer / "lifecycles", + observer / "workspace", + drift_snapshot_dir(coordination_root), + coordination_root / "logs" / "providers" / "status", + coordination_root / "logs" / "providers" / "setup", + coordination_root / "temp" / "worktree-start", + coordination_root / "temp" / "tool-reports", + ] + # Deliberately NOT under coordination_root/worktrees: those trees carry each task's + # live provider-runtime (Postgres data, grepai indexes) which the daemon user cannot + # read and which churn on container writes -- watching them recursively both crashed + # the watcher (permission-denied on the container data dir) and would have re-projected + # on every WAL write. Worktree provider-state.json changes are infrequent and are + # heartbeat-covered; central provider status is already watched via logs/providers. + return [path for path in candidates if path.is_dir()] + + +def is_projection_input_event(path: str) -> bool: + """Filter one raw watch event down to genuine projection inputs. + + Drops atomic-write temp files, dot-prefixed sidecar temps, the projection's own + outputs, and the ``workspace/`` non-input churn (see :data:`_EXCLUDED_WORKSPACE_NAMES`). + """ + name = os.path.basename(path.rstrip("/\\")) + if name.endswith(".tmp") or name.startswith("."): + return False + if name in _EXCLUDED_NAMES: + return False + if name in _EXCLUDED_WORKSPACE_NAMES: + parent = os.path.basename(os.path.dirname(path)) + if parent == "workspace": + return False + return True + + +class WakeTarget(Protocol): + """What a watcher drives: the pacer's change/health inputs (structural, test-fakeable).""" + + def notify_change(self) -> None: ... + + def set_watcher_healthy(self, healthy: bool) -> None: ... + + +class ChangeWatch(Protocol): + """Structural contract for projection change watchers (the projector's seam). + + Mirrors ``LandingStateRefresh``: the projector owns the task lifecycle, tests inject + fakes, and :class:`ProjectionInputWatcher` is the live implementation. + """ + + async def run(self, pacer: WakeTarget) -> None: ... + + +class ChangePacer: + """The wake scheduler: change-or-heartbeat waiting with debounce, max-delay and floor. + + One instance belongs to one projector run loop. The watcher task feeds it + (:meth:`notify_change` / :meth:`set_watcher_healthy`); the run loop awaits + :meth:`wait` once per tick. Scheduling rules (all monotonic-clock): + + * **floor** -- never two projections closer than ``interval`` apart (``--interval`` + keeps its meaning as the fast-path cadence floor, so ``interval=100`` test + projectors stay quiet exactly as before); + * **debounce** -- a change projects ``debounce`` after the *last* change of its burst; + * **max delay** -- a sustained burst still projects within ``max_delay`` (= ``interval``) + of its *first* change (R2: the busy world keeps the former cadence); + * **heartbeat** -- with no changes, project every ``heartbeat`` seconds (R3/R4); + * **degraded** -- while the watcher is unhealthy, tick at the fixed ``interval`` + exactly like the pre-adaptive loop (R7 fail-open). + """ + + def __init__( + self, + *, + interval: float, + heartbeat: float = DEFAULT_HEARTBEAT_SECONDS, + debounce: float = DEBOUNCE_SECONDS, + max_delay: float | None = None, + ) -> None: + self._interval = max(interval, 0.0) + self._heartbeat = max(heartbeat, self._interval) + self._debounce = min(debounce, self._interval) if self._interval > 0 else debounce + self._max_delay = max_delay if max_delay is not None else self._interval + self._event = asyncio.Event() + self._first_pending: float | None = None + self._last_pending: float | None = None + # Start degraded: until the watcher reports its watches established, pace at the + # fixed interval so there is no detection blind spot at boot (heartbeat would be + # too slow a floor for changes racing the watcher startup). + self._watcher_healthy = False + self._last_wake = time.monotonic() + + def notify_change(self) -> None: + """Record a (batch of) input change(s); wakes :meth:`wait` to reschedule.""" + now = time.monotonic() + if self._first_pending is None: + self._first_pending = now + self._last_pending = now + self._event.set() + + def set_watcher_healthy(self, healthy: bool) -> None: + """Flip between change-driven and fixed-interval (degraded) pacing.""" + self._watcher_healthy = healthy + self._event.set() + + def _next_deadline(self) -> tuple[float, str]: + """Pure scheduling core: (monotonic deadline, wake reason) for the current state.""" + floor = self._last_wake + self._interval + if not self._watcher_healthy: + return floor, "interval" + if self._first_pending is not None and self._last_pending is not None: + settle = self._last_pending + self._debounce + bound = self._first_pending + self._max_delay + return min(max(settle, floor), max(bound, floor)), "change" + return self._last_wake + self._heartbeat, "heartbeat" + + async def wait(self) -> str: + """Sleep until the next projection is due; returns the wake reason. + + Pending changes are consumed at wake: changes observed *during* the following + projection accumulate for the next cycle, so nothing is ever lost to a tick. + """ + while True: + now = time.monotonic() + deadline, reason = self._next_deadline() + if now >= deadline: + self._first_pending = None + self._last_pending = None + self._event.clear() + self._last_wake = now + return reason + self._event.clear() + try: + async with asyncio.timeout(deadline - now): + await self._event.wait() + except TimeoutError: + pass + + +class ProjectionInputWatcher: + """Watch the projection-input roots and drive a :class:`ChangePacer`. + + Lifecycle mirrors the landing refresher: created by ``create_app`` for live serving + (never for ``--sim`` -- replay must stay time-driven, its feeder only writes *inside* + a tick), started/cancelled by ``Projector.run``. One watch failure never kills the + projector: the pacer drops to fixed-interval ticking (loudly) and the watch is + retried every :data:`WATCH_REFRESH_SECONDS`. + """ + + def __init__( + self, + config: McpRuntimeConfig, + *, + refresh_seconds: float = WATCH_REFRESH_SECONDS, + ) -> None: + self._config = config + self._refresh_seconds = refresh_seconds + + async def run(self, pacer: WakeTarget) -> None: + """Watch forever, feeding ``pacer``; degrade loudly on any failure (R7).""" + if watchfiles is None: + logger.error( + "watchfiles is not installed: change-driven projection is DISABLED and the " + "projector falls back to fixed-interval ticking (install the 'watchfiles' " + "dependency to restore adaptive pacing)" + ) + pacer.set_watcher_healthy(False) + return + established_before = False + while True: + try: + # Root derivation sits INSIDE the retry guard: a transient stat/glob failure + # must follow the same loud degrade-and-retry path as a watch failure -- not + # escape run() and kill the task for good (losing the periodic self-heal). + roots = await asyncio.to_thread(projection_input_roots, self._config) + if not roots: + # A fresh/empty coordination tree: nothing to watch yet. Not an error -- + # fixed-interval pacing until the first input dir appears. + pacer.set_watcher_healthy(False) + await asyncio.sleep(self._refresh_seconds) + continue + # Returns when the root set changed (restart with the fresh set). + await self._watch_once(roots, pacer, reconcile=established_before) + except asyncio.CancelledError: + raise + except Exception: + logger.exception( + "projection input watcher FAILED; falling back to fixed-interval " + "ticking until the watch re-establishes (retry in %.0fs)", + self._refresh_seconds, + ) + pacer.set_watcher_healthy(False) + await asyncio.sleep(self._refresh_seconds) + established_before = True + + async def _watch_once(self, roots: list[Path], pacer: WakeTarget, *, reconcile: bool) -> None: + """One watch generation over a fixed root set; returns when the set moved.""" + assert watchfiles is not None + stop = asyncio.Event() + refresher = asyncio.create_task(self._stop_when_roots_change(roots, stop)) + try: + pacer.set_watcher_healthy(True) + if reconcile: + # inotify has no replay: reconcile whatever happened while the watch + # was down or being re-registered with one debounced projection. + pacer.notify_change() + async for changes in watchfiles.awatch( + *roots, + watch_filter=self._watch_filter(), + debounce=_AWATCH_DEBOUNCE_MS, + step=_AWATCH_STEP_MS, + stop_event=stop, + recursive=True, + # Defence-in-depth: the watched roots are deliberately container-free, but an + # unreadable subdir must degrade to skipping it, never crash the whole watch. + ignore_permission_denied=True, + ): + if changes: + pacer.notify_change() + finally: + refresher.cancel() + try: + await refresher + except asyncio.CancelledError: + pass + except Exception: + logger.exception("watch-root refresher failed during watcher shutdown") + + def _watch_filter(self) -> Callable[[Change, str], bool]: + assert watchfiles is not None + default = watchfiles.DefaultFilter() + + def _filter(change: Change, path: str) -> bool: + return default(change, path) and is_projection_input_event(path) + + return _filter + + async def _stop_when_roots_change(self, current: list[Path], stop: asyncio.Event) -> None: + """Re-derive the root set periodically; stop the watch (to restart) when it moved.""" + while True: + await asyncio.sleep(self._refresh_seconds) + fresh = await asyncio.to_thread(projection_input_roots, self._config) + if fresh != current: + stop.set() + return diff --git a/mcp/src/agents_remember/serving/daemon.py b/mcp/src/agents_remember/serving/daemon.py index 2cdafa63..6b99ba4a 100644 --- a/mcp/src/agents_remember/serving/daemon.py +++ b/mcp/src/agents_remember/serving/daemon.py @@ -173,6 +173,7 @@ def spawn( port: int, version: str, interval: float = 1.0, + heartbeat: float | None = None, ) -> DaemonState: """Launch the detached foreground CLI and record it immediately. @@ -198,8 +199,10 @@ def spawn( str(port), "--interval", str(interval), - "--no-access-log", ] + if heartbeat is not None: + command += ["--heartbeat", str(heartbeat)] + command.append("--no-access-log") with log_path.open("ab") as log: process = subprocess.Popen( command, @@ -255,11 +258,12 @@ def ensure( port: int, version: str = SERVER_VERSION, interval: float = 1.0, + heartbeat: float | None = None, ) -> EnsureResult: """Adopt a healthy daemon, spawn a missing one, restart a mismatched one. - ``interval`` reaches the child only when this call spawns (or restarts) it; - an adopted daemon keeps the cadence it was started with. + ``interval`` / ``heartbeat`` reach the child only when this call spawns (or + restarts) it; an adopted daemon keeps the cadences it was started with. """ directory = daemon_dir(config) directory.mkdir(parents=True, exist_ok=True) @@ -275,7 +279,13 @@ def ensure( ) try: return _ensure_locked( - config, directory, host=host, port=port, version=version, interval=interval + config, + directory, + host=host, + port=port, + version=version, + interval=interval, + heartbeat=heartbeat, ) finally: fcntl.flock(lock, fcntl.LOCK_UN) @@ -289,6 +299,7 @@ def _ensure_locked( port: int, version: str, interval: float, + heartbeat: float | None = None, ) -> EnsureResult: current, alive = probe(directory) mismatch = "" @@ -301,7 +312,9 @@ def _ensure_locked( ) mismatch = _describe_mismatch(current, host=host, port=port, version=version) stop(directory) - state = spawn(config, host=host, port=port, version=version, interval=interval) + state = spawn( + config, host=host, port=port, version=version, interval=interval, heartbeat=heartbeat + ) if not _wait_ready(state): if _pid_alive(state.pid): detail = ( diff --git a/mcp/src/agents_remember/serving/projector.py b/mcp/src/agents_remember/serving/projector.py index 022fe584..f7aeeca1 100644 --- a/mcp/src/agents_remember/serving/projector.py +++ b/mcp/src/agents_remember/serving/projector.py @@ -1,10 +1,20 @@ """The shared projection projector: one tick loop fanned out to every client. -A single background task ticks :func:`project_and_write` on an interval (writing the -atomic ``latest-state.json`` as a side benefit), diffs each new projection against the -last, and broadcasts the per-entity deltas to all subscribed SSE connections. N clients -therefore cost one re-projection per tick -- what makes the single multiplexed -EventSource (note 09) scale. Reads go only through ``McpRuntimeConfig`` (North-Star #5). +A single background task ticks :func:`project_and_write` (writing the atomic +``latest-state.json`` as a side benefit), diffs each new projection against the last, and +broadcasts the per-entity deltas to all subscribed SSE connections. N clients therefore +cost one re-projection per tick -- what makes the single multiplexed EventSource (note 09) +scale. Reads go only through ``McpRuntimeConfig`` (North-Star #5). + +Adaptive waking (260712-PTS-L3): with a ``change_watcher`` the pacemaker is no longer an +unconditional ``sleep(interval)`` -- the loop wakes on debounced input changes (floored to +one projection per ``interval``; ``--interval`` keeps meaning the fast-path cadence floor) +or on a slow ``heartbeat`` when nothing changed, so a quiet daemon idles near zero CPU. +Freshness bounds: change -> SSE delta within debounce + projection time (plus the interval +floor when busy); ``/api/state`` staleness and time-derived field resolution are bounded +by the heartbeat. Without a watcher (sim replay, existing tests) the loop keeps the exact +fixed-interval behaviour, and a failed watcher degrades back to it loudly (fail-open). +The tick body itself is byte-identical either way -- see ``serving/change_watcher.py``. Two seams keep this generic across live and sim (slice 4b): ``now`` is the clock the tick projects at (a replay clock under sim, wall-clock UTC live), and ``before_tick`` is @@ -31,6 +41,7 @@ from uuid import uuid4 from agents_remember.observer.projection_store import project_and_write +from agents_remember.serving.change_watcher import DEFAULT_HEARTBEAT_SECONDS, ChangePacer from agents_remember.serving.delta import ( DeltaEvent, StableProjectionState, @@ -43,6 +54,7 @@ from agents_remember.observer.landing_state import LandingStateRefresh from agents_remember.observer.projection import WorkspaceProjection from agents_remember.observer.projection_store import ProviderStateRefresh + from agents_remember.serving.change_watcher import ChangeWatch logger = logging.getLogger(__name__) @@ -54,6 +66,24 @@ def _utcnow() -> datetime: return datetime.now(UTC) +async def _shutdown_task(task: asyncio.Task[None] | None, label: str) -> None: + """Cancel-and-await one background task at projector shutdown. + + Both background tasks log their own cycle failures. This guard is still required for + an already-dead task: its stored exception must not replace the Projector cancellation + and skip the serving lifespan's terminal-host cleanup. + """ + if task is None: + return + task.cancel() + try: + await task + except asyncio.CancelledError: + pass + except Exception: + logger.exception("%s task failed before projector shutdown", label) + + class Projector: """Owns the latest projection, a monotonic sequence, and the subscriber fan-out.""" @@ -62,10 +92,12 @@ def __init__( config: McpRuntimeConfig, *, interval: float = 1.0, + heartbeat: float | None = None, now: Callable[[], datetime] | None = None, before_tick: Callable[[datetime], object] | None = None, provider_refresher: ProviderStateRefresh | None = None, landing_refresher: LandingStateRefresh | None = None, + change_watcher: ChangeWatch | None = None, ) -> None: self._config = config self._interval = interval @@ -73,6 +105,23 @@ def __init__( self._before_tick = before_tick self._provider_refresher = provider_refresher self._landing_refresher = landing_refresher + # Adaptive waking (260712-PTS-L3): with a watcher, the run loop paces via the + # ChangePacer (change-driven + heartbeat, floored to one tick per ``interval``). + # Without one -- sim replay and the injected-now() tests -- it keeps the exact + # legacy ``sleep(interval)`` pacemaker. + self._change_watcher = change_watcher + self._pacer: ChangePacer | None = ( + ChangePacer( + interval=interval, + heartbeat=heartbeat if heartbeat is not None else DEFAULT_HEARTBEAT_SECONDS, + ) + if change_watcher is not None + else None + ) + # Instrumentation (R7 tests + ops): successful projections since run() started, + # and why the last tick woke ("change" | "heartbeat" | "interval"). + self.projection_count = 0 + self.last_wake_reason: str | None = None # (seq, projection) publish as ONE tuple: /api/state runs in a threadpool, so pairing # them via two attributes could tear (a bumped seq read against the previous snapshot # would hand a poller a stale body under a fresh ETag). @@ -98,20 +147,34 @@ async def prime(self) -> None: self._published = (0, first) async def run(self) -> None: - """Tick forever: re-project, diff, broadcast. One bad tick never kills the loop.""" + """Tick on wake: re-project, diff, broadcast. One bad tick never kills the loop.""" landing_task = ( asyncio.create_task(self._landing_refresher.run()) if self._landing_refresher is not None else None ) + watch_task = ( + asyncio.create_task(self._change_watcher.run(self._pacer)) + if self._change_watcher is not None and self._pacer is not None + else None + ) + if watch_task is not None: + # A watcher that dies (or returns) must not leave the pacer believing changes + # are still detected -- that would silently stretch every wake to the heartbeat. + # Fail-open (R7): drop to fixed-interval ticking and say so loudly. + watch_task.add_done_callback(self._on_watch_task_done) try: while True: - await asyncio.sleep(self._interval) + if self._pacer is None: + await asyncio.sleep(self._interval) + else: + self.last_wake_reason = await self._pacer.wait() try: current = await asyncio.to_thread(self._tick_sync, self._now()) except Exception: logger.exception("projection tick failed; retrying next interval") continue + self.projection_count += 1 current_stable = stable_projection_state(current) seq, previous = self._published for delta in diff_projection( @@ -125,17 +188,21 @@ async def run(self) -> None: self._latest_stable = current_stable self._published = (seq, current) finally: - if landing_task is not None: - landing_task.cancel() - try: - await landing_task - except asyncio.CancelledError: - pass - except Exception: - # The refresher logs cycle failures itself. This shutdown guard is still - # required for an already-dead task: its stored exception must not replace the - # Projector cancellation and skip the serving lifespan's terminal-host cleanup. - logger.exception("landing refresher task failed before projector shutdown") + await _shutdown_task(watch_task, "change watcher") + await _shutdown_task(landing_task, "landing refresher") + + def _on_watch_task_done(self, task: asyncio.Task[None]) -> None: + """R7 fail-open: a finished watcher task degrades pacing to the fixed interval.""" + if self._pacer is None or task.cancelled(): + return + exception = task.exception() + if exception is not None: + logger.error( + "change watcher task died; falling back to fixed-interval ticking every %.1fs", + self._interval, + exc_info=exception, + ) + self._pacer.set_watcher_healthy(False) def _tick_sync(self, moment: datetime) -> WorkspaceProjection: """Run the optional pre-tick hook, then project at ``moment`` (off the loop thread).""" diff --git a/mcp/src/agents_remember/worktrees/git_worktree_manager.py b/mcp/src/agents_remember/worktrees/git_worktree_manager.py index f4ae4d7e..2bad2cb5 100644 --- a/mcp/src/agents_remember/worktrees/git_worktree_manager.py +++ b/mcp/src/agents_remember/worktrees/git_worktree_manager.py @@ -21,6 +21,7 @@ command_attach, command_cleanup, command_closeout, + command_heal_leaf_ids, command_integrate, command_start, command_status, @@ -103,6 +104,7 @@ status_result, ) from agents_remember.worktrees.modules.sync import sync_result +from agents_remember.worktrees.worktree_contract import heal_contract_leaf_ids __all__ = [ "ENTITY_FINGERPRINT_ALGORITHM", @@ -124,6 +126,7 @@ "command_attach", "command_cleanup", "command_closeout", + "command_heal_leaf_ids", "command_integrate", "command_start", "command_status", @@ -147,6 +150,7 @@ "finalize_result", "has_changes", "head_commit", + "heal_contract_leaf_ids", "integrate_result", "integration_branch", "is_ancestor", diff --git a/mcp/src/agents_remember/worktrees/leaf_refs.py b/mcp/src/agents_remember/worktrees/leaf_refs.py index beb34e13..b154baa1 100644 --- a/mcp/src/agents_remember/worktrees/leaf_refs.py +++ b/mcp/src/agents_remember/worktrees/leaf_refs.py @@ -135,6 +135,19 @@ def resolve_leaf_ref( ) +def canonical_leaf_doc_ids(repo_name: str, task_root: Path) -> frozenset[str]: + """The doc ids provable for one task root — the heal fast-path index (260712-PTS-L1). + + One bounded ``*.json`` scan of ``task_root`` (never the whole tasks tree): a leaf + contract whose ``leaf_id`` already is one of these ids is canonical and needs no + resolution walk. ``repo_name`` only shapes the internal qualified ids, never the + returned doc ids. + """ + return frozenset( + candidate.doc_id for candidate in _leaf_candidates_for_root(repo_name, task_root) + ) + + def leaf_ref_enclosure_aliases( coordination_root: Path, repo_name: str, diff --git a/mcp/src/agents_remember/worktrees/modules/cli.py b/mcp/src/agents_remember/worktrees/modules/cli.py index 853abf3e..57c16868 100644 --- a/mcp/src/agents_remember/worktrees/modules/cli.py +++ b/mcp/src/agents_remember/worktrees/modules/cli.py @@ -10,7 +10,7 @@ from agents_remember.worktrees.modules.closeout import closeout_result from agents_remember.worktrees.modules.integrate import integrate_result from agents_remember.worktrees.modules.start import attach_result, start_result, status_result -from agents_remember.worktrees.worktree_contract import ContractError +from agents_remember.worktrees.worktree_contract import ContractError, heal_contract_leaf_ids def parse_json_stdout(stdout: str) -> object: @@ -59,6 +59,17 @@ def command_cleanup(args: argparse.Namespace) -> int: return result.returncode +def command_heal_leaf_ids(args: argparse.Namespace) -> int: + """The deliberate on-demand seam for :func:`heal_contract_leaf_ids` (260712-PTS-L1). + + Healing legacy stem-shaped leaf ids is a one-shot migration walk, never a per-read + side effect — run this once against a coordination root (or at daemon startup) + instead of relying on ``load_contract`` to normalize.""" + report = heal_contract_leaf_ids(args.coordination_root, dry_run=args.dry_run) + print(json.dumps(report, indent=2)) + return 0 + + def add_common(parser: argparse.ArgumentParser) -> None: parser.add_argument("--code-repository-name", help="Code repository name to resolve.") parser.add_argument( @@ -131,6 +142,17 @@ def build_parser() -> argparse.ArgumentParser: cleanup.add_argument("--approved", action="store_true") cleanup.add_argument("--dry-run", action="store_true") cleanup.set_defaults(func=command_cleanup) + + heal = subparsers.add_parser( + "heal-leaf-ids", + help=( + "Rewrite legacy stem-shaped leaf ids to canonical doc ids across the " + "active leaf enclosures (one-shot, idempotent, loud)." + ), + ) + heal.add_argument("--coordination-root", type=Path, required=True) + heal.add_argument("--dry-run", action="store_true") + heal.set_defaults(func=command_heal_leaf_ids) return parser diff --git a/mcp/src/agents_remember/worktrees/worktree_contract.py b/mcp/src/agents_remember/worktrees/worktree_contract.py index 26ce1249..57b4420b 100644 --- a/mcp/src/agents_remember/worktrees/worktree_contract.py +++ b/mcp/src/agents_remember/worktrees/worktree_contract.py @@ -8,14 +8,20 @@ from __future__ import annotations import json +import logging from dataclasses import dataclass, field, replace from pathlib import Path from agents_remember.errors import AgentsRememberError -from agents_remember.worktrees.leaf_refs import LeafRefResolutionError, resolve_leaf_ref +from agents_remember.worktrees.leaf_refs import ( + LeafRefResolutionError, + canonical_leaf_doc_ids, + resolve_leaf_ref, +) from agents_remember.worktrees.task_resolver import ( SERIES_CONTRACT_FILENAME, TaskResolutionError, + iter_leaf_enclosure_contracts, leaf_enclosure_path, resolve_active_task_root, series_contract_path, @@ -23,6 +29,8 @@ task_root_for, ) +logger = logging.getLogger(__name__) + CONTRACT_SCHEMA = "ar-series-contract/v1" VALID_MEMORY_MODES = {"internal", "external", "disabled"} VALID_KINDS = {"series", "leaf"} @@ -212,11 +220,18 @@ def default_series_contract( def load_contract(path: Path) -> WorktreeContract: + """Read + parse + validate one contract file — deliberately walk-free (260712-PTS-L1). + + The read path performs zero tasks-tree traversal: no leaf-ref resolution, no + series-contract iteration, no glob. A legacy stem-shaped ``leaf_id`` is returned + verbatim; normalization is a write-time concern (``write_contract``) plus the + explicit :func:`heal_contract_leaf_ids` migration for contracts already on disk. + """ if not path.exists(): raise ContractError(f"worktree contract does not exist: {path}") front_matter = _extract_front_matter(path.read_text(encoding="utf-8")) data = _parse_limited_yaml(front_matter) - contract = normalize_contract_leaf_id(_contract_from_data(data, path), keep_unresolved=True) + contract = _contract_from_data(data, path) validate_contract(contract) return contract @@ -228,6 +243,84 @@ def write_contract(path: Path, contract: WorktreeContract) -> None: path.write_text(contract_to_text(contract), encoding="utf-8") +def heal_contract_leaf_ids(coordination_root: Path, *, dry_run: bool = False) -> dict[str, object]: + """Heal legacy stem-shaped leaf ids across the active leaf-enclosure population. + + The deliberate, one-shot successor to the per-read normalization ``load_contract`` + used to run (260712-PTS-L1): walk ``tasks/`` once via + ``iter_leaf_enclosure_contracts`` — the exact population the projection readers + consume — map each legacy ``leaf_id`` to its canonical doc id through the same + resolution the read path used to apply (``normalize_contract_leaf_id``), and + rewrite only the contracts whose id actually changes. Idempotent and cheap on + re-run: a contract whose ``leaf_id`` already is a doc id of the task root its + enclosure lives in is skipped via a per-root index (one bounded ``*.json`` scan, + cached per root) without any resolution walk. Loud by design: every rewrite logs + one line here and lands in the returned report; an unresolvable or malformed + entry is reported, never fatal to the sweep. Never invoked implicitly by any + read path — reach it through the ``heal-leaf-ids`` CLI command or call it + directly (e.g. once at daemon startup). + """ + healed: list[dict[str, str]] = [] + errors: list[dict[str, str]] = [] + canonical = 0 + unchanged = 0 + doc_id_index: dict[Path, frozenset[str]] = {} + for path in iter_leaf_enclosure_contracts(coordination_root / "tasks"): + try: + contract = load_contract(path) + except (ContractError, OSError) as error: + errors.append({"contract": path.as_posix(), "error": str(error)}) + continue + if contract.kind != "leaf" or not contract.leaf_id: + unchanged += 1 + continue + # The task root the enclosure PHYSICALLY lives in (enclosures//series-contract.md), + # not the recorded one: a stale recorded root must degrade to the slow path, not a skip. + task_root = path.parent.parent.parent + try: + known = doc_id_index.get(task_root) + if known is None: + known = canonical_leaf_doc_ids(contract.repo_name, task_root) + doc_id_index[task_root] = known + if contract.leaf_id in known: + canonical += 1 + continue + healed_contract = normalize_contract_leaf_id(contract, keep_unresolved=True) + except Exception as error: # one malformed task doc never aborts the sweep + errors.append({"contract": path.as_posix(), "error": str(error)}) + continue + if healed_contract.leaf_id == contract.leaf_id: + unchanged += 1 + continue + logger.info( + "heal_contract_leaf_ids: %s leaf_id %r -> %r%s", + path.as_posix(), + contract.leaf_id, + healed_contract.leaf_id, + " (dry run)" if dry_run else "", + ) + if not dry_run: + try: + write_contract(path, healed_contract) + except Exception as error: # a failed rewrite is reported, never fatal to the sweep + errors.append({"contract": path.as_posix(), "error": str(error)}) + continue + healed.append( + { + "contract": path.as_posix(), + "from": contract.leaf_id, + "to": healed_contract.leaf_id, + } + ) + return { + "healed": healed, + "canonical": canonical, + "unchanged": unchanged, + "errors": errors, + "dryRun": dry_run, + } + + def normalize_contract_leaf_id( contract: WorktreeContract, *, keep_unresolved: bool = False ) -> WorktreeContract: diff --git a/mcp/tests/test_change_watcher.py b/mcp/tests/test_change_watcher.py new file mode 100644 index 00000000..c5524292 --- /dev/null +++ b/mcp/tests/test_change_watcher.py @@ -0,0 +1,438 @@ +"""260712-PTS-L3: change-driven projection pacing. + +Covers the watcher module's three layers plus the projector integration: + +* the derived watch-root list (R1) and the input-event filter (self-trigger safety); +* the :class:`ChangePacer` scheduling core (debounce, max-delay bound, interval floor, + heartbeat, degraded mode) -- deterministic via the pure ``_next_deadline``; +* the projector run loop with an injected watcher (R7): a quiet world projects only at + heartbeat cadence, a write burst coalesces to a bounded projection count, a single + write projects within the debounce bound, and watcher failure degrades loudly to + fixed-interval ticking (fail-open); +* one real-``watchfiles`` end-to-end pass (inotify -> debounce -> projection). +""" + +from __future__ import annotations + +import asyncio +import contextlib +import tempfile +import time +import unittest +from pathlib import Path +from unittest import mock + +from agents_remember.mcp.config import McpRuntimeConfig +from agents_remember.serving import change_watcher as change_watcher_module +from agents_remember.serving.change_watcher import ( + ChangePacer, + ProjectionInputWatcher, + WakeTarget, + is_projection_input_event, + projection_input_roots, +) +from agents_remember.serving.projector import Projector + + +def _config(tmp: Path) -> McpRuntimeConfig: + return McpRuntimeConfig( + config_path=tmp / "settings.json", + coordination_root=tmp, + workspace_root=tmp, + transcript_root=tmp / "logs" / "mcp", + ) + + +class ProjectionInputRootsTests(unittest.TestCase): + def setUp(self) -> None: + self._dir = tempfile.TemporaryDirectory() + self.tmp = Path(self._dir.name) + + def tearDown(self) -> None: + self._dir.cleanup() + + def test_empty_tree_yields_no_roots(self) -> None: + self.assertEqual(projection_input_roots(_config(self.tmp)), []) + + def test_derives_the_projection_input_surfaces_and_nothing_else(self) -> None: + observer = self.tmp / "logs" / "observer" + expected = [ + self.tmp / "tasks", + observer / "lifecycles", + observer / "workspace", + observer / "drift", + self.tmp / "logs" / "providers" / "status", + self.tmp / "logs" / "providers" / "setup", + self.tmp / "temp" / "worktree-start", + self.tmp / "temp" / "tool-reports", + ] + for path in expected: + path.mkdir(parents=True) + # A worktree group: NOTHING under worktrees/ is a watch root. Its provider-runtime + # holds live container data (Postgres/grepai) that is unreadable to the daemon user + # and churns on every container write -- watching it crashed the watcher and would + # have re-projected on WAL writes. provider-state.json changes are heartbeat-covered. + runtime = self.tmp / "worktrees" / "repo" / "group-ar" / "provider-runtime" + runtime.mkdir(parents=True) + checkout = self.tmp / "worktrees" / "repo" / "group-ar" / "src" + checkout.mkdir(parents=True) + # Non-input observer dirs stay unwatched (metrics churn, quarantines). + (observer / "providers").mkdir(parents=True) + (observer / "quarantine-1").mkdir(parents=True) + + roots = projection_input_roots(_config(self.tmp)) + self.assertEqual(roots, expected) + # Regression guard: no watch root may ever fall under worktrees/ (container data). + self.assertFalse(any((self.tmp / "worktrees") in root.parents for root in roots)) + self.assertNotIn(runtime, roots) + self.assertNotIn(checkout, roots) + self.assertNotIn(observer / "providers", roots) + + def test_missing_surfaces_are_skipped_until_they_exist(self) -> None: + (self.tmp / "tasks").mkdir() + self.assertEqual(projection_input_roots(_config(self.tmp)), [self.tmp / "tasks"]) + + +class InputEventFilterTests(unittest.TestCase): + def test_projection_inputs_pass(self) -> None: + for path in ( + "/c/logs/observer/lifecycles/L1/events.jsonl", + "/c/logs/observer/lifecycles/L1/heartbeat.json", + "/c/logs/observer/lifecycles/L1/gates.jsonl", + "/c/logs/observer/workspace/gates.jsonl", + "/c/logs/observer/workspace/operator-inbox.jsonl", + "/c/logs/observer/workspace/attention-dismissals.jsonl", + "/c/logs/observer/workspace/expectation-rows.jsonl", + "/c/logs/observer/drift/repo.json", + "/c/tasks/agents-remember/260712_x/task.json", + ): + self.assertTrue(is_projection_input_event(path), path) + + def test_self_writes_and_non_input_churn_are_dropped(self) -> None: + for path in ( + # The projection's own per-tick outputs: letting these through would make + # every tick re-wake the next one (an infinite debounce-paced loop). + "/c/logs/observer/latest-state.json", + "/c/logs/observer/latest-metrics.json", + "/c/logs/observer/latest-state.json.01HZX.tmp", + "/c/tasks/agents-remember/task.json.01HZX.tmp", + # workspace/ non-inputs: raw river + cursor/locks, supervisor heartbeat. + "/c/logs/observer/workspace/events.jsonl", + "/c/logs/observer/workspace/events.cursor.json", + "/c/logs/observer/workspace/events.lock", + # Created "a+b" by every inbox access (incl. each tick's read_agent_pickups): + # its boot-time creation must not cost a spurious change-tick (review F1). + "/c/logs/observer/workspace/operator-inbox.lock", + "/c/logs/observer/workspace/supervisor-heartbeat.json", + "/c/logs/observer/workspace/.events.cursor.json.123.tmp", + ): + self.assertFalse(is_projection_input_event(path), path) + + def test_lifecycle_events_are_not_confused_with_the_workspace_river(self) -> None: + # Same basename as the river, different (input) directory. + self.assertTrue(is_projection_input_event("/c/logs/observer/lifecycles/L1/events.jsonl")) + + +class ChangePacerDeadlineTests(unittest.TestCase): + """The pure scheduling core, driven with synthetic monotonic instants.""" + + def _pacer(self, *, interval: float = 1.0, heartbeat: float = 15.0) -> ChangePacer: + pacer = ChangePacer(interval=interval, heartbeat=heartbeat, debounce=0.1) + pacer._last_wake = 100.0 + pacer._watcher_healthy = True + return pacer + + def test_idle_world_waits_for_the_heartbeat(self) -> None: + pacer = self._pacer() + deadline, reason = pacer._next_deadline() + self.assertEqual((deadline, reason), (115.0, "heartbeat")) + + def test_lone_change_projects_after_the_debounce(self) -> None: + pacer = self._pacer() + pacer._first_pending = pacer._last_pending = 105.0 + deadline, reason = pacer._next_deadline() + self.assertEqual((deadline, reason), (105.1, "change")) + + def test_change_is_floored_to_one_projection_per_interval(self) -> None: + pacer = self._pacer() + # Change lands 50ms after the last wake: the floor (interval=1.0) wins. + pacer._first_pending = pacer._last_pending = 100.05 + deadline, reason = pacer._next_deadline() + self.assertEqual((deadline, reason), (101.0, "change")) + + def test_sustained_burst_is_bounded_by_max_delay(self) -> None: + pacer = self._pacer() + # First change at 105.0, still being extended at 105.95: the max-delay bound + # (interval-sized, R2) beats the ever-sliding settle window. + pacer._first_pending = 105.0 + pacer._last_pending = 105.95 + deadline, reason = pacer._next_deadline() + self.assertEqual((deadline, reason), (106.0, "change")) + + def test_degraded_watcher_ticks_at_the_fixed_interval(self) -> None: + pacer = self._pacer() + pacer._watcher_healthy = False + pacer._first_pending = pacer._last_pending = 100.2 + deadline, reason = pacer._next_deadline() + self.assertEqual((deadline, reason), (101.0, "interval")) + + def test_heartbeat_never_undercuts_the_interval(self) -> None: + pacer = ChangePacer(interval=10.0, heartbeat=1.0) + pacer._last_wake = 100.0 + pacer._watcher_healthy = True + deadline, _ = pacer._next_deadline() + self.assertEqual(deadline, 110.0) + + +class _FakeWatcher: + """A ChangeWatch fake: reports healthy, then emits on demand.""" + + def __init__(self) -> None: + self.pacer: WakeTarget | None = None + self.started = asyncio.Event() + self.stopped = asyncio.Event() + + async def run(self, pacer: WakeTarget) -> None: + self.pacer = pacer + pacer.set_watcher_healthy(True) + self.started.set() + try: + await asyncio.Event().wait() + finally: + self.stopped.set() + + def emit(self) -> None: + assert self.pacer is not None + self.pacer.notify_change() + + +class _CrashingWatcher: + async def run(self, pacer: WakeTarget) -> None: + pacer.set_watcher_healthy(True) + raise RuntimeError("watch backend exploded") + + +class AdaptiveProjectorTests(unittest.IsolatedAsyncioTestCase): + def setUp(self) -> None: + self._dir = tempfile.TemporaryDirectory() + self.tmp = Path(self._dir.name) + + def tearDown(self) -> None: + self._dir.cleanup() + + async def _run_projector(self, projector: Projector) -> asyncio.Task[None]: + task = asyncio.create_task(projector.run()) + + async def _cancel() -> None: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + self.addAsyncCleanup(_cancel) + return task + + async def _wait_for(self, predicate, timeout: float) -> None: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return + await asyncio.sleep(0.01) + self.fail(f"condition not reached within {timeout:.1f}s") + + async def test_quiet_world_projects_only_at_heartbeat_cadence(self) -> None: + watcher = _FakeWatcher() + projector = Projector( + _config(self.tmp), interval=0.02, heartbeat=0.2, change_watcher=watcher + ) + await self._run_projector(projector) + await asyncio.wait_for(watcher.started.wait(), timeout=1) + start_count = projector.projection_count + await asyncio.sleep(0.65) + beats = projector.projection_count - start_count + # ~3 heartbeats in 0.65s at 0.2s cadence; the former behaviour (interval pacing) + # would have produced ~32. Loose bounds absorb slow-CI tick durations. + self.assertGreaterEqual(beats, 1) + self.assertLessEqual(beats, 5) + self.assertEqual(projector.last_wake_reason, "heartbeat") + + async def test_single_change_projects_within_the_debounce_bound(self) -> None: + watcher = _FakeWatcher() + projector = Projector( + _config(self.tmp), interval=0.02, heartbeat=30.0, change_watcher=watcher + ) + await self._run_projector(projector) + await asyncio.wait_for(watcher.started.wait(), timeout=1) + # Let boot-time ticks (degraded-start + floor) drain, then measure a lone change. + await asyncio.sleep(0.2) + baseline = projector.projection_count + emitted_at = time.monotonic() + watcher.emit() + await self._wait_for(lambda: projector.projection_count > baseline, timeout=2.0) + latency = time.monotonic() - emitted_at + self.assertEqual(projector.last_wake_reason, "change") + # debounce (clamped to interval=0.02) + one projection + generous CI slack. + self.assertLess(latency, 1.0) + + async def test_burst_coalesces_to_a_bounded_projection_count(self) -> None: + watcher = _FakeWatcher() + projector = Projector( + _config(self.tmp), interval=0.15, heartbeat=30.0, change_watcher=watcher + ) + await self._run_projector(projector) + await asyncio.wait_for(watcher.started.wait(), timeout=1) + await asyncio.sleep(0.3) # drain boot ticks; pass the first interval floor + baseline = projector.projection_count + for _ in range(25): + watcher.emit() + await asyncio.sleep(0.002) + await asyncio.sleep(0.7) + projected = projector.projection_count - baseline + # 25 writes -> one debounced projection (a second is tolerated if the burst + # straddles a floor boundary); unconditional 1-per-write would give 25. + self.assertGreaterEqual(projected, 1) + self.assertLessEqual(projected, 2) + self.assertEqual(projector.last_wake_reason, "change") + + async def test_missing_watchfiles_degrades_loudly_to_fixed_interval(self) -> None: + config = _config(self.tmp) + projector = Projector( + config, interval=0.05, heartbeat=30.0, change_watcher=ProjectionInputWatcher(config) + ) + with ( + mock.patch.object(change_watcher_module, "watchfiles", None), + self.assertLogs("agents_remember.serving.change_watcher", level="ERROR") as logs, + ): + await self._run_projector(projector) + await self._wait_for(lambda: projector.projection_count >= 3, timeout=3.0) + # Ticks at the fixed 0.05s interval despite the 30s heartbeat: fail-open (R7). + self.assertEqual(projector.last_wake_reason, "interval") + self.assertIn("watchfiles is not installed", "\n".join(logs.output)) + + async def test_crashed_watcher_task_degrades_loudly_to_fixed_interval(self) -> None: + projector = Projector( + _config(self.tmp), + interval=0.05, + heartbeat=30.0, + change_watcher=_CrashingWatcher(), + ) + with self.assertLogs("agents_remember.serving.projector", level="ERROR") as logs: + await self._run_projector(projector) + await self._wait_for(lambda: projector.projection_count >= 3, timeout=3.0) + self.assertEqual(projector.last_wake_reason, "interval") + self.assertIn("change watcher task died", "\n".join(logs.output)) + + async def test_root_derivation_failure_retries_instead_of_dying(self) -> None: + # Review F3: a transient stat/glob failure while deriving the watch roots must + # follow the same loud degrade-and-retry path as a watch failure -- not escape + # run() and permanently kill the watcher task (losing the periodic self-heal). + config = _config(self.tmp) + (self.tmp / "tasks").mkdir() + watcher = ProjectionInputWatcher(config, refresh_seconds=0.05) + + class RecordingPacer: + def __init__(self) -> None: + self.health: list[bool] = [] + self.changes = 0 + + def notify_change(self) -> None: + self.changes += 1 + + def set_watcher_healthy(self, healthy: bool) -> None: + self.health.append(healthy) + + real_roots = change_watcher_module.projection_input_roots + calls = 0 + + def flaky_roots(cfg: McpRuntimeConfig) -> list[Path]: + nonlocal calls + calls += 1 + if calls == 1: + raise RuntimeError("transient derivation failure") + return real_roots(cfg) + + pacer = RecordingPacer() + with ( + mock.patch.object(change_watcher_module, "projection_input_roots", flaky_roots), + self.assertLogs("agents_remember.serving.change_watcher", level="ERROR") as logs, + ): + task = asyncio.create_task(watcher.run(pacer)) + try: + # Degrades loudly on the failed derivation, then recovers on the retry + # cycle: a healthy=True report proves the watch re-established. + await self._wait_for(lambda: True in pacer.health, timeout=3.0) + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + self.assertIn(False, pacer.health) # the loud degrade came first + self.assertIn("projection input watcher FAILED", "\n".join(logs.output)) + + async def test_run_owns_the_watcher_task_lifecycle(self) -> None: + watcher = _FakeWatcher() + projector = Projector( + _config(self.tmp), interval=100, heartbeat=100, change_watcher=watcher + ) + task = asyncio.create_task(projector.run()) + await asyncio.wait_for(watcher.started.wait(), timeout=1) + task.cancel() + with self.assertRaises(asyncio.CancelledError): + await task + self.assertTrue(watcher.stopped.is_set()) + + async def test_without_a_watcher_the_legacy_interval_pacing_is_kept(self) -> None: + # The sim/replay and injected-now() path: no watcher, no pacer -- the loop must + # keep ticking unconditionally every interval exactly as before this change. + projector = Projector(_config(self.tmp), interval=0.05) + self.assertIsNone(projector._pacer) + await self._run_projector(projector) + await asyncio.sleep(0.4) + self.assertGreaterEqual(projector.projection_count, 2) + self.assertIsNone(projector.last_wake_reason) + + +class RealWatchfilesIntegrationTests(unittest.IsolatedAsyncioTestCase): + """One end-to-end pass over the real inotify backend (Linux CI + dev machines).""" + + def setUp(self) -> None: + self._dir = tempfile.TemporaryDirectory() + self.tmp = Path(self._dir.name) + + def tearDown(self) -> None: + self._dir.cleanup() + + async def test_a_real_write_under_tasks_triggers_a_change_projection(self) -> None: + if change_watcher_module.watchfiles is None: # pragma: no cover + self.skipTest("watchfiles not installed") + config = _config(self.tmp) + (self.tmp / "tasks").mkdir() + projector = Projector( + config, + interval=0.05, + heartbeat=60.0, + change_watcher=ProjectionInputWatcher(config), + ) + task = asyncio.create_task(projector.run()) + try: + # Give awatch time to register its watches, then let boot ticks drain. + await asyncio.sleep(0.8) + baseline = projector.projection_count + (self.tmp / "tasks" / "note.md").write_text("changed\n", encoding="utf-8") + deadline = time.monotonic() + 5.0 + while time.monotonic() < deadline: + if projector.projection_count > baseline: + break + await asyncio.sleep(0.02) + self.assertGreater( + projector.projection_count, + baseline, + "a write under tasks/ did not wake the projector within 5s", + ) + self.assertEqual(projector.last_wake_reason, "change") + finally: + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/mcp/tests/test_dashboard_daemon.py b/mcp/tests/test_dashboard_daemon.py index 4dcf7bfc..67cc42d8 100644 --- a/mcp/tests/test_dashboard_daemon.py +++ b/mcp/tests/test_dashboard_daemon.py @@ -192,12 +192,25 @@ def test_spawn_launches_the_detached_cli_and_records_state(self) -> None: self.assertEqual(command[command.index("--config") + 1], str(config.config_path)) self.assertEqual(command[command.index("--port") + 1], "9100") self.assertEqual(command[command.index("--interval") + 1], "1.0") + self.assertNotIn("--heartbeat", command) self.assertIn("--no-access-log", command) self.assertTrue(popen.call_args.kwargs["start_new_session"]) self.assertEqual(state.pid, 777) self.assertEqual(daemon.read_state(daemon.daemon_dir(config)), state) daemon._spawned.remove(fake) + def test_spawn_forwards_an_explicit_heartbeat_to_the_child(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + config = make_config(Path(tmp)) + fake = mock.Mock() + fake.pid = 779 + with mock.patch.object(daemon.subprocess, "Popen", return_value=fake) as popen: + daemon.spawn(config, host="127.0.0.1", port=9100, version="9.9.9", heartbeat=20.0) + command = popen.call_args.args[0] + self.assertEqual(command[command.index("--heartbeat") + 1], "20.0") + self.assertIn("--no-access-log", command) + daemon._spawned.remove(fake) + def test_spawn_rotates_the_previous_log(self) -> None: with tempfile.TemporaryDirectory() as tmp: config = make_config(Path(tmp)) @@ -232,7 +245,7 @@ def test_absent_daemon_is_started(self) -> None: self.assertEqual(result.action, "started") self.assertEqual(result.state, spawned) spawn.assert_called_once_with( - self.config, host="127.0.0.1", port=9000, version="1.0", interval=1.0 + self.config, host="127.0.0.1", port=9000, version="1.0", interval=1.0, heartbeat=None ) stop.assert_not_called() diff --git a/mcp/tests/test_leaf_ref_resolution.py b/mcp/tests/test_leaf_ref_resolution.py index b7e067da..9f1308e9 100644 --- a/mcp/tests/test_leaf_ref_resolution.py +++ b/mcp/tests/test_leaf_ref_resolution.py @@ -1,16 +1,25 @@ from __future__ import annotations +import contextlib +import io +import json import tempfile import unittest from dataclasses import replace from pathlib import Path +from typing import cast +from unittest import mock from agents_remember.tasks import TaskDocument, write_task_doc from agents_remember.worktrees.leaf_refs import LeafRefResolutionError, resolve_leaf_ref +from agents_remember.worktrees.modules.cli import main as worktree_cli_main from agents_remember.worktrees.worktree_contract import ( + WorktreeContract, contract_to_text, default_contract, + heal_contract_leaf_ids, load_contract, + normalize_contract_leaf_id, ) @@ -60,6 +69,31 @@ def _write_leaf( ) +def _persisted_legacy_contract( + root: Path, + *, + master: str = "260700_master", + leaf_id: str = "01_legacy-leaf", +) -> WorktreeContract: + """A leaf contract persisted with a legacy stem-shaped leaf id (pre-heal on-disk state).""" + contract = default_contract( + task_name=master, + repo_name="repo-a", + workflow_kind="light-task", + memory_mode="disabled", + coordination_root=root, + code_repo_path=root / "repo-a", + code_source_branch="main", + code_work_branch="ar/legacy", + code_base_commit="abc123", + worktree_name="legacy", + leaf_id=leaf_id, + ) + contract.contract_path.parent.mkdir(parents=True, exist_ok=True) + contract.contract_path.write_text(contract_to_text(contract), encoding="utf-8") + return contract + + class LeafRefResolutionTests(unittest.TestCase): def test_resolves_qualified_doc_id_doc_id_and_legacy_slug(self) -> None: with tempfile.TemporaryDirectory() as tmp: @@ -185,36 +219,18 @@ def test_malformed_sibling_artifact_json_is_ignored(self) -> None: self.assertEqual(resolved.qualified_id, "repo-a/260700_master/260700-L1") - def test_load_contract_ignores_sibling_artifact_json_while_normalizing(self) -> None: + def test_load_contract_returns_legacy_leaf_id_verbatim(self) -> None: + """260712-PTS-L1 R1/R5: the read path never heals — the raw stem id comes back.""" with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) _write_leaf(root) - task_root = root / "tasks" / "repo-a" / "260700_master" - (task_root / "lint-inventory-baseline.json").write_text( - '{"summary":{"findings":14},"items":[{"path":"a.py","line":1}]}', - encoding="utf-8", - ) - contract = default_contract( - task_name="260700_master", - repo_name="repo-a", - workflow_kind="light-task", - memory_mode="disabled", - coordination_root=root, - code_repo_path=root / "repo-a", - code_source_branch="main", - code_work_branch="ar/legacy", - code_base_commit="abc123", - worktree_name="legacy", - leaf_id="01_legacy-leaf", - ) - contract.contract_path.parent.mkdir(parents=True, exist_ok=True) - contract.contract_path.write_text(contract_to_text(contract), encoding="utf-8") + contract = _persisted_legacy_contract(root) loaded = load_contract(contract.contract_path) - self.assertEqual(loaded.leaf_id, "260700-L1") + self.assertEqual(loaded.leaf_id, "01_legacy-leaf") - def test_load_contract_keeps_leaf_id_when_task_resolution_is_unproven(self) -> None: + def test_heal_keeps_leaf_id_when_task_resolution_is_unproven(self) -> None: with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) task_root = root / "tasks" / "repo-a" / "duplicate" @@ -240,9 +256,171 @@ def test_load_contract_keeps_leaf_id_when_task_resolution_is_unproven(self) -> N active.mkdir(parents=True) (active / "series-contract.md").write_text("---\n---\n", encoding="utf-8") + report = heal_contract_leaf_ids(root) loaded = load_contract(contract.contract_path) self.assertEqual(loaded.leaf_id, "legacy-id") + self.assertEqual(report["healed"], []) + self.assertEqual(report["errors"], []) + + def test_load_contract_performs_zero_tree_traversal(self) -> None: + """260712-PTS-L1 R6a: the read path costs one file read + parse + validate. + + Every walk entry point the old normalization dragged in is patched to fail + loud — the leaf-ref resolver, the series-contract iterators, and pathlib's + directory-walking primitives themselves — so any regression that re-adds a + tasks-tree traversal to ``load_contract`` trips immediately. + """ + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_leaf(root) + contract = _persisted_legacy_contract(root) + + def _no_walk(name: str): + def _trip(*_args: object, **_kwargs: object) -> object: + raise AssertionError(f"load_contract must not call {name}") + + return _trip + + with ( + mock.patch( + "agents_remember.worktrees.worktree_contract.resolve_leaf_ref", + side_effect=_no_walk("resolve_leaf_ref"), + ), + mock.patch( + "agents_remember.worktrees.worktree_contract.resolve_active_task_root", + side_effect=_no_walk("resolve_active_task_root"), + ), + mock.patch( + "agents_remember.worktrees.task_resolver.iter_active_series_contracts", + side_effect=_no_walk("iter_active_series_contracts"), + ), + mock.patch( + "agents_remember.worktrees.task_resolver.iter_leaf_enclosure_contracts", + side_effect=_no_walk("iter_leaf_enclosure_contracts"), + ), + mock.patch.object(Path, "glob", _no_walk("Path.glob")), + mock.patch.object(Path, "rglob", _no_walk("Path.rglob")), + mock.patch.object(Path, "iterdir", _no_walk("Path.iterdir")), + ): + loaded = load_contract(contract.contract_path) + + self.assertEqual(loaded.leaf_id, "01_legacy-leaf") + self.assertEqual(loaded.task_id, "260700_MASTER") + + def test_heal_maps_legacy_stem_id_exactly_like_read_time_normalization(self) -> None: + """260712-PTS-L1 R6b: heal = the exact mapping the removed read-path walk produced.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_leaf(root) + # The sibling artifact JSON must stay ignored on the heal walk, exactly as + # it was on the removed read-time normalization walk. + task_root = root / "tasks" / "repo-a" / "260700_master" + (task_root / "lint-inventory-baseline.json").write_text( + '{"summary":{"findings":14},"items":[{"path":"a.py","line":1}]}', + encoding="utf-8", + ) + contract = _persisted_legacy_contract(root) + raw = load_contract(contract.contract_path) + expected = normalize_contract_leaf_id(raw, keep_unresolved=True).leaf_id + + report = heal_contract_leaf_ids(root) + + healed = load_contract(contract.contract_path) + self.assertEqual(healed.leaf_id, "260700-L1") + self.assertEqual(healed.leaf_id, expected) + self.assertEqual( + report["healed"], + [ + { + "contract": contract.contract_path.as_posix(), + "from": "01_legacy-leaf", + "to": "260700-L1", + } + ], + ) + self.assertEqual(report["errors"], []) + + def test_heal_is_idempotent_and_skips_canonical_contracts_without_resolution(self) -> None: + """260712-PTS-L1 R6c: a second sweep rewrites nothing and never resolves.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_leaf(root) + contract = _persisted_legacy_contract(root) + + first = heal_contract_leaf_ids(root) + healed_text = contract.contract_path.read_text(encoding="utf-8") + + with mock.patch( + "agents_remember.worktrees.worktree_contract.resolve_leaf_ref", + side_effect=AssertionError( + "a canonical contract must be skipped without a resolution walk" + ), + ): + second = heal_contract_leaf_ids(root) + + self.assertEqual(len(cast("list[object]", first["healed"])), 1) + self.assertEqual(second["healed"], []) + self.assertEqual(second["canonical"], 1) + self.assertEqual(second["errors"], []) + self.assertEqual( + contract.contract_path.read_text(encoding="utf-8"), + healed_text, + ) + + def test_heal_dry_run_reports_without_writing(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_leaf(root) + contract = _persisted_legacy_contract(root) + before = contract.contract_path.read_text(encoding="utf-8") + + report = heal_contract_leaf_ids(root, dry_run=True) + + self.assertTrue(report["dryRun"]) + self.assertEqual( + report["healed"], + [ + { + "contract": contract.contract_path.as_posix(), + "from": "01_legacy-leaf", + "to": "260700-L1", + } + ], + ) + self.assertEqual(contract.contract_path.read_text(encoding="utf-8"), before) + + def test_heal_reports_an_unreadable_contract_and_continues(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_leaf(root) + contract = _persisted_legacy_contract(root) + torn = root / "tasks" / "repo-a" / "260700_master" / "enclosures" / "00_torn" + torn.mkdir(parents=True) + (torn / "series-contract.md").write_text("not a contract", encoding="utf-8") + + report = heal_contract_leaf_ids(root) + + errors = cast("list[dict[str, str]]", report["errors"]) + self.assertEqual(len(errors), 1) + self.assertIn("00_torn", errors[0]["contract"]) + self.assertEqual(load_contract(contract.contract_path).leaf_id, "260700-L1") + + def test_heal_cli_command_is_the_on_demand_seam(self) -> None: + """260712-PTS-L1 R3: the deliberate invocation seam — explicit, loud, never per read.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + _write_leaf(root) + contract = _persisted_legacy_contract(root) + + stdout = io.StringIO() + with contextlib.redirect_stdout(stdout): + returncode = worktree_cli_main(["heal-leaf-ids", "--coordination-root", str(root)]) + + self.assertEqual(returncode, 0) + payload = json.loads(stdout.getvalue()) + self.assertEqual(len(payload["healed"]), 1) + self.assertEqual(load_contract(contract.contract_path).leaf_id, "260700-L1") def test_light_task_doc_is_indexed_as_candidate(self) -> None: with tempfile.TemporaryDirectory() as tmp: diff --git a/mcp/tests/test_projection_scaling_cs6.py b/mcp/tests/test_projection_scaling_cs6.py index 83bb6d17..14e8567f 100644 --- a/mcp/tests/test_projection_scaling_cs6.py +++ b/mcp/tests/test_projection_scaling_cs6.py @@ -7,11 +7,14 @@ from __future__ import annotations +import contextlib import json +import os import sys import tempfile import time import unittest +from collections.abc import Iterator from dataclasses import replace from datetime import UTC, datetime, timedelta from pathlib import Path @@ -23,7 +26,9 @@ from _scaling import assert_bounded_count from agents_remember.controlplane.records import GateRecord from agents_remember.controlplane.store import GateStore -from agents_remember.observer import projection_store, snapshots +from agents_remember.mcp.config import McpRuntimeConfig +from agents_remember.observer import contract_snapshot, drift_snapshots, projection_store, snapshots +from agents_remember.observer.contract_snapshot import ContractSnapshotCache from agents_remember.observer.events import Event from agents_remember.observer.paths import observer_logs_root from agents_remember.observer.projection import ( @@ -36,13 +41,20 @@ from agents_remember.observer.snapshots import ( TASK_DOCUMENT_SCHEMA, TASK_DOCUMENT_SUMMARY_LIMIT, + read_enclosures, + read_engine_process_facts, read_gates, read_series_documents, read_task_document_body, read_task_documents, ) from agents_remember.observer.store import EventStore -from agents_remember.worktrees.worktree_contract import default_contract, write_contract +from agents_remember.worktrees.task_resolver import ENCLOSURES_DIR, SERIES_CONTRACT_FILENAME +from agents_remember.worktrees.worktree_contract import ( + WorktreeContract, + default_contract, + write_contract, +) NOW = datetime(2026, 7, 9, 12, 0, 0, tzinfo=UTC) @@ -480,5 +492,273 @@ def test_over_budget_warning_is_rate_limited_across_ticks(self) -> None: ) +class ContractSnapshotSharedPassTests(unittest.TestCase): + """260712-PTS-L2: one shared contract pass per tick with a stat-identity parse cache. + + The projection previously enumerated + re-parsed EVERY leaf enclosure contract three + times per tick (read_enclosures, read_engine_process_facts, drift-snapshot pruning). + These prove the shared ``ContractSnapshot`` performs one pass per tick (R1), that the + cross-tick cache parses only changed files (R2/R7), that retention is bounded to the + live contract set (R3), and that reader outputs are behavior-identical (R4). + """ + + def _seed_contracts(self, coordination_root: Path, count: int) -> list[WorktreeContract]: + contracts: list[WorktreeContract] = [] + for index in range(count): + contract = default_contract( + task_name=f"scaling task {index}", + repo_name="repo", + workflow_kind="light-task", + memory_mode="disabled", + coordination_root=coordination_root, + code_repo_path=coordination_root / "repo", + code_source_branch="main", + code_work_branch=f"ar/leaf-{index}", + code_base_commit="0" * 40, + worktree_name=f"leaf-{index}", + lifecycle_id=f"LC-{index}", + ) + contract.contract_path.parent.mkdir(parents=True, exist_ok=True) + write_contract(contract.contract_path, contract) + contracts.append(contract) + return contracts + + def _seed_malformed_contract(self, coordination_root: Path) -> Path: + path = ( + coordination_root + / "tasks" + / "repo" + / "broken-task" + / ENCLOSURES_DIR + / "broken-leaf" + / SERIES_CONTRACT_FILENAME + ) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("this is not a worktree contract\n", encoding="utf-8") + return path + + @contextlib.contextmanager + def _counting_parses(self) -> Iterator[list[Path]]: + """Count load_contract calls made by the snapshot builder (its only parse site).""" + calls: list[Path] = [] + original = contract_snapshot.load_contract + + def counting(path: Path) -> WorktreeContract: + calls.append(path) + return original(path) + + with patch.object(contract_snapshot, "load_contract", counting): + yield calls + + @staticmethod + def _bump_mtime(path: Path) -> None: + """Force a stat-identity change even on coarse-timestamp filesystems.""" + stat = path.stat() + os.utime(path, ns=(stat.st_atime_ns, stat.st_mtime_ns + 1_000_000)) + + @staticmethod + def _step_past_ctime_granule() -> None: + """Sleep past the kernel's coarse-clock tick so the next metadata change gets a + distinct ctime_ns (ctime cannot be set explicitly the way _bump_mtime sets mtime; + at HZ=100 the inode-timestamp granule is 10ms, so 50ms clears it comfortably).""" + time.sleep(0.05) + + def test_parse_counts_n_then_zero_then_exactly_the_changed_file(self) -> None: + """R7/R2: N parses on tick 1, zero on an unchanged tick 2, only the change on tick 3.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + contracts = self._seed_contracts(root, 5) + tasks_root = root / "tasks" + cache = ContractSnapshotCache() + + with self._counting_parses() as calls: + first = cache.build(tasks_root) + self.assertEqual(len(first.contracts), 5) + self.assertEqual(len(calls), 5) # tick 1: every contract parsed once, not 3x + + with self._counting_parses() as calls: + second = cache.build(tasks_root) + assert_bounded_count(len(calls), 0, label="unchanged-tick contract parses") + self.assertEqual(dict(first.contracts), dict(second.contracts)) + + changed = contracts[2] + write_contract(changed.contract_path, replace(changed, lifecycle_id="LC-CHANGED")) + self._bump_mtime(changed.contract_path) + with self._counting_parses() as calls: + third = cache.build(tasks_root) + self.assertEqual(calls, [changed.contract_path]) # tick 3: exactly the changed file + self.assertEqual(third.contracts[changed.contract_path].lifecycle_id, "LC-CHANGED") + + def test_full_projection_tick_enumerates_once_and_reparses_nothing_unchanged(self) -> None: + """R1: project_and_write drives ONE enumeration per tick for all three consumers.""" + with tempfile.TemporaryDirectory() as tmp: + tmp_path = Path(tmp).resolve() + coord = tmp_path / "coord" + coord.mkdir(parents=True) + self._seed_contracts(coord, 4) + config = McpRuntimeConfig( + config_path=coord / "mcp.settings.json", + coordination_root=coord, + workspace_root=tmp_path / "ws", + transcript_root=coord / "logs", + ) + snapshots._task_doc_cache.clear() + projection_store._repo_surface_cache.clear() + self.addCleanup(snapshots._task_doc_cache.clear) + self.addCleanup(projection_store._repo_surface_cache.clear) + walks = {"count": 0} + original_iter = contract_snapshot.iter_leaf_enclosure_contracts + + def counting_iter(tasks_root: Path) -> Iterator[Path]: + walks["count"] += 1 + return original_iter(tasks_root) + + with ( + patch.object(projection_store, "_contract_snapshot_cache", ContractSnapshotCache()), + patch.object(contract_snapshot, "iter_leaf_enclosure_contracts", counting_iter), + self._counting_parses() as calls, + ): + projection_store.project_and_write(config, now=NOW) + first_walks, first_parses = walks["count"], len(calls) + projection_store.project_and_write(config, now=NOW + timedelta(seconds=1)) + second_walks = walks["count"] - first_walks + second_parses = len(calls) - first_parses + + assert_bounded_count(first_walks, 1, label="tick-1 contract enumerations") + self.assertEqual(first_parses, 4) # one parse per contract on the cold tick + assert_bounded_count(second_walks, 1, label="tick-2 contract enumerations") + assert_bounded_count(second_parses, 0, label="tick-2 contract re-parses") + + def test_reader_outputs_equal_with_and_without_shared_snapshot(self) -> None: + """R4: enclosure rows, engine facts, and prune keys are identical in both modes.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + contracts = self._seed_contracts(root, 3) + contracts[0].code_worktree.mkdir(parents=True) # one live worktree for prune keys + broken = self._seed_malformed_contract(root) + tasks_root = root / "tasks" + cache = ContractSnapshotCache() + cold = cache.build(tasks_root) + warm = cache.build(tasks_root) # cache-hit build must be output-identical too + + self.assertEqual(len(cold.contracts), 3) + self.assertIn(broken, cold.skipped) # malformed: skipped, never fatal (as before) + + standalone_enclosures = read_enclosures(root) + self.assertEqual(standalone_enclosures, read_enclosures(root, contracts=cold)) + self.assertEqual(standalone_enclosures, read_enclosures(root, contracts=warm)) + + standalone_facts = read_engine_process_facts(root) + self.assertEqual(standalone_facts, read_engine_process_facts(root, contracts=cold)) + self.assertEqual(standalone_facts, read_engine_process_facts(root, contracts=warm)) + + standalone_keys = drift_snapshots._active_worktree_snapshot_keys(root) + self.assertEqual( + standalone_keys, + drift_snapshots._active_worktree_snapshot_keys(root, contracts=cold), + ) + self.assertEqual( + standalone_keys, + {(contracts[0].code_worktree.name, contracts[0].code_work_branch)}, + ) + + def test_cache_drops_entries_for_contracts_that_left_the_enumeration(self) -> None: + """R3: retention is bounded to the live path set across task churn.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + contracts = self._seed_contracts(root, 3) + tasks_root = root / "tasks" + cache = ContractSnapshotCache() + cache.build(tasks_root) + self.assertEqual(len(cache._entries), 3) + + contracts[1].contract_path.unlink() + snapshot = cache.build(tasks_root) + self.assertEqual(len(snapshot.contracts), 2) + self.assertNotIn(contracts[1].contract_path, cache._entries) + self.assertEqual(len(cache._entries), 2) + + def test_permission_flip_invalidates_cache_instead_of_serving_stale_parse(self) -> None: + """Review hardening: chmod changes ctime only -- the cache must skip, not serve forever. + + Pre-L2, an unreadable contract was skipped every tick (PermissionError). A cache + keyed only on (mtime_ns, size) would keep serving the old good parse FOREVER after + a chmod 000, because chmod changes neither; ctime_ns in the identity restores the + pre-L2 degradation (and, unlike the same-stat rewrite window, this case never + self-heals without it). + """ + if os.geteuid() == 0: + self.skipTest("root ignores file modes; the permission flip cannot be observed") + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + [contract] = self._seed_contracts(root, 1) + tasks_root = root / "tasks" + cache = ContractSnapshotCache() + first = cache.build(tasks_root) + self.assertIn(contract.contract_path, first.contracts) + + # ctime is kernel-assigned from the coarse clock; step past the granule so the + # chmod lands on a distinct ctime_ns (production chmods are never in the same + # tick as the original contract write). Tempdir cleanup can unlink a 000-mode + # file (unlink needs directory perms only), so no chmod-back cleanup is needed. + self._step_past_ctime_granule() + os.chmod(contract.contract_path, 0o000) + second = cache.build(tasks_root) + self.assertNotIn(contract.contract_path, second.contracts) + self.assertIn(contract.contract_path, second.skipped) + + # Restoring readability self-heals on the next build (chmod bumps ctime again, + # and the failed build already dropped the stale entry). + os.chmod(contract.contract_path, 0o644) + third = cache.build(tasks_root) + self.assertIn(contract.contract_path, third.contracts) + + def test_rewrite_with_pinned_mtime_and_size_is_still_detected(self) -> None: + """Review hardening: an os.utime-pinned same-size rewrite still bumps ctime -> re-parse.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + [contract] = self._seed_contracts(root, 1) + path = contract.contract_path + tasks_root = root / "tasks" + cache = ContractSnapshotCache() + first = cache.build(tasks_root) + self.assertEqual(first.contracts[path].lifecycle_id, "LC-0") + before = path.stat() + + # Same-length content rewrite, then pin (mtime_ns, size) back to the cached + # identity. Step past the coarse-clock granule first so the rewrite's + # kernel-assigned ctime_ns is distinct from the cached one. + self._step_past_ctime_granule() + path.write_text( + path.read_text(encoding="utf-8").replace("LC-0", "LC-9"), encoding="utf-8" + ) + os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns)) + after = path.stat() + self.assertEqual( + (before.st_mtime_ns, before.st_size), (after.st_mtime_ns, after.st_size) + ) + + second = cache.build(tasks_root) + self.assertEqual(second.contracts[path].lifecycle_id, "LC-9") + + def test_malformed_contract_is_retried_every_build_not_cached(self) -> None: + """R2 containment: a parse failure degrades exactly as before -- skip + retry next tick.""" + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp).resolve() + self._seed_contracts(root, 1) + broken = self._seed_malformed_contract(root) + tasks_root = root / "tasks" + cache = ContractSnapshotCache() + with self._counting_parses() as calls: + first = cache.build(tasks_root) + second = cache.build(tasks_root) + self.assertIn(broken, first.skipped) + self.assertIn(broken, second.skipped) + # The broken file is re-attempted each build (a transient error self-heals); + # the healthy contract is parsed exactly once across both builds. + self.assertEqual(calls.count(broken), 2) + self.assertEqual(len(calls), 3) + + if __name__ == "__main__": unittest.main() diff --git a/mcp/tests/test_serving.py b/mcp/tests/test_serving.py index cabbffdd..3cd9c261 100644 --- a/mcp/tests/test_serving.py +++ b/mcp/tests/test_serving.py @@ -475,7 +475,10 @@ def _client_with_held_projection( ) patcher.start() self.addCleanup(patcher.stop) - return TestClient(create_app(_config(self.tmp), interval=interval)) + # watch_changes=False: this world changes only through the mocked project_and_write, + # which no filesystem watcher can observe -- the tick loop must stay interval-paced + # (the live contract for watcher-invisible changes is the heartbeat bound instead). + return TestClient(create_app(_config(self.tmp), interval=interval, watch_changes=False)) def _get_until(self, client: TestClient, *, etag: str, want_status: int) -> httpx.Response: """Poll /api/state with If-None-Match until the tick loop publishes ``want_status``.""" @@ -989,6 +992,7 @@ def test_dashboard_subcommand_parsing(self) -> None: self.assertEqual(namespace.port, 9999) self.assertEqual(namespace.host, "127.0.0.1") self.assertEqual(namespace.interval, 1.0) + self.assertIsNone(namespace.heartbeat) self.assertIs(namespace.func, cli_dashboard.run) @@ -999,6 +1003,7 @@ def _args(self, **overrides: object) -> argparse.Namespace: "host": "127.0.0.1", "port": 8765, "interval": 1.0, + "heartbeat": None, "reload": False, "sim": None, "sim_speed": "1", @@ -1607,6 +1612,7 @@ def _args(self, **overrides: object) -> argparse.Namespace: "host": "127.0.0.1", "port": 8765, "interval": 1.0, + "heartbeat": None, "reload": False, "sim": None, "sim_speed": "1",