284dbaf537
Activation reaches plugin discovery before the application dependencies exist. Give PM its own locked Python project and runtime so it can install or repair the application without importing that dependency tree. Keep PM outside the application workspace. A shared uv workspace resolves the application graph and cannot provide this isolation. Route mutations through an isolated worker and preserve transaction callbacks, cancellation, custom package registrations, and correlated receipts. Use the same runtime builder for source installs and packaged payloads. Keep offline wheelhouse support in that builder. Nix builds the independent PM lock as a separate derivation. Refuse lazy-disabled bootstrap before installing tools or dependencies. Move first-party YAML readers and writers to ruamel. Keep the application lock's transitive PyYAML requirements for third-party packages. Verification: - Focused canonical Python suite: 177 passed, 1 host-gated skip. - Electron backend probes: 12 passed. Electron typecheck passed. - Both uv locks, scoped lint, Bash syntax, and whitespace checks passed. - Cold activation, corrupt-app repair, offline staging, and relocation ran. - Built and exercised the Nix PM runtime and standalone YAML merge script. Six broader caller test files retain the same 24 failing test IDs as an archive of HEAD. The existing real-home guard blocks those tests before they can exercise the affected paths. No full-suite pass is claimed. Native Windows signing and full Bionic package execution remain unverified.
126 lines
6.2 KiB
Python
126 lines
6.2 KiB
Python
"""Isolated production-library controls; no transport/provider calls."""
|
|
import contextlib
|
|
import importlib.util
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
from collections import OrderedDict
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
repo = Path(sys.argv[1]).resolve()
|
|
home = Path(tempfile.mkdtemp(prefix="idle-controls-"))
|
|
for key in list(os.environ):
|
|
if key.startswith("HERMES_") or key.endswith(("_API_KEY", "_TOKEN")):
|
|
os.environ.pop(key, None)
|
|
os.environ.update(HOME=str(home), HERMES_HOME=str(home / ".hermes"), SESSION_IDLE_MINUTES="1", SESSION_RESET_HOUR="0")
|
|
sys.path.insert(0, str(repo))
|
|
import hermes_yaml as yaml
|
|
from gateway.config import load_gateway_config, GatewayConfig, Platform
|
|
from gateway.session import SessionStore, SessionSource
|
|
from gateway.run import GatewayRunner
|
|
from gateway.agent_cache_pressure import AgentCacheBounds
|
|
|
|
Path(os.environ["HERMES_HOME"]).mkdir(exist_ok=True)
|
|
Path(os.environ["HERMES_HOME"], "config.yaml").write_text(yaml.safe_dump({"session_reset": {"mode": "both", "idle_minutes": 1}, "gateway": {"session_reset": {"mode": "daily", "at_hour": 0}}}), encoding="utf-8")
|
|
config = load_gateway_config()
|
|
assert "default_reset_policy" not in config.to_dict()
|
|
source = SessionSource(platform=Platform.TELEGRAM, chat_id="cache", user_id="control")
|
|
store = SessionStore(home / "sessions", config)
|
|
entry = store.get_or_create_session(source)
|
|
messages = [{"role": "user", "content": "persist across soft release"}]
|
|
store.append_to_transcript(entry.session_id, messages[0])
|
|
results = {"legacy_yaml_env_ignored": True, "cache": []}
|
|
|
|
for strategy in ("ttl", "lru", "pressure"):
|
|
runner = object.__new__(GatewayRunner)
|
|
runner._agent_cache_lock = threading.Lock()
|
|
runner._agent_cache_bounds_cache = AgentCacheBounds(max_size=1, idle_ttl_secs=1, memory_high_mb=1, protect_recent=0)
|
|
runner.session_store = store
|
|
trace = []
|
|
released = threading.Event()
|
|
def commit(msgs):
|
|
assert msgs == messages
|
|
Path(home, strategy + "-memory.json").write_text(json.dumps(msgs), encoding="utf-8")
|
|
trace.append("memory")
|
|
def release():
|
|
assert json.loads(Path(home, strategy + "-memory.json").read_text(encoding="utf-8")) == messages
|
|
trace.append("release")
|
|
released.set()
|
|
agent = SimpleNamespace(_memory_manager=object(), _session_messages=list(messages),
|
|
_last_activity_ts=time.time() - 86400, _last_flushed_db_idx=len(messages),
|
|
commit_memory_session=commit, release_clients=release)
|
|
active = SimpleNamespace(_last_activity_ts=time.time() - 86400)
|
|
runner._running_agents = {"active": active}
|
|
runner._agent_cache = OrderedDict([(entry.session_key, (agent, "sig")), ("active", (active, "sig"))])
|
|
if strategy == "ttl":
|
|
assert runner._sweep_idle_cached_agents() == 1
|
|
elif strategy == "pressure":
|
|
unflushed = SimpleNamespace(_session_messages=[{"role": "user", "content": "not on disk"}], _last_flushed_db_idx=0)
|
|
runner._agent_cache["unflushed"] = (unflushed, "sig")
|
|
assert runner._sweep_agent_cache_under_pressure() == 1
|
|
assert "unflushed" in runner._agent_cache
|
|
else:
|
|
with runner._agent_cache_lock:
|
|
runner._enforce_agent_cache_cap()
|
|
assert released.wait(10), strategy
|
|
assert trace == ["memory", "release"]
|
|
assert "active" in runner._agent_cache
|
|
resumed = store.get_or_create_session(source)
|
|
assert resumed.session_id == entry.session_id
|
|
assert store.load_transcript(resumed.session_id)[0]["content"] == messages[0]["content"]
|
|
assert store._db.get_session(resumed.session_id)["end_reason"] is None
|
|
results["cache"].append({"strategy": strategy, "order": trace, "same_transcript": True, "active_preserved": True})
|
|
|
|
store._db.end_session(entry.session_id, "ws_orphan_reap")
|
|
recovered = store.get_or_create_session(source)
|
|
assert recovered.session_id == entry.session_id
|
|
assert store._db.get_session(entry.session_id)["end_reason"] is None
|
|
results["ws_orphan_end_reopens_same_transcript"] = True
|
|
|
|
store.suspend_session(entry.session_key)
|
|
suspended = store.get_or_create_session(source)
|
|
assert suspended.session_id != entry.session_id
|
|
assert store._db.get_session(entry.session_id)["end_reason"] == "suspended"
|
|
results["explicit_suspension_rotates"] = True
|
|
store.append_to_transcript(suspended.session_id, messages[0])
|
|
# Historical finalization wrote both the flag and durable reset boundary.
|
|
store._db.set_expiry_finalized(suspended.session_id)
|
|
store._db.promote_to_session_reset(suspended.session_id)
|
|
store._db.replace_gateway_routing_entries({}, scope=store._routing_scope())
|
|
store._entries.clear()
|
|
fenced = store.get_or_create_session(source)
|
|
assert fenced.session_id != suspended.session_id
|
|
results["historical_expiry_fence_preserved"] = True
|
|
store._db.close()
|
|
|
|
spec = importlib.util.spec_from_file_location("migration_probe", repo / "optional-skills/migration/openclaw-migration/scripts/openclaw_to_hermes.py")
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
source_dir = home / "migration-source"
|
|
target = home / "migration-target"
|
|
source_dir.mkdir()
|
|
target.mkdir()
|
|
(target / "config.yaml").write_text("memory: {}\n", encoding="utf-8")
|
|
before = (target / "config.yaml").read_bytes()
|
|
migrator = module.Migrator(source_dir, target, True, None, False, False, home / "archive-output")
|
|
migrator.migrate_session_config({"session": {"reset": {"mode": "daily", "atHour": 3}, "identityLinks": {"test": "identity"}}})
|
|
assert (target / "config.yaml").read_bytes() == before
|
|
assert json.loads((migrator.archive_dir / "session-config.json").read_text(encoding="utf-8")) == {"identityLinks": {"test": "identity"}}
|
|
results["migration_timers_ignored_advanced_archived"] = True
|
|
from contextlib import redirect_stdout
|
|
from io import StringIO
|
|
from hermes_cli.cli_info_mixin import CLIInfoMixin
|
|
status_output = StringIO()
|
|
with redirect_stdout(status_output):
|
|
CLIInfoMixin._show_gateway_status(object.__new__(CLIInfoMixin))
|
|
assert "Error loading gateway config" not in status_output.getvalue()
|
|
assert "Conversations persist" in status_output.getvalue()
|
|
results["gateway_status_without_policy"] = True
|
|
print(json.dumps(results, indent=2), flush=True)
|